chore: tests

fix: function name

Author: sangeethailango

apps/api/plane/app/views/page/base.py

@@ -40,7 +40,7 @@
 from plane.bgtasks.page_transaction_task import page_transaction
 from plane.bgtasks.page_version_task import page_version
 from plane.bgtasks.recent_visited_task import recent_visited_task
-from plane.bgtasks.copy_s3_object import copy_s3_objects
+from plane.bgtasks.copy_s3_object import copy_s3_objects_of_description_and_assets
 
 
 def unarchive_archive_page_and_descendants(page_id, archived_at):
@@ -606,7 +606,7 @@ def post(self, request, slug, project_id, page_id):
         )
 
         # Copy the s3 objects uploaded in the page
-        copy_s3_objects.delay(
+        copy_s3_objects_of_description_and_assets.delay(
             entity_name="PAGE",
             entity_identifier=page.id,
             project_id=project_id,

apps/api/plane/bgtasks/copy_s3_object.py

@@ -126,7 +126,9 @@ def copy_assets(entity, entity_identifier, project_id, asset_ids, user_id):
 
 
 @shared_task
-def copy_s3_objects(entity_name, entity_identifier, project_id, slug, user_id):
+def copy_s3_objects_of_description_and_assets(
+    entity_name, entity_identifier, project_id, slug, user_id
+):
     """
     Step 1: Extract asset ids from the description_html of the entity
     Step 2: Duplicate the assets

apps/api/plane/tests/unit/bg_tasks/test_copy_s3_objects.py

@@ -1,40 +1,38 @@
 import pytest
-from plane.db.models import User, Workspace, Project, ProjectMember, Issue, FileAsset
+from plane.db.models import Project, ProjectMember, Issue, FileAsset
 from unittest.mock import patch, MagicMock
-from plane.bgtasks.copy_s3_object import copy_s3_objects
+from plane.bgtasks.copy_s3_object import (
+    copy_s3_objects_of_description_and_assets,
+    copy_assets,
+)
 import base64
 
 
 @pytest.mark.unit
 class TestCopyS3Objects:
-    """Test the copy_s3_objects function"""
-
-    @pytest.mark.django_db
-    @patch("plane.bgtasks.copy_s3_object.S3Storage")
-    def test_copy_s3_objects(self, mock_s3_storage):
-        # Create test data
-        test_user = User.objects.create(
-            email="[email protected]", first_name="Test", last_name="User"
-        )
-
-        workspace = Workspace.objects.create(
-            name="Test Workspace", slug="test-workspace", owner=test_user
-        )
+    """Test the copy_s3_objects_of_description_and_assets function"""
 
+    @pytest.fixture
+    def project(self, create_user, workspace):
         project = Project.objects.create(
             name="Test Project", identifier="test-project", workspace=workspace
         )
-        ProjectMember.objects.create(project=project, member=test_user)
 
-        issue = Issue.objects.create(
+        ProjectMember.objects.create(project=project, member=create_user)
+        return project
+
+    @pytest.fixture
+    def issue(self, workspace, project):
+        return Issue.objects.create(
             name="Test Issue",
             workspace=workspace,
-            project=project,
+            project_id=project.id,
             description_html=f'<div><image-component src="35e8b958-6ee5-43ce-ae56-fb0e776f421e"></image-component><image-component src="97988198-274f-4dfe-aa7a-4c0ffc684214"></image-component></div>',
         )
 
-        # Create test file assets
-        asset1 = FileAsset.objects.create(
+    @pytest.fixture
+    def file_asset(self, workspace, project, issue):
+        return FileAsset.objects.create(
             issue=issue,
             workspace=workspace,
             project=project,
@@ -45,10 +43,15 @@ def test_copy_s3_objects(self, mock_s3_storage):
                 "type": "image/jpeg",
             },
             id="35e8b958-6ee5-43ce-ae56-fb0e776f421e",
-            entity_type="ISSUE_DESCRIPTION",  # Set the correct entity type
+            entity_type="ISSUE_DESCRIPTION",
         )
 
-        asset2 = FileAsset.objects.create(
+    @pytest.mark.django_db
+    @patch("plane.bgtasks.copy_s3_object.S3Storage")
+    def test_copy_s3_objects_of_description_and_assets(
+        self, mock_s3_storage, create_user, workspace, project, issue, file_asset
+    ):
+        FileAsset.objects.create(
             issue=issue,
             workspace=workspace,
             project=project,
@@ -59,9 +62,11 @@ def test_copy_s3_objects(self, mock_s3_storage):
                 "type": "application/pdf",
             },
             id="97988198-274f-4dfe-aa7a-4c0ffc684214",
-            entity_type="ISSUE_DESCRIPTION",  # Set the correct entity type
+            entity_type="ISSUE_DESCRIPTION",
         )
 
+        issue.save()
+
         # Set up mock S3 storage
         mock_storage_instance = MagicMock()
         mock_s3_storage.return_value = mock_storage_instance
@@ -76,39 +81,102 @@ def test_copy_s3_objects(self, mock_s3_storage):
             }
 
             # Call the actual function (not .delay())
-            copy_s3_objects(
-                "ISSUE", issue.id, project.id, "test-workspace", test_user.id
+            copy_s3_objects_of_description_and_assets(
+                "ISSUE", issue.id, project.id, "test-workspace", create_user.id
             )
 
         # Assert that copy_object was called for each asset
         assert mock_storage_instance.copy_object.call_count == 2
 
-        # Get the copy operations that were performed
-        copy_calls = mock_storage_instance.copy_object.call_args_list
-
-        # Sort assets and calls by asset name to ensure they match
-        assets = sorted([asset1, asset2], key=lambda x: str(x.asset))
-        sorted_calls = sorted(
-            copy_calls, key=lambda x: str(x[0][0])
-        )  # Sort by source path as string
-
-        # Verify both assets were copied
-        for asset, call in zip(assets, sorted_calls):
-            args = call[0]
-            # Verify source path matches original asset
-            assert str(asset.asset) == str(
-                args[0]
-            )  # Convert both to strings for comparison
-            # Verify destination path
-            assert str(workspace.id) in str(args[1])
-            assert asset.attributes["name"] in str(args[1])
-
         # Get the updated issue and its new assets
         updated_issue = Issue.objects.get(id=issue.id)
         new_assets = FileAsset.objects.filter(
             issue=updated_issue,
-            entity_type="ISSUE_DESCRIPTION",  # Filter by the correct entity type
+            entity_type="ISSUE_DESCRIPTION",
         )
 
         # Verify new assets were created
-        assert new_assets.count() == 4
+        assert new_assets.count() == 4  # 2 original + 2 copied
+
+    @pytest.mark.django_db
+    @patch("plane.bgtasks.copy_s3_object.S3Storage")
+    def test_copy_assets_successful(
+        self, mock_s3_storage, workspace, project, issue, file_asset
+    ):
+        """Test successful copying of assets"""
+        # Arrange
+        mock_storage_instance = MagicMock()
+        mock_s3_storage.return_value = mock_storage_instance
+
+        # Act
+        result = copy_assets(
+            entity=issue,
+            entity_identifier=issue.id,
+            project_id=project.id,
+            asset_ids=[file_asset.id],
+            user_id=issue.created_by_id,
+        )
+
+        # Assert
+        # Verify S3 copy was called
+        mock_storage_instance.copy_object.assert_called_once()
+
+        # Verify new asset was created
+        assert len(result) == 1
+        new_asset_id = result[0]["new_asset_id"]
+        new_asset = FileAsset.objects.get(id=new_asset_id)
+
+        # Verify asset properties were copied correctly
+        assert new_asset.workspace == workspace
+        assert new_asset.project_id == project.id
+        assert new_asset.entity_type == file_asset.entity_type
+        assert new_asset.attributes == file_asset.attributes
+        assert new_asset.size == file_asset.size
+        assert new_asset.is_uploaded is True
+
+    @pytest.mark.django_db
+    @patch("plane.bgtasks.copy_s3_object.S3Storage")
+    def test_copy_assets_empty_asset_ids(
+        self, mock_s3_storage, workspace, project, issue
+    ):
+        """Test copying with empty asset_ids list"""
+        # Arrange
+        mock_storage_instance = MagicMock()
+        mock_s3_storage.return_value = mock_storage_instance
+
+        # Act
+        result = copy_assets(
+            entity=issue,
+            entity_identifier=issue.id,
+            project_id=project.id,
+            asset_ids=[],
+            user_id=issue.created_by_id,
+        )
+
+        # Assert
+        assert result == []
+        mock_storage_instance.copy_object.assert_not_called()
+
+    @pytest.mark.django_db
+    @patch("plane.bgtasks.copy_s3_object.S3Storage")
+    def test_copy_assets_nonexistent_asset(
+        self, mock_s3_storage, workspace, project, issue
+    ):
+        """Test copying with non-existent asset ID"""
+        # Arrange
+        mock_storage_instance = MagicMock()
+        mock_s3_storage.return_value = mock_storage_instance
+        non_existent_id = "00000000-0000-0000-0000-000000000000"
+
+        # Act
+        result = copy_assets(
+            entity=issue,
+            entity_identifier=issue.id,
+            project_id=project.id,
+            asset_ids=[non_existent_id],
+            user_id=issue.created_by_id,
+        )
+
+        # Assert
+        assert result == []
+        mock_storage_instance.copy_object.assert_not_called()