Add deb tests

backports #9185
https://pulp.plan.io/issues/9185

fixes #9194

(cherry picked from commit 5d7101b)

CHANGES/9194.misc

@@ -0,0 +1,2 @@
+Add deb tests
+(backported from #9185)

pulp_2to3_migration/tests/functional/common_plans.py

@@ -5,6 +5,55 @@
 import json
 
 
+DEB_COMPLEX_PLAN = json.dumps({
+    "plugins": [{
+        "type": "deb",
+        "repositories": [
+            {
+                "name": "debian-empty",
+                "pulp2_importer_repository_id": "debian-empty",
+                "repository_versions": [
+                    {
+                        "pulp2_repository_id": "debian-empty",  # content count: 0
+                        "pulp2_distributor_repository_ids": ["debian-empty"]
+                    }
+                ]
+            },
+            {
+                "name": "debian",
+                "pulp2_importer_repository_id": "debian",
+                "repository_versions": [
+                    {
+                        "pulp2_repository_id": "debian",
+                        "pulp2_distributor_repository_ids": ["debian"]
+                    }
+                ]
+            },
+            {
+                "name": "debian-complex-dists",
+                "pulp2_importer_repository_id": "debian-complex-dists",
+                "repository_versions": [
+                    {
+                        "pulp2_repository_id": "debian-complex-dists",
+                        "pulp2_distributor_repository_ids": ["debian-complex-dists"]
+                    }
+                ]
+            },
+            {
+                "name": "debian_update",
+                "pulp2_importer_repository_id": "debian_update",
+                "repository_versions": [
+                    {
+                        "pulp2_repository_id": "debian_update",
+                        "pulp2_distributor_repository_ids": ["debian_update"]
+                    }
+                ]
+            },
+        ],
+
+    }]
+})
+
 FILE_COMPLEX_PLAN = json.dumps({
     "plugins": [{
         "type": "iso",
@@ -116,6 +165,7 @@
 
 FILE_SIMPLE_PLAN = json.dumps({"plugins": [{"type": "iso"}]})
 RPM_SIMPLE_PLAN = json.dumps({"plugins": [{"type": "rpm"}]})
+DEB_SIMPLE_PLAN = json.dumps({"plugins": [{"type": "deb"}]})
 
 FILE_DISTRIBUTOR_DIFF_PLAN = json.dumps({
     "plugins": [{

pulp_2to3_migration/tests/functional/deb_base.py

@@ -0,0 +1,160 @@
+import time
+
+from pulpcore.client.pulpcore import Configuration
+from pulpcore.client.pulp_deb import (
+    ApiClient as DebApiClient,
+    ContentPackageReleaseComponentsApi,
+    ContentPackagesApi,
+    ContentReleaseArchitecturesApi,
+    ContentReleaseComponentsApi,
+    ContentReleasesApi,
+    DistributionsAptApi,
+    PublicationsAptApi,
+    RemotesAptApi,
+    RepositoriesAptApi,
+    RepositoriesDebVersionsApi,
+)
+from pulpcore.client.pulp_2to3_migration import (
+    ApiClient as MigrationApiClient,
+    MigrationPlansApi,
+    Pulp2ContentApi,
+    Pulp2RepositoriesApi,
+)
+
+from pulp_2to3_migration.tests.functional.util import get_psql_smash_cmd
+
+from pulp_smash import cli
+from pulp_smash import config as smash_config
+from pulp_smash.pulp3.bindings import monitor_task, monitor_task_group
+
+from .constants import BINDINGS_CONFIGURATION, TRUNCATE_TABLES_QUERY_BASH
+
+
+class BaseTestDeb:
+    """
+    Test DEB migration re-runs.
+    """
+
+    smash_cfg = smash_config.get_config()
+    smash_cli_client = cli.Client(smash_cfg)
+    plan_initial = None
+    plan_rerun = None
+
+    @classmethod
+    def setUpClass(cls):
+        """
+        Create all the client instances needed to communicate with Pulp.
+        """
+        configuration = Configuration(**BINDINGS_CONFIGURATION)
+
+        deb_client = DebApiClient(configuration)
+        migration_client = MigrationApiClient(configuration)
+
+        # Create api clients for all resource types
+        cls.deb_repo_api = RepositoriesAptApi(deb_client)
+        cls.deb_repo_versions_api = RepositoriesDebVersionsApi(deb_client)
+        cls.deb_remote_api = RemotesAptApi(deb_client)
+        cls.deb_distribution_api = DistributionsAptApi(deb_client)
+        cls.deb_publication_api = PublicationsAptApi(deb_client)
+        cls.deb_content_apis = {
+            "package": ContentPackagesApi(deb_client),
+            "architecture": ContentReleaseArchitecturesApi(deb_client),
+            "component": ContentReleaseComponentsApi(deb_client),
+            "release": ContentReleasesApi(deb_client),
+            "package_component": ContentPackageReleaseComponentsApi(deb_client),
+        }
+        cls.migration_plans_api = MigrationPlansApi(migration_client)
+        cls.pulp2content_api = Pulp2ContentApi(migration_client)
+        cls.pulp2repositories_api = Pulp2RepositoriesApi(migration_client)
+
+    @classmethod
+    def run_migration(cls, plan, run_params={}):
+        """
+        Run a migration plan.
+
+        Args:
+            plan(str): A migration plan to run, in JSON format.
+            run_params(dict): parameters for the `run` call. Optional.
+
+        Returns:
+            task(pulpcore.app.models.Task): a migration task created for this plan
+
+        """
+        mp = cls.migration_plans_api.create({"plan": plan})
+        mp_run_response = cls.migration_plans_api.run(mp.pulp_href, run_params)
+        task = monitor_task(mp_run_response.task)
+        monitor_task_group(task.task_group)
+        return task
+
+    @classmethod
+    def tearDownClass(cls):
+        """
+        Clean up the database after the set of tests is run.
+        """
+        cmd = get_psql_smash_cmd(TRUNCATE_TABLES_QUERY_BASH)
+        cls.smash_cli_client.run(cmd, sudo=True)
+        time.sleep(0.5)
+
+
+class RepoInfo(dict):
+    """
+    Wrapper object for repo information dict.
+
+    Calculates expectations properly depending on whether a simple migration plan is used, or not.
+    """
+
+    def __init__(self, *args, is_simple=False, **kwargs):
+        """Init.
+
+        Args:
+            is_simple (bool): Whether the expectations should be for a simple or complex plan.
+        """
+        super().__init__(*args, **kwargs)
+        self.is_simple = is_simple
+
+    @property
+    def repositories(self):
+        """Return dictionaries of content counts."""
+        repositories = self.get("content_rerun", None) or self["content_initial"]
+        if self.is_simple:
+            return {k: v for k, v in repositories.items() if v}
+        else:
+            return repositories
+
+    @property
+    def new_repositories(self):
+        """Return the dictionaries for new repositories only."""
+        repos = []
+        for repo_name, new_content in self["content_rerun"].items():
+            if self.is_simple:
+                initial_content = self["content_initial"].get(repo_name, None)
+                if not initial_content and new_content:
+                    repos.append(repo_name)
+            else:
+                if repo_name not in self["content_initial"]:
+                    repos.append(repo_name)
+        return repos
+
+    @property
+    def content_total(self):
+        """Return content count dictionary."""
+        # we can't just sum up the content counts because the same content could be in
+        # multiple repos
+        return self["content_total"]
+
+    @property
+    def remotes(self):
+        """Return the count of remotes."""
+        # for complex plans, you can use one remote for many repos, so we can't assume
+        # the number of repos == the number of repositories
+        return self["remotes"]
+
+    @property
+    def publications(self):
+        """Return the count of publications."""
+        return len(self.repositories)
+
+    @property
+    def distributions(self):
+        """Return the count of distributions."""
+        return len(self.repositories)

pulp_2to3_migration/tests/functional/test_deb_repo.py

@@ -0,0 +1,141 @@
+import os
+import unittest
+from pulp_2to3_migration.tests.functional.util import set_pulp2_snapshot
+
+from .common_plans import DEB_SIMPLE_PLAN, DEB_COMPLEX_PLAN
+from .constants import FIXTURES_BASE_URL
+from .deb_base import BaseTestDeb, RepoInfo
+
+PULP_2_DEB_DATA = {
+    "remotes": 3,
+    "content_initial": {
+        "debian-empty": {},
+        "debian": {
+            "package": 4,
+            "package_component": 4,
+            "component": 2,
+            "architecture": 2,
+            "release": 1,
+        },
+        "debian-complex-dists": {
+            "package": 4,
+            "package_component": 4,
+            "component": 2,
+            "architecture": 2,
+            "release": 1,
+        },
+        "debian_update": {
+            "package": 0,
+            "package_component": 0,
+            "component": 2,
+            "architecture": 0,
+            "release": 1,
+        },
+    },
+    "content_total": {
+        "package": 8,
+        "package_component": 8,
+        "component": 6,
+        "architecture": 4,
+        "release": 3,
+    },
+}
+
+
+class BaseTestDebRepo(BaseTestDeb):
+    """
+    Test DEB repo, importer and distributor migration.
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        """
+        Create all the client instances needed to communicate with Pulp and run a migration.
+        """
+        super().setUpClass()
+
+        set_pulp2_snapshot(name="deb_base_4repos")
+        cls.run_migration(cls.plan_initial)
+
+    def test_deb_repo_migration(self):
+        """
+        Test that DEB repos are correctly migrated.
+
+        Check that names are migrated correctly and that the number of versions and content count is
+        correct.
+        """
+        self.assertEqual(
+            self.deb_repo_api.list().count, len(self.repo_info.repositories)
+        )
+
+        # content count in total
+        for content_type, api in self.deb_content_apis.items():
+            with self.subTest(content_type=content_type):
+                self.assertEqual(
+                    api.list().count,
+                    self.repo_info.content_total[content_type],
+                    "content_total[%r] does not match" % content_type,
+                )
+
+        for repo in self.deb_repo_api.list().results:
+            with self.subTest(repo=repo):
+                version_count = 2 if self.repo_info.repositories[repo.name] else 1
+                self.assertEqual(
+                    self.deb_repo_versions_api.list(repo.pulp_href).count, version_count
+                )
+                # content count per repo
+                for content_type, api in self.deb_content_apis.items():
+                    with self.subTest(content_type=content_type):
+                        repo_content = api.list(
+                            repository_version=repo.latest_version_href
+                        )
+                        self.assertEqual(
+                            repo_content.count,
+                            self.repo_info.repositories[repo.name].get(content_type, 0),
+                            "content_type %r does not match for repo %r"
+                            % (content_type, repo.name),
+                        )
+
+    def test_deb_importer_migration(self):
+        """
+        Test that DEB importers are correctly migrated.
+        """
+        self.assertEqual(self.deb_remote_api.list().count, self.repo_info.remotes)
+        for remote in self.deb_remote_api.list().results:
+            with self.subTest(remote=remote):
+                repo_name = "-".join(remote.name.split("-")[1:])
+                repo_url = os.path.join(FIXTURES_BASE_URL, repo_name) + "/"
+                self.assertEqual(remote.url, repo_url)
+
+    def test_deb_distributor_migration(self):
+        """
+        Test that DEB distributors are correctly migrated.
+        """
+        self.assertEqual(
+            self.deb_publication_api.list().count, self.repo_info.publications
+        )
+        self.assertEqual(
+            self.deb_distribution_api.list().count, self.repo_info.distributions
+        )
+        for dist in self.deb_distribution_api.list().results:
+            with self.subTest(dist=dist):
+                base_path = "-".join(dist.name.split("-")[1:])
+                self.assertEqual(dist.base_path, base_path)
+
+
+class TestDebRepoMigrationSimplePlan(BaseTestDebRepo, unittest.TestCase):
+    """
+    Test DEB repo migration using simple migration plan.
+    """
+
+    plan_initial = DEB_SIMPLE_PLAN
+    repo_info = RepoInfo(PULP_2_DEB_DATA, is_simple=True)
+
+
+class TestDebRepoMigrationComplexPlan(BaseTestDebRepo, unittest.TestCase):
+    """
+    Test DEB repo migration using complex migration plan.
+    """
+
+    plan_initial = DEB_COMPLEX_PLAN
+    repo_info = RepoInfo(PULP_2_DEB_DATA)