Test - rpm copy package

Test case for sctrict copy of RPM packages with its dependencies.

closes: #6789
https://pulp.plan.io/issues/6789

CHANGES/6789.misc

@@ -0,0 +1 @@
+Added two test cases for strict copy of packages with depsolving.

coverage.md

@@ -30,7 +30,7 @@ Manual Coverage
 | **Copy** |  |  |
 | As a user, I can copy any content by adding it to a repository with modify/ endpoint (but nothing is copied automatically, and invalid repositories will fail to validate for some definition of “invalid”) | NO |  |
 | As a user, I can copy any content by href using Copy API | PART |  |
-| As a user, I can copy RPM package and its dependencies (if depsolving=True) | NO |  |
+| As a user, I can copy RPM package and its dependencies (if depsolving=True) | Yes | to empty and non-empty repository |
 | As a user, I can copy Advisory and packages it refers to (and their dependencies if depsolving=True) by copying the Advisory | PART |  |
 | As a user, I can copy Modulemd and its artifacts by copying the Modulemd | NO |  |
 | As a user, I can copy Modulemd with its artifacts and its module dependencies and artifacts’ dependencies (if depsolving=True). | NO |  |

pulp_rpm/tests/functional/api/test_copy.py

@@ -16,6 +16,7 @@
 
 from pulp_rpm.tests.functional.constants import (
     KICKSTART_CONTENT_PATH,
+    PULP_TYPE_PACKAGE,
     RPM_KICKSTART_CONTENT_NAME,
     RPM_FIXTURE_SUMMARY,
     RPM_KICKSTART_FIXTURE_URL,
@@ -26,9 +27,16 @@
     UPDATERECORD_CONTENT_PATH,
     RPM_CONTENT_PATH,
 )
-from pulp_rpm.tests.functional.utils import gen_rpm_remote, rpm_copy
+from pulp_rpm.tests.functional.utils import gen_rpm_client, gen_rpm_remote, monitor_task, rpm_copy
 from pulp_rpm.tests.functional.utils import set_up_module as setUpModule  # noqa:F401
 
+from pulpcore.client.pulp_rpm import (
+    ContentPackagesApi,
+    RepositoriesRpmApi,
+    RpmRepositorySyncURL,
+    RemotesRpmApi,
+)
+
 
 class BaseCopy(unittest.TestCase):
     """Base-class for shared code for copy-test-subclasses."""
@@ -265,3 +273,137 @@ def test_all_content_recursive(self):
         dc = self.client.get(f"{RPM_CONTENT_PATH}?repository_version={latest_href}")
         dest_content = [c["pulp_href"] for c in dc["results"]]
         self.assertEqual(sorted(rpms_to_copy), sorted(dest_content))
+
+class StrictPackageCopyTestCase(unittest.TestCase):
+    """Test strict copy of package and its dependencies."""
+
+    @classmethod
+    def setUpClass(cls):
+        """Create class-wide variables."""
+        cls.cfg = config.get_config()
+        cls.client = gen_rpm_client()
+        cls.repo_api = RepositoriesRpmApi(cls.client)
+        cls.remote_api = RemotesRpmApi(cls.client)
+        cls.rpm_content_api = ContentPackagesApi(cls.client)
+        delete_orphans(cls.cfg)
+
+    def test_sctrict_copy_package_to_empty_repo(self):
+        """Test copy package and its dependecies to empty repository.
+
+        - Create repository and populate it
+        - Create empty repository
+        - Use 'copy' to copy 'whale' package with dependencies
+        - assert package and its dependencies were copied
+        """
+        empty_repo = self.repo_api.create(gen_repo())
+        self.addCleanup(self.repo_api.delete, empty_repo.pulp_href)
+
+        repo = self.repo_api.create(gen_repo())
+        self.addCleanup(self.repo_api.delete, repo.pulp_href)
+
+        body = gen_rpm_remote(url=RPM_UNSIGNED_FIXTURE_URL)
+        remote = self.remote_api.create(body)
+        self.addCleanup(self.remote_api.delete, remote.pulp_href)
+
+        repository_sync_data = RpmRepositorySyncURL(remote=remote.pulp_href)
+        sync_response = self.repo_api.sync(repo.pulp_href, repository_sync_data)
+        monitor_task(sync_response.task)
+
+        repo = self.repo_api.read(repo.pulp_href)
+        whale_package_href = [
+            pkg
+            for pkg in get_content(repo.to_dict())['rpm.package']
+            if pkg['name'] == 'whale'
+        ][0]['pulp_href']
+        package_to_copy = []
+        package_to_copy.append(whale_package_href)
+
+        config = [{
+            'source_repo_version': repo.latest_version_href,
+            'dest_repo': empty_repo.pulp_href,
+            'content': package_to_copy
+        }]
+
+        rpm_copy(self.cfg, config, recursive=True)
+        empty_repo = self.repo_api.read(empty_repo.pulp_href)
+        empty_repo_packages = [
+            pkg['name']
+            for pkg in get_content(empty_repo.to_dict())['rpm.package']
+        ]
+
+        # assert that only 3 packages are copied (original package with its two dependencies)
+        self.assertEqual(len(empty_repo_packages), 3)
+        # assert dependecies package names
+        self.assertIn('stork', empty_repo_packages)
+        self.assertIn('shark', empty_repo_packages)
+
+    def test_sctrict_copy_package_to_existing_repo(self):
+        """Test copy package and its dependecies to empty repository.
+
+        - Create repository and populate it
+        - Create second repository with package fullfiling test package dependency
+        - Use 'copy' to copy 'whale' package with dependencies
+        - assert package and its missing dependencies were copied
+        """
+        # prepare final_repo - copy to repository
+        final_repo = self.repo_api.create(gen_repo())
+        self.addCleanup(self.repo_api.delete, final_repo.pulp_href)
+
+        body = gen_rpm_remote(url=RPM_UNSIGNED_FIXTURE_URL)
+        remote = self.remote_api.create(body)
+        self.addCleanup(self.remote_api.delete, remote.pulp_href)
+
+        repository_sync_data = RpmRepositorySyncURL(remote=remote.pulp_href)
+        sync_response = self.repo_api.sync(final_repo.pulp_href, repository_sync_data)
+        monitor_task(sync_response.task)
+
+        final_repo = self.repo_api.read(final_repo.pulp_href)
+
+        # prepare repository - copy from repository
+        repo = self.repo_api.create(gen_repo())
+        self.addCleanup(self.repo_api.delete, repo.pulp_href)
+
+        body = gen_rpm_remote(url=RPM_UNSIGNED_FIXTURE_URL)
+        remote = self.remote_api.create(body)
+        self.addCleanup(self.remote_api.delete, remote.pulp_href)
+
+        repository_sync_data = RpmRepositorySyncURL(remote=remote.pulp_href)
+        sync_response = self.repo_api.sync(repo.pulp_href, repository_sync_data)
+        monitor_task(sync_response.task)
+
+        repo = self.repo_api.read(repo.pulp_href)
+
+        # remove one dependecny package from final repository
+        data = {
+            "remove_content_units": [
+                pkg.pulp_href
+                for pkg in self.rpm_content_api.list().results
+                if pkg.name == "shark" or pkg.name == "whale"
+            ]
+        }
+        response = self.repo_api.modify(final_repo.pulp_href, data)
+        monitor_task(response.task)
+        final_repo = self.repo_api.read(final_repo.pulp_href)
+
+        # get package to copy
+        whale_package_href = [
+            pkg
+            for pkg in get_content(repo.to_dict())['rpm.package']
+            if pkg['name'] == 'whale'
+        ][0]['pulp_href']
+        package_to_copy = []
+        package_to_copy.append(whale_package_href)
+
+        config = [{
+            'source_repo_version': repo.latest_version_href,
+            'dest_repo': final_repo.pulp_href,
+            'content': package_to_copy
+        }]
+
+        copy_response = rpm_copy(self.cfg, config, recursive=True)
+
+        # check only two pakcages was copied, original package to copy and only one
+        # of its dependency as one is already present
+        self.assertEqual(
+            copy_response['content_summary']['added'][PULP_TYPE_PACKAGE]['count'], 2
+        )