Merge pull request #878 from ATIX-AG/add-advanced-copy-tests

Add tests for advanced copy task

CHANGES/758.misc

@@ -0,0 +1 @@
+Added test cases for advanced copy task.

pulp_deb/tests/functional/api/test_copy.py

@@ -0,0 +1,71 @@
+import pytest
+
+from pulp_deb.tests.functional.constants import (
+    DEB_ADVANCED_COPY_FIXTURE_SUMMARY,
+    DEB_FULL_ADVANCED_COPY_FIXTURE_SUMMARY,
+)
+
+
+@pytest.mark.parallel
+def test_copy(
+    deb_init_and_sync,
+    deb_repository_factory,
+    apt_package_api,
+    deb_copy_content,
+    deb_get_repository_by_href,
+    deb_get_added_content_summary,
+):
+    """Test whether the copy operation can successfully copy a single package."""
+    source_repo, _ = deb_init_and_sync()
+    target_repo = deb_repository_factory()
+    package = apt_package_api.list(package="frigg").results[0]
+    deb_copy_content(
+        source_repo_version=source_repo.latest_version_href,
+        dest_repo=target_repo.pulp_href,
+        content=[package.pulp_href],
+    )
+
+    target_repo = deb_get_repository_by_href(target_repo.pulp_href)
+    assert DEB_ADVANCED_COPY_FIXTURE_SUMMARY == deb_get_added_content_summary(target_repo)
+
+
+@pytest.mark.parallel
+def test_copy_all(
+    deb_init_and_sync,
+    deb_repository_factory,
+    deb_copy_content,
+    deb_get_repository_by_href,
+    deb_get_added_content_summary,
+):
+    """Test whether the copy operation can successfully copy all packages of a repository."""
+    source_repo, _ = deb_init_and_sync()
+    target_repo = deb_repository_factory()
+    deb_copy_content(
+        source_repo_version=source_repo.latest_version_href,
+        dest_repo=target_repo.pulp_href,
+    )
+
+    target_repo = deb_get_repository_by_href(target_repo.pulp_href)
+    assert DEB_FULL_ADVANCED_COPY_FIXTURE_SUMMARY == deb_get_added_content_summary(target_repo)
+
+
+# FIXME: Can be enabled once the following is fixed: https://github.com/pulp/pulp_deb/issues/870
+@pytest.mark.skip("Skip - due to faulty behaviour in the copy task.")
+@pytest.mark.parallel
+def test_copy_empty_content(
+    deb_init_and_sync,
+    deb_repository_factory,
+    deb_copy_content,
+    deb_get_repository_by_href,
+):
+    """Test whether the copy operation does not copy if the content is empty."""
+    source_repo, _ = deb_init_and_sync()
+    target_repo = deb_repository_factory()
+    deb_copy_content(
+        source_repo_version=source_repo.latest_version_href,
+        dest_repo=target_repo.pulp_href,
+        content=[],
+    )
+
+    target_repo = deb_get_repository_by_href(target_repo.pulp_href)
+    assert target_repo.latest_version_href.endswith("/versions/0/")

pulp_deb/tests/functional/conftest.py

@@ -19,7 +19,9 @@
     ContentReleasesApi,
     ContentReleaseComponentsApi,
     ContentReleaseFilesApi,
+    Copy,
     DebAptPublication,
+    DebCopyApi,
     DebVerbatimPublication,
     DistributionsAptApi,
     PublicationsAptApi,
@@ -86,6 +88,12 @@ def apt_verbatim_publication_api(apt_client):
     return PublicationsVerbatimApi(apt_client)
 
 
+@pytest.fixture(scope="session")
+def apt_copy_api(apt_client):
+    """Fixture for APT copy api."""
+    return DebCopyApi(apt_client)
+
+
 @pytest.fixture(scope="session")
 def apt_distribution_api(apt_client):
     """Fixture for APT distribution API."""
@@ -453,6 +461,29 @@ def _deb_sync_repository(remote, repo):
     return _deb_sync_repository
 
 
+@pytest.fixture(scope="class")
+def deb_copy_content(apt_copy_api, monitor_task):
+    """Fixture that copies deb content from a source repository version to a target repository."""
+
+    def _deb_copy_content(source_repo_version, dest_repo, content=None, structured=True):
+        """Copy deb content from a source repository version to a target repository.
+
+        :param source_repo_version: The repository version href from where the content is copied.
+        :dest_repo: The repository href where the content should be copied to.
+        :content: List of package hrefs that should be copied from the source. Default: None
+        :structured: Whether or not the content should be structured copied. Default: True
+        :returns: The task of the copy operation.
+        """
+        config = {"source_repo_version": source_repo_version, "dest_repo": dest_repo}
+        if content is not None:
+            config["content"] = content
+        data = Copy(config=[config], structured=structured)
+        response = apt_copy_api.copy_content(data)
+        return monitor_task(response.task)
+
+    return _deb_copy_content
+
+
 @pytest.fixture(scope="session")
 def deb_signing_script_path(signing_gpg_homedir_path):
     """A fixture for a script that is suited for signing packages."""

pulp_deb/tests/functional/constants.py

@@ -73,6 +73,40 @@ def _clean_dict(d):
     }
 )
 
+DEB_ADVANCED_COPY_FIXTURE_SUMMARY = _clean_dict(
+    {
+        DEB_RELEASE_NAME: 2,
+        # FIXME: this value needs to be adapted once the copy task is refactored.
+        # see: https://github.com/pulp/pulp_deb/issues/870
+        DEB_RELEASE_ARCHITECTURE_NAME: 3,
+        DEB_RELEASE_COMPONENT_NAME: 2,
+        DEB_RELEASE_FILE_NAME: 0,
+        DEB_PACKAGE_INDEX_NAME: 0,
+        DEB_PACKAGE_RELEASE_COMPONENT_NAME: 2,
+        DEB_INSTALLER_FILE_INDEX_NAME: 0,
+        DEB_PACKAGE_NAME: 1,
+        DEB_INSTALLER_PACKAGE_NAME: 0,
+        DEB_GENERIC_CONTENT_NAME: 0,
+    }
+)
+
+DEB_FULL_ADVANCED_COPY_FIXTURE_SUMMARY = _clean_dict(
+    {
+        DEB_RELEASE_NAME: 2,
+        # FIXME: this value needs to be adapted once the copy task is refactored.
+        # see: https://github.com/pulp/pulp_deb/issues/870
+        DEB_RELEASE_ARCHITECTURE_NAME: 3,
+        DEB_RELEASE_COMPONENT_NAME: 3,
+        DEB_RELEASE_FILE_NAME: 0,
+        DEB_PACKAGE_INDEX_NAME: 0,
+        DEB_PACKAGE_RELEASE_COMPONENT_NAME: 7,
+        DEB_INSTALLER_FILE_INDEX_NAME: 0,
+        DEB_PACKAGE_NAME: 4,
+        DEB_INSTALLER_PACKAGE_NAME: 0,
+        DEB_GENERIC_CONTENT_NAME: 0,
+    }
+)
+
 DEB_FIXTURE_PACKAGE_COUNT = DEB_FIXTURE_SUMMARY.get(DEB_PACKAGE_NAME, 0)
 
 DEB_REPORT_CODE_SKIP_RELEASE = "sync.release_file.was_skipped"