This repository has been archived by the owner on Jan 9, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 24
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Closes #9185 https://pulp.plan.io/issues/9185
- Loading branch information
1 parent
0867a81
commit 5d7101b
Showing
4 changed files
with
352 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1 @@ | ||
Add deb tests |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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, | ||
RepositoriesAptVersionsApi, | ||
) | ||
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 = RepositoriesAptVersionsApi(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) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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) |