diff --git a/flake8.cfg b/flake8.cfg index 0c7c24139..188d8b1c1 100644 --- a/flake8.cfg +++ b/flake8.cfg @@ -1,6 +1,6 @@ [flake8] exclude = ./docs/*,*/migrations/* -ignore = W503,Q000,D100,D104,D106,D200,D401,D402,D202 +ignore = W503,Q000,D100,D104,D106,D200,D401,D402,E203 max-line-length = 100 # Flake8-quotes extension codes @@ -16,4 +16,4 @@ max-line-length = 100 # D200: one-line docstring should fit on one line with quotes # D401: first line should be imperative (nitpicky) # D402: first line should not be the function’s “signature” (false positives) -# D202: no blank line after docstring--disabled until https://github.com/PyCQA/pydocstyle/issues/361 is fixed +# E203: no whitespace around ':'. disabled until https://github.com/PyCQA/pycodestyle/issues/373 is fixed diff --git a/functest_requirements.txt b/functest_requirements.txt index 283bcf391..11db5441a 100644 --- a/functest_requirements.txt +++ b/functest_requirements.txt @@ -1,2 +1,2 @@ -git+https://github.com/PulpQE/pulp-smash.git#egg=pulp-smash +git+https://github.com/pulp/pulp-smash.git#egg=pulp-smash pytest diff --git a/pulp_deb/app/serializers/content_serializers.py b/pulp_deb/app/serializers/content_serializers.py index d1c005447..1870ab63d 100644 --- a/pulp_deb/app/serializers/content_serializers.py +++ b/pulp_deb/app/serializers/content_serializers.py @@ -68,7 +68,6 @@ class GenericContentSerializer(SingleArtifactContentUploadSerializer, ContentChe def deferred_validate(self, data): """Validate the GenericContent data.""" - data = super().deferred_validate(data) data["sha256"] = data["artifact"].sha256 @@ -389,13 +388,12 @@ def __init__(self, *args, **kwargs): def deferred_validate(self, data): """Validate that the artifact is a package and extract it's values.""" - data = super().deferred_validate(data) try: package_paragraph = debfile.DebFile(fileobj=data["artifact"].file).debcontrol() except Exception: # TODO: Be more specific - raise ValidationError(_("Not a valid Deb Package")) + raise ValidationError(_("Unable to read Deb Package")) from822_serializer = self.Meta.from822_serializer.from822(data=package_paragraph) from822_serializer.is_valid(raise_exception=True) @@ -469,7 +467,6 @@ class PackageSerializer(BasePackageSerializer): def deferred_validate(self, data): """Validate for 'normal' Package (not installer).""" - data = super().deferred_validate(data) if data.get("section") == "debian-installer": @@ -489,7 +486,6 @@ class InstallerPackageSerializer(BasePackageSerializer): def deferred_validate(self, data): """Validate for InstallerPackage.""" - data = super().deferred_validate(data) if data.get("section") != "debian-installer": diff --git a/pulp_deb/tests/functional/api/test_crud_content_unit.py b/pulp_deb/tests/functional/api/test_crud_content_unit.py index 7971f31f8..2bba1ad85 100644 --- a/pulp_deb/tests/functional/api/test_crud_content_unit.py +++ b/pulp_deb/tests/functional/api/test_crud_content_unit.py @@ -2,24 +2,25 @@ """Tests that perform actions over content unit.""" import unittest -from requests.exceptions import HTTPError +from tempfile import NamedTemporaryFile -from pulp_smash import api, config, utils -from pulp_smash.exceptions import TaskReportError -from pulp_smash.pulp3.constants import ARTIFACTS_PATH +from pulp_smash import utils from pulp_smash.pulp3.utils import delete_orphans from pulp_deb.tests.functional.constants import ( - DEB_GENERIC_CONTENT_PATH, DEB_GENERIC_CONTENT_URL, - DEB_PACKAGE_PATH, DEB_PACKAGE_URL, ) from pulp_deb.tests.functional.utils import ( + deb_generic_content_api, + deb_package_api, + gen_artifact, gen_deb_content_attrs, gen_deb_content_upload_attrs, gen_deb_package_attrs, gen_deb_package_upload_attrs, + monitor_task, + PulpTaskError, skip_if, ) from pulp_deb.tests.functional.utils import set_up_module as setUpModule # noqa:F401 @@ -32,43 +33,41 @@ class GenericContentUnitTestCase(unittest.TestCase): * `Pulp #2872 `_ * `Pulp #3445 `_ - * `Pulp Smash #870 `_ + * `Pulp Smash #870 `_ """ gen_content_attrs = staticmethod(gen_deb_content_attrs) gen_content_verify_attrs = staticmethod(gen_deb_content_attrs) - CONTENT_PATH = DEB_GENERIC_CONTENT_PATH + content_api = deb_generic_content_api CONTENT_URL = DEB_GENERIC_CONTENT_URL @classmethod def setUpClass(cls): """Create class-wide variable.""" - cls.cfg = config.get_config() - delete_orphans(cls.cfg) + delete_orphans() cls.content_unit = {} - cls.client = api.Client(cls.cfg, api.json_handler) - files = {"file": utils.http_get(cls.CONTENT_URL)} - cls.artifact = cls.client.post(ARTIFACTS_PATH, files=files) + cls.artifact = gen_artifact(cls.CONTENT_URL) @classmethod def tearDownClass(cls): """Clean class-wide variable.""" - delete_orphans(cls.cfg) + delete_orphans() def test_01_create_content_unit(self): """Create content unit.""" attrs = self.gen_content_attrs(self.artifact) - call_report = self.client.post(self.CONTENT_PATH, data=attrs) - created_resources = next(api.poll_spawned_tasks(self.cfg, call_report))["created_resources"] - self.content_unit.update(self.client.get(created_resources[0])) - for key, val in self.gen_content_verify_attrs(self.artifact).items(): + response = self.content_api.create(**attrs) + created_resources = monitor_task(response.task) + content_unit = self.content_api.read(created_resources[0]) + self.content_unit.update(content_unit.to_dict()) + for key, val in attrs.items(): with self.subTest(key=key): self.assertEqual(self.content_unit[key], val) @skip_if(bool, "content_unit", False) def test_02_read_content_unit(self): """Read a content unit by its href.""" - content_unit = self.client.get(self.content_unit["pulp_href"]) + content_unit = self.content_api.read(self.content_unit["pulp_href"]).to_dict() for key, val in self.content_unit.items(): with self.subTest(key=key): self.assertEqual(content_unit[key], val) @@ -76,13 +75,11 @@ def test_02_read_content_unit(self): @skip_if(bool, "content_unit", False) def test_02_read_content_units(self): """Read a content unit by its relative_path.""" - page = self.client.get( - self.CONTENT_PATH, params={"relative_path": self.content_unit["relative_path"]} - ) - self.assertEqual(len(page["results"]), 1) + page = self.content_api.list(relative_path=self.content_unit["relative_path"]) + self.assertEqual(len(page.results), 1) for key, val in self.content_unit.items(): with self.subTest(key=key): - self.assertEqual(page["results"][0][key], val) + self.assertEqual(page.results[0].to_dict()[key], val) @skip_if(bool, "content_unit", False) def test_03_partially_update(self): @@ -91,9 +88,10 @@ def test_03_partially_update(self): This HTTP method is not supported and a HTTP exception is expected. """ attrs = self.gen_content_attrs(self.artifact) - with self.assertRaises(HTTPError) as exc: - self.client.patch(self.content_unit["pulp_href"], attrs) - self.assertEqual(exc.exception.response.status_code, 405) + with self.assertRaises(AttributeError) as exc: + self.content_api.partial_update(self.content_unit["pulp_href"], attrs) + msg = "object has no attribute 'partial_update'" + self.assertIn(msg, exc.exception.args[0]) @skip_if(bool, "content_unit", False) def test_03_fully_update(self): @@ -102,9 +100,10 @@ def test_03_fully_update(self): This HTTP method is not supported and a HTTP exception is expected. """ attrs = self.gen_content_attrs(self.artifact) - with self.assertRaises(HTTPError) as exc: - self.client.put(self.content_unit["pulp_href"], attrs) - self.assertEqual(exc.exception.response.status_code, 405) + with self.assertRaises(AttributeError) as exc: + self.content_api.update(self.content_unit["pulp_href"], attrs) + msg = "object has no attribute 'update'" + self.assertIn(msg, exc.exception.args[0]) @skip_if(bool, "content_unit", False) def test_04_delete(self): @@ -112,9 +111,10 @@ def test_04_delete(self): This HTTP method is not supported and a HTTP exception is expected. """ - with self.assertRaises(HTTPError) as exc: - self.client.delete(self.content_unit["pulp_href"]) - self.assertEqual(exc.exception.response.status_code, 405) + with self.assertRaises(AttributeError) as exc: + self.content_api.delete(self.content_unit["pulp_href"]) + msg = "object has no attribute 'delete'" + self.assertIn(msg, exc.exception.args[0]) class PackageTestCase(GenericContentUnitTestCase): @@ -122,7 +122,7 @@ class PackageTestCase(GenericContentUnitTestCase): gen_content_attrs = staticmethod(gen_deb_package_attrs) gen_content_verify_attrs = staticmethod(gen_deb_package_attrs) - CONTENT_PATH = DEB_PACKAGE_PATH + content_api = deb_package_api CONTENT_URL = DEB_PACKAGE_URL @@ -136,28 +136,31 @@ class GenericContentUnitUploadTestCase(unittest.TestCase): gen_content_upload_attrs = staticmethod(gen_deb_content_upload_attrs) gen_content_upload_verify_attrs = staticmethod(gen_deb_content_upload_attrs) - CONTENT_PATH = DEB_GENERIC_CONTENT_PATH + content_api = deb_generic_content_api CONTENT_URL = DEB_GENERIC_CONTENT_URL @classmethod def setUpClass(cls): """Create class-wide variable.""" - cls.cfg = config.get_config() - delete_orphans(cls.cfg) + delete_orphans() cls.content_unit = {} - cls.client = api.Client(cls.cfg, api.smart_handler) - cls.files = {"file": utils.http_get(cls.CONTENT_URL)} + cls.file = utils.http_get(cls.CONTENT_URL) cls.attrs = cls.gen_content_upload_attrs() @classmethod def tearDownClass(cls): """Clean class-wide variable.""" - delete_orphans(cls.cfg) + delete_orphans() def test_01_create_content_unit(self): """Create content unit.""" - content_unit = self.client.post(self.CONTENT_PATH, data=self.attrs, files=self.files) - self.content_unit.update(content_unit) + with NamedTemporaryFile() as temp_file: + temp_file.write(self.file) + temp_file.flush() + response = self.content_api.create(**self.attrs, file=temp_file.name) + created_resources = monitor_task(response.task) + content_unit = self.content_api.read(created_resources[0]) + self.content_unit.update(content_unit.to_dict()) for key, val in self.attrs.items(): with self.subTest(key=key): self.assertEqual(self.content_unit[key], val) @@ -165,7 +168,7 @@ def test_01_create_content_unit(self): @skip_if(bool, "content_unit", False) def test_02_read_content_unit(self): """Read a content unit by its href.""" - content_unit = self.client.get(self.content_unit["pulp_href"]) + content_unit = self.content_api.read(self.content_unit["pulp_href"]).to_dict() for key, val in self.content_unit.items(): with self.subTest(key=key): self.assertEqual(content_unit[key], val) @@ -173,21 +176,23 @@ def test_02_read_content_unit(self): @skip_if(bool, "content_unit", False) def test_02_read_content_units(self): """Read a content unit by its relative_path.""" - page = self.client.using_handler(api.json_handler).get( - self.CONTENT_PATH, params={"relative_path": self.content_unit["relative_path"]} - ) - self.assertEqual(len(page["results"]), 1) + page = self.content_api.list(relative_path=self.content_unit["relative_path"]) + self.assertEqual(len(page.results), 1) for key, val in self.content_unit.items(): with self.subTest(key=key): - self.assertEqual(page["results"][0][key], val) + self.assertEqual(page.results[0].to_dict()[key], val) @skip_if(bool, "content_unit", False) def test_03_fail_duplicate_content_unit(self): """Create content unit.""" - with self.assertRaises(TaskReportError) as exc: - self.client.post(self.CONTENT_PATH, data=self.attrs, files=self.files) - self.assertEqual(exc.exception.task["state"], "failed") - error = exc.exception.task["error"] + with NamedTemporaryFile() as temp_file: + temp_file.write(self.file) + temp_file.flush() + response = self.content_api.create(**self.attrs, file=temp_file.name) + with self.assertRaises(PulpTaskError) as exc: + monitor_task(response.task) + self.assertEqual(exc.exception.task.state, "failed") + error = exc.exception.task.error for key in ("already", "relative", "path", "sha256"): self.assertIn(key, error["description"].lower(), error) @@ -197,7 +202,10 @@ def test_03_duplicate_content_unit(self): attrs = self.attrs.copy() # Packages types only validate the filename, so we can prepend something to the path. attrs["relative_path"] = "moved-" + self.content_unit["relative_path"] - self.client.post(self.CONTENT_PATH, data=attrs, files=self.files) + with NamedTemporaryFile() as temp_file: + temp_file.write(self.file) + temp_file.flush() + self.content_api.create(**attrs, file=temp_file.name) class PackageUnitUploadTestCase(GenericContentUnitUploadTestCase): @@ -205,7 +213,7 @@ class PackageUnitUploadTestCase(GenericContentUnitUploadTestCase): gen_content_upload_attrs = staticmethod(gen_deb_package_upload_attrs) gen_content_upload_verify_attrs = staticmethod(gen_deb_package_upload_attrs) - CONTENT_PATH = DEB_PACKAGE_PATH + content_api = deb_package_api CONTENT_URL = DEB_PACKAGE_URL @@ -219,19 +227,13 @@ class DuplicateGenericContentUnit(unittest.TestCase): gen_content_attrs = staticmethod(gen_deb_content_attrs) gen_content_verify_attrs = staticmethod(gen_deb_content_attrs) - CONTENT_PATH = DEB_GENERIC_CONTENT_PATH + content_api = deb_generic_content_api CONTENT_URL = DEB_GENERIC_CONTENT_URL - @classmethod - def setUpClass(cls): - """Create class-wide variables.""" - cls.cfg = config.get_config() - cls.client = api.Client(cls.cfg, api.smart_handler) - @classmethod def tearDownClass(cls): """Clean created resources.""" - delete_orphans(cls.cfg) + delete_orphans() def test_raise_error(self): """Create a duplicate content unit using same relative_path. @@ -241,19 +243,21 @@ def test_raise_error(self): In order to raise an HTTP error, the same ``artifact`` and the same ``relative_path`` should be used. """ - delete_orphans(self.cfg) - files = {"file": utils.http_get(self.CONTENT_URL)} - artifact = self.client.post(ARTIFACTS_PATH, files=files) + delete_orphans() + artifact = gen_artifact(self.CONTENT_URL) attrs = self.gen_content_attrs(artifact) # create first content unit. - self.client.post(self.CONTENT_PATH, attrs) + response = self.content_api.create(**attrs) + created_resources = monitor_task(response.task) + self.content_api.read(created_resources[0]) # using the same attrs used to create the first content unit. - with self.assertRaises(TaskReportError) as exc: - self.client.post(self.CONTENT_PATH, attrs) - self.assertEqual(exc.exception.task["state"], "failed") - error = exc.exception.task["error"] + response = self.content_api.create(**attrs) + with self.assertRaises(PulpTaskError) as exc: + monitor_task(response.task) + self.assertEqual(exc.exception.task.state, "failed") + error = exc.exception.task.error for key in ("already", "relative", "path", "sha256"): self.assertIn(key, error["description"].lower(), error) @@ -265,18 +269,21 @@ def test_non_error(self): In order to avoid an HTTP error, use the same ``artifact`` and different ``relative_path``. """ - delete_orphans(self.cfg) - files = {"file": utils.http_get(self.CONTENT_URL)} - artifact = self.client.post(ARTIFACTS_PATH, files=files) + delete_orphans() + artifact = gen_artifact(self.CONTENT_URL) attrs = self.gen_content_attrs(artifact) # create first content unit. - content_unit = self.client.post(self.CONTENT_PATH, attrs) + response = self.content_api.create(**attrs) + created_resources = monitor_task(response.task) + content_unit = self.content_api.read(created_resources[0]) # Packages types only validate the filename, so we can prepend something to the path. - attrs["relative_path"] = "moved-" + content_unit["relative_path"] + attrs["relative_path"] = "moved-" + content_unit.relative_path # create second content unit. - self.client.post(self.CONTENT_PATH, attrs) + response = self.content_api.create(**attrs) + created_resources = monitor_task(response.task) + content_unit = self.content_api.read(created_resources[0]) class DuplicatePackageUnit(DuplicateGenericContentUnit): @@ -284,5 +291,5 @@ class DuplicatePackageUnit(DuplicateGenericContentUnit): gen_content_attrs = staticmethod(gen_deb_package_attrs) gen_content_verify_attrs = staticmethod(gen_deb_package_attrs) - CONTENT_PATH = DEB_PACKAGE_PATH + content_api = deb_package_api CONTENT_URL = DEB_PACKAGE_URL diff --git a/pulp_deb/tests/functional/api/test_crud_remotes.py b/pulp_deb/tests/functional/api/test_crud_remotes.py index 929d1e75d..cfee64f42 100644 --- a/pulp_deb/tests/functional/api/test_crud_remotes.py +++ b/pulp_deb/tests/functional/api/test_crud_remotes.py @@ -3,14 +3,19 @@ from random import choice import unittest -from requests.exceptions import HTTPError - -from pulp_smash import api, config, utils - -from pulp_deb.tests.functional.constants import DOWNLOAD_POLICIES, DEB_REMOTE_PATH -from pulp_deb.tests.functional.utils import skip_if, gen_deb_remote +from pulp_smash import utils + +from pulp_deb.tests.functional.constants import DOWNLOAD_POLICIES +from pulp_deb.tests.functional.utils import ( + deb_remote_api, + gen_deb_remote, + monitor_task, + skip_if, +) from pulp_deb.tests.functional.utils import set_up_module as setUpModule # noqa:F401 +from pulpcore.client.pulp_deb.exceptions import ApiException + class CRUDRemotesTestCase(unittest.TestCase): """CRUD remotes.""" @@ -18,78 +23,80 @@ class CRUDRemotesTestCase(unittest.TestCase): @classmethod def setUpClass(cls): """Create class-wide variables.""" - cls.cfg = config.get_config() - cls.client = api.Client(cls.cfg, api.json_handler) + cls.remote_api = deb_remote_api def test_01_create_remote(self): """Create a remote.""" body = _gen_verbose_remote() - type(self).remote = self.client.post(DEB_REMOTE_PATH, body) + type(self).remote = self.remote_api.create(body) for key in ("username", "password"): del body[key] for key, val in body.items(): with self.subTest(key=key): - self.assertEqual(self.remote[key], val) + self.assertEqual(self.remote.to_dict()[key], val, key) @skip_if(bool, "remote", False) def test_02_create_same_name(self): """Try to create a second remote with an identical name. See: `Pulp Smash #1055 - `_. + `_. """ body = gen_deb_remote() - body["name"] = self.remote["name"] - with self.assertRaises(HTTPError): - self.client.post(DEB_REMOTE_PATH, body) + body["name"] = self.remote.name + with self.assertRaises(ApiException): + self.remote_api.create(body) @skip_if(bool, "remote", False) def test_02_read_remote(self): """Read a remote by its href.""" - remote = self.client.get(self.remote["pulp_href"]) - for key, val in self.remote.items(): + remote = self.remote_api.read(self.remote.pulp_href) + for key, val in self.remote.to_dict().items(): with self.subTest(key=key): - self.assertEqual(remote[key], val) + self.assertEqual(remote.to_dict()[key], val, key) @skip_if(bool, "remote", False) def test_02_read_remotes(self): """Read a remote by its name.""" - page = self.client.get(DEB_REMOTE_PATH, params={"name": self.remote["name"]}) - self.assertEqual(len(page["results"]), 1) - for key, val in self.remote.items(): + page = self.remote_api.list(name=self.remote.name) + self.assertEqual(len(page.results), 1) + for key, val in self.remote.to_dict().items(): with self.subTest(key=key): - self.assertEqual(page["results"][0][key], val) + self.assertEqual(page.results[0].to_dict()[key], val, key) @skip_if(bool, "remote", False) def test_03_partially_update(self): """Update a remote using HTTP PATCH.""" body = _gen_verbose_remote() - self.client.patch(self.remote["pulp_href"], body) + response = self.remote_api.partial_update(self.remote.pulp_href, body) + monitor_task(response.task) for key in ("username", "password"): del body[key] - type(self).remote = self.client.get(self.remote["pulp_href"]) + type(self).remote = self.remote_api.read(self.remote.pulp_href) for key, val in body.items(): with self.subTest(key=key): - self.assertEqual(self.remote[key], val) + self.assertEqual(self.remote.to_dict()[key], val, key) @skip_if(bool, "remote", False) def test_04_fully_update(self): """Update a remote using HTTP PUT.""" body = _gen_verbose_remote() - self.client.put(self.remote["pulp_href"], body) + response = self.remote_api.update(self.remote.pulp_href, body) + monitor_task(response.task) for key in ("username", "password"): del body[key] - type(self).remote = self.client.get(self.remote["pulp_href"]) + type(self).remote = self.remote_api.read(self.remote.pulp_href) for key, val in body.items(): with self.subTest(key=key): - self.assertEqual(self.remote[key], val) + self.assertEqual(self.remote.to_dict()[key], val, key) @skip_if(bool, "remote", False) def test_05_delete(self): """Delete a remote.""" - self.client.delete(self.remote["pulp_href"]) - with self.assertRaises(HTTPError): - self.client.get(self.remote["pulp_href"]) + response = self.remote_api.delete(self.remote.pulp_href) + monitor_task(response.task) + with self.assertRaises(ApiException): + self.remote_api.read(self.remote.pulp_href) class CreateRemoteNoURLTestCase(unittest.TestCase): @@ -101,12 +108,93 @@ def test_all(self): This test targets the following issues: * `Pulp #3395 `_ - * `Pulp Smash #984 `_ + * `Pulp Smash #984 `_ """ body = gen_deb_remote() del body["url"] - with self.assertRaises(HTTPError): - api.Client(config.get_config()).post(DEB_REMOTE_PATH, body) + with self.assertRaises(ApiException): + deb_remote_api.create(body) + + +class RemoteDownloadPolicyTestCase(unittest.TestCase): + """Verify download policy behavior for valid and invalid values. + + In Pulp 3, there are are different download policies. + + This test targets the following testing scenarios: + + 1. Creating a remote without a download policy. + Verify the creation is successful and immediate it is policy applied. + 2. Change the remote policy from default. + Verify the change is successful. + 3. Attempt to change the remote policy to an invalid string. + Verify an ApiException is given for the invalid policy as well + as the policy remaining unchanged. + + For more information on the remote policies, see the Pulp3 + API on an installed server: + + * /pulp/api/v3/docs/#operation` + + This test targets the following issues: + + * `Pulp #4420 `_ + * `Pulp #3763 `_ + """ + + @classmethod + def setUpClass(cls): + """Create class-wide variables.""" + cls.remote_api = deb_remote_api + cls.remote = {} + cls.policies = DOWNLOAD_POLICIES + cls.body = _gen_verbose_remote() + + @classmethod + def tearDownClass(cls): + """Clean class-wide variable.""" + results = cls.remote_api.list().to_dict()["results"] + for result in results: + cls.remote_api.delete(result["pulp_href"]) + + def test_01_no_defined_policy(self): + """Verify the default policy `immediate`. + + When no policy is defined, the default policy of `immediate` + is applied. + """ + del self.body["policy"] + self.remote.update(self.remote_api.create(self.body).to_dict()) + self.assertEqual(self.remote["policy"], "immediate", self.remote) + + @skip_if(len, "policies", 1) + def test_02_change_policy(self): + """Verify ability to change policy to value other than the default. + + Update the remote policy to a valid value other than `immedaite` + and verify the new set value. + """ + changed_policy = choice([item for item in self.policies if item != "immediate"]) + response = self.remote_api.partial_update( + self.remote["pulp_href"], {"policy": changed_policy} + ) + monitor_task(response.task) + self.remote.update(self.remote_api.read(self.remote["pulp_href"]).to_dict()) + self.assertEqual(self.remote["policy"], changed_policy, self.remote) + + @skip_if(bool, "remote", False) + def test_03_invalid_policy(self): + """Verify an invalid policy does not update the remote policy. + + Get the current remote policy. + Attempt to update the remote policy to an invalid value. + Verify the policy remains the same. + """ + remote = self.remote_api.read(self.remote["pulp_href"]).to_dict() + with self.assertRaises(ApiException): + self.remote_api.partial_update(self.remote["pulp_href"], {"policy": utils.uuid4()}) + self.remote.update(self.remote_api.read(self.remote["pulp_href"]).to_dict()) + self.assertEqual(remote["policy"], self.remote["policy"], self.remote) def _gen_verbose_remote(): diff --git a/pulp_deb/tests/functional/api/test_download_content.py b/pulp_deb/tests/functional/api/test_download_content.py index 05f140e46..a5aae7328 100644 --- a/pulp_deb/tests/functional/api/test_download_content.py +++ b/pulp_deb/tests/functional/api/test_download_content.py @@ -5,34 +5,42 @@ from random import choice from urllib.parse import urljoin -from pulp_smash import api, config, utils -from pulp_smash.pulp3.utils import download_content_unit, gen_distribution, gen_repo, sync - -from pulp_deb.tests.functional.constants import ( - DEB_DISTRIBUTION_PATH, - DEB_FIXTURE_URL, - DEB_REMOTE_PATH, - DEB_REPO_PATH, - DOWNLOAD_POLICIES, -) +from pulp_smash import config, utils +from pulp_smash.pulp3.utils import download_content_unit, gen_distribution, gen_repo + +from pulp_deb.tests.functional.constants import DEB_FIXTURE_URL, DOWNLOAD_POLICIES from pulp_deb.tests.functional.utils import ( - create_deb_publication, - create_verbatim_publication, + deb_repository_api, + deb_remote_api, + deb_apt_publication_api, + deb_verbatim_publication_api, + deb_distribution_api, gen_deb_remote, get_deb_content_unit_paths, get_deb_verbatim_content_unit_paths, + monitor_task, ) from pulp_deb.tests.functional.utils import set_up_module as setUpModule # noqa:F401 +from pulpcore.client.pulp_deb import ( + RepositorySyncURL, + DebDebPublication, + DebVerbatimPublication, +) + class DownloadContentTestCase(unittest.TestCase): """Verify whether content served by pulp can be downloaded.""" class Meta: - create_publication = create_deb_publication - DISTRIBUTION_PATH = DEB_DISTRIBUTION_PATH + publication_api = deb_apt_publication_api get_content_unit_paths = get_deb_content_unit_paths + @staticmethod + def Publication(*args, **kwargs): + """Delagate Publication constructor.""" + return DebDebPublication(structured=True, simple=True, *args, **kwargs) + def test_all(self): """Download content from Pulp. Content is synced with different policies. @@ -66,32 +74,39 @@ def do_test(self, policy): This test targets the following issues: * `Pulp #2895 `_ - * `Pulp Smash #872 `_ + * `Pulp Smash #872 `_ """ - cfg = config.get_config() - client = api.Client(cfg, api.json_handler) + repo_api = deb_repository_api + remote_api = deb_remote_api + publication_api = self.Meta.publication_api + distribution_api = deb_distribution_api - repo = client.post(DEB_REPO_PATH, gen_repo()) - self.addCleanup(client.delete, repo["pulp_href"]) + repo = repo_api.create(gen_repo()) + self.addCleanup(repo_api.delete, repo.pulp_href) body = gen_deb_remote() - remote = client.post(DEB_REMOTE_PATH, body) - self.addCleanup(client.delete, remote["pulp_href"]) + remote = remote_api.create(body) + self.addCleanup(remote_api.delete, remote.pulp_href) - sync(cfg, remote, repo) - repo = client.get(repo["pulp_href"]) + # Sync a Repository + repository_sync_data = RepositorySyncURL(remote=remote.pulp_href) + sync_response = repo_api.sync(repo.pulp_href, repository_sync_data) + monitor_task(sync_response.task) + repo = repo_api.read(repo.pulp_href) # Create a publication. - publication = self.Meta.create_publication(cfg, repo) - self.addCleanup(client.delete, publication["pulp_href"]) + publish_data = self.Meta.Publication(repository=repo.pulp_href) + publish_response = publication_api.create(publish_data) + publication_href = monitor_task(publish_response.task)[0] + self.addCleanup(publication_api.delete, publication_href) # Create a distribution. body = gen_distribution() - body["publication"] = publication["pulp_href"] - distribution = client.using_handler(api.task_handler).post( - self.Meta.DISTRIBUTION_PATH, body - ) - self.addCleanup(client.delete, distribution["pulp_href"]) + body["publication"] = publication_href + distribution_response = distribution_api.create(body) + distribution_href = monitor_task(distribution_response.task)[0] + distribution = distribution_api.read(distribution_href) + self.addCleanup(distribution_api.delete, distribution.pulp_href) # Pick a content unit (of each type), and download it from both Pulp Fixtures… unit_paths = [ @@ -103,10 +118,12 @@ def do_test(self, policy): ] # …and Pulp. - contents = [ - download_content_unit(cfg, distribution, unit_path[1]) for unit_path in unit_paths - ] - pulp_hashes = [hashlib.sha256(content).hexdigest() for content in contents] + pulp_hashes = [] + cfg = config.get_config() + for unit_path in unit_paths: + content = download_content_unit(cfg, distribution.to_dict(), unit_path[1]) + pulp_hashes.append(hashlib.sha256(content).hexdigest()) + self.assertEqual(fixtures_hashes, pulp_hashes) @@ -114,6 +131,6 @@ class VerbatimDownloadContentTestCase(DownloadContentTestCase): """Verify whether content served by pulp can be downloaded.""" class Meta: - create_publication = create_verbatim_publication - DISTRIBUTION_PATH = DEB_DISTRIBUTION_PATH + Publication = DebVerbatimPublication + publication_api = deb_verbatim_publication_api get_content_unit_paths = get_deb_verbatim_content_unit_paths diff --git a/pulp_deb/tests/functional/api/test_download_policies.py b/pulp_deb/tests/functional/api/test_download_policies.py index eb8b34512..520261ca0 100644 --- a/pulp_deb/tests/functional/api/test_download_policies.py +++ b/pulp_deb/tests/functional/api/test_download_policies.py @@ -1,44 +1,52 @@ # coding=utf-8 """Tests for Pulp`s download policies.""" -from random import choice import unittest -from pulp_smash import api, config -from pulp_smash.pulp3.constants import ARTIFACTS_PATH, ON_DEMAND_DOWNLOAD_POLICIES from pulp_smash.pulp3.utils import ( delete_orphans, gen_repo, get_added_content_summary, get_content_summary, - sync, ) from pulp_deb.tests.functional.constants import ( DEB_FIXTURE_PACKAGE_COUNT, DEB_FIXTURE_SUMMARY, - DEB_PACKAGE_PATH, - DEB_REMOTE_PATH, - DEB_REPO_PATH, + DOWNLOAD_POLICIES, +) +from pulp_deb.tests.functional.utils import ( + artifact_api, + deb_package_api, + deb_apt_publication_api, + deb_repository_api, + deb_remote_api, + gen_deb_remote, + monitor_task, + skip_if, ) -from pulp_deb.tests.functional.utils import create_deb_publication, gen_deb_remote from pulp_deb.tests.functional.utils import set_up_module as setUpModule # noqa:F401 +from pulpcore.client.pulp_deb import ( + DebDebPublication, + RepositorySyncURL, +) + class SyncPublishDownloadPolicyTestCase(unittest.TestCase): """Sync/Publish a repository with different download policies. - This test targets the following issues: + This test targets the following issue: `Pulp #4126 `_ - `Pulp #4418 `_ """ @classmethod def setUpClass(cls): """Create class-wide variables.""" - cls.cfg = config.get_config() - cls.client = api.Client(cls.cfg, api.page_handler) + cls.DP_ON_DEMAND = "on_demand" in DOWNLOAD_POLICIES + cls.DP_STREAMED = "streamed" in DOWNLOAD_POLICIES + @skip_if(bool, "DP_ON_DEMAND", False) def test_on_demand(self): """Sync and publish with ``on_demand`` download policy. @@ -48,6 +56,7 @@ def test_on_demand(self): self.do_sync("on_demand") self.do_publish("on_demand") + @skip_if(bool, "DP_STREAMED", False) def test_streamed(self): """Sync and publish with ``streamend`` download policy. @@ -77,45 +86,61 @@ def do_sync(self, download_policy): """ # delete orphans to assure that no content units are present on the # file system - delete_orphans(self.cfg) - repo = self.client.post(DEB_REPO_PATH, gen_repo()) - self.addCleanup(self.client.delete, repo["pulp_href"]) + delete_orphans() + repo_api = deb_repository_api + remote_api = deb_remote_api + + repo = repo_api.create(gen_repo()) + self.addCleanup(repo_api.delete, repo.pulp_href) body = gen_deb_remote(policy=download_policy) - remote = self.client.post(DEB_REMOTE_PATH, body) - self.addCleanup(self.client.delete, remote["pulp_href"]) + remote = remote_api.create(body) + self.addCleanup(remote_api.delete, remote.pulp_href) # Sync the repository. - self.assertEqual(repo["latest_version_href"], f"{repo['pulp_href']}versions/0/") - sync(self.cfg, remote, repo) - repo = self.client.get(repo["pulp_href"]) + self.assertEqual(repo.latest_version_href, f"{repo.pulp_href}versions/0/") + repository_sync_data = RepositorySyncURL(remote=remote.pulp_href) + sync_response = repo_api.sync(repo.pulp_href, repository_sync_data) + monitor_task(sync_response.task) + repo = repo_api.read(repo.pulp_href) - self.assertIsNotNone(repo["latest_version_href"]) - self.assertDictEqual(get_content_summary(repo), DEB_FIXTURE_SUMMARY) - self.assertDictEqual(get_added_content_summary(repo), DEB_FIXTURE_SUMMARY) + self.assertIsNotNone(repo.latest_version_href) + self.assertDictEqual(get_content_summary(repo.to_dict()), DEB_FIXTURE_SUMMARY) + self.assertDictEqual(get_added_content_summary(repo.to_dict()), DEB_FIXTURE_SUMMARY) # Sync the repository again. - latest_version_href = repo["latest_version_href"] - sync(self.cfg, remote, repo) - repo = self.client.get(repo["pulp_href"]) + latest_version_href = repo.latest_version_href + sync_response = repo_api.sync(repo.pulp_href, repository_sync_data) + monitor_task(sync_response.task) + repo = repo_api.read(repo.pulp_href) - self.assertEqual(latest_version_href, repo["latest_version_href"]) - self.assertDictEqual(get_content_summary(repo), DEB_FIXTURE_SUMMARY) + self.assertEqual(latest_version_href, repo.latest_version_href) + self.assertDictEqual(get_content_summary(repo.to_dict()), DEB_FIXTURE_SUMMARY) def do_publish(self, download_policy): """Publish repository synced with lazy download policy.""" - repo = self.client.post(DEB_REPO_PATH, gen_repo()) - self.addCleanup(self.client.delete, repo["pulp_href"]) + publication_api = deb_apt_publication_api + repo_api = deb_repository_api + remote_api = deb_remote_api + + repo = repo_api.create(gen_repo()) + self.addCleanup(repo_api.delete, repo.pulp_href) body = gen_deb_remote(policy=download_policy) - remote = self.client.post(DEB_REMOTE_PATH, body) - self.addCleanup(self.client.delete, remote["pulp_href"]) + remote = remote_api.create(body) + self.addCleanup(remote_api.delete, remote.pulp_href) - sync(self.cfg, remote, repo) - repo = self.client.get(repo["pulp_href"]) + repository_sync_data = RepositorySyncURL(remote=remote.pulp_href) + sync_response = repo_api.sync(repo.pulp_href, repository_sync_data) + monitor_task(sync_response.task) + repo = repo_api.read(repo.pulp_href) - publication = create_deb_publication(self.cfg, repo) - self.assertIsNotNone(publication["repository_version"], publication) + publish_data = DebDebPublication(simple=True, repository=repo.pulp_href) + publish_response = publication_api.create(publish_data) + publication_href = monitor_task(publish_response.task)[0] + self.addCleanup(publication_api.delete, publication_href) + publication = publication_api.read(publication_href) + self.assertIsNotNone(publication.repository_version, publication) class LazySyncedContentAccessTestCase(unittest.TestCase): @@ -133,13 +158,15 @@ class LazySyncedContentAccessTestCase(unittest.TestCase): @classmethod def setUpClass(cls): """Create class-wide variables.""" - cls.cfg = config.get_config() - cls.client = api.Client(cls.cfg, api.page_handler) + cls.DP_ON_DEMAND = "on_demand" in DOWNLOAD_POLICIES + cls.DP_STREAMED = "streamed" in DOWNLOAD_POLICIES + @skip_if(bool, "DP_ON_DEMAND", False) def test_on_demand(self): """Test ``on_demand``. See :meth:`do_test`.""" self.do_test("on_demand") + @skip_if(bool, "DP_STREAMED", False) def test_streamed(self): """Test ``streamed``. See :meth:`do_test`.""" self.do_test("streamed") @@ -148,25 +175,31 @@ def do_test(self, policy): """Access lazy synced content on using content endpoint.""" # delete orphans to assure that no content units are present on the # file system - delete_orphans(self.cfg) - repo = self.client.post(DEB_REPO_PATH, gen_repo()) - self.addCleanup(self.client.delete, repo["pulp_href"]) + delete_orphans() + repo_api = deb_repository_api + remote_api = deb_remote_api + packages_api = deb_package_api + + repo = repo_api.create(gen_repo()) + self.addCleanup(repo_api.delete, repo.pulp_href) body = gen_deb_remote(policy=policy) - remote = self.client.post(DEB_REMOTE_PATH, body) - self.addCleanup(self.client.delete, remote["pulp_href"]) + remote = remote_api.create(body) + self.addCleanup(remote_api.delete, remote.pulp_href) # Sync the repository. - self.assertEqual(repo["latest_version_href"], f"{repo['pulp_href']}versions/0/") - sync(self.cfg, remote, repo) - repo = self.client.get(repo["pulp_href"]) - self.assertIsNotNone(repo["latest_version_href"]) + self.assertEqual(repo.latest_version_href, f"{repo.pulp_href}versions/0/") + repository_sync_data = RepositorySyncURL(remote=remote.pulp_href) + sync_response = repo_api.sync(repo.pulp_href, repository_sync_data) + monitor_task(sync_response.task) + repo = repo_api.read(repo.pulp_href) + self.assertEqual(repo.latest_version_href, f"{repo.pulp_href}versions/1/") # Assert that no HTTP error was raised. # Assert that the number of units present is according to the synced # feed. - content = self.client.get(DEB_PACKAGE_PATH) - self.assertEqual(len(content), DEB_FIXTURE_PACKAGE_COUNT, content) + content = packages_api.list() + self.assertEqual(content.count, DEB_FIXTURE_PACKAGE_COUNT, content) class SwitchDownloadPolicyTestCase(unittest.TestCase): @@ -180,37 +213,58 @@ class SwitchDownloadPolicyTestCase(unittest.TestCase): * `Pulp #4467 `_ """ - def test_all(self): - """Perform a lazy sync and change to immeditae to force download.""" + @classmethod + def setUpClass(cls): + """Create class-wide variables.""" + cls.DP_ON_DEMAND = "on_demand" in DOWNLOAD_POLICIES + cls.DP_STREAMED = "streamed" in DOWNLOAD_POLICIES + + @skip_if(bool, "DP_ON_DEMAND", False) + def test_on_demand(self): + """Test ``on_demand``. See :meth:`do_test`.""" + self.do_test("on_demand") + + @skip_if(bool, "DP_STREAMED", False) + def test_streamed(self): + """Test ``streamed``. See :meth:`do_test`.""" + self.do_test("streamed") + + def do_test(self, policy): + """Perform a lazy sync and change to immediate to force download.""" NON_LAZY_ARTIFACT_COUNT = 13 - cfg = config.get_config() # delete orphans to assure that no content units are present on the # file system - delete_orphans(cfg) - client = api.Client(cfg, api.page_handler) + delete_orphans() + repo_api = deb_repository_api + remote_api = deb_remote_api - repo = client.post(DEB_REPO_PATH, gen_repo()) - self.addCleanup(client.delete, repo["pulp_href"]) + repo = repo_api.create(gen_repo()) + self.addCleanup(repo_api.delete, repo.pulp_href) - body = gen_deb_remote(policy=choice(ON_DEMAND_DOWNLOAD_POLICIES)) - remote = client.post(DEB_REMOTE_PATH, body) - self.addCleanup(client.delete, remote["pulp_href"]) + body = gen_deb_remote(policy=policy) + remote = remote_api.create(body) + self.addCleanup(remote_api.delete, remote.pulp_href) # Sync the repository using a lazy download policy - sync(cfg, remote, repo) - artifacts = client.get(ARTIFACTS_PATH) - self.assertEqual(len(artifacts), NON_LAZY_ARTIFACT_COUNT, artifacts) + repository_sync_data = RepositorySyncURL(remote=remote.pulp_href) + sync_response = repo_api.sync(repo.pulp_href, repository_sync_data) + monitor_task(sync_response.task) + artifacts = artifact_api.list() + self.assertEqual(artifacts.count, NON_LAZY_ARTIFACT_COUNT, artifacts) # Update the policy to immediate - client.patch(remote["pulp_href"], {"policy": "immediate"}) - remote = client.get(remote["pulp_href"]) - self.assertEqual(remote["policy"], "immediate") + update_response = remote_api.partial_update(remote.pulp_href, {"policy": "immediate"}) + monitor_task(update_response.task) + remote = remote_api.read(remote.pulp_href) + self.assertEqual(remote.policy, "immediate") # Sync using immediate download policy - sync(cfg, remote, repo) + repository_sync_data = RepositorySyncURL(remote=remote.pulp_href) + sync_response = repo_api.sync(repo.pulp_href, repository_sync_data) + monitor_task(sync_response.task) # Assert that missing artifacts are downloaded - artifacts = client.get(ARTIFACTS_PATH) + artifacts = artifact_api.list() self.assertEqual( - len(artifacts), NON_LAZY_ARTIFACT_COUNT + DEB_FIXTURE_PACKAGE_COUNT, artifacts + artifacts.count, NON_LAZY_ARTIFACT_COUNT + DEB_FIXTURE_PACKAGE_COUNT, artifacts ) diff --git a/pulp_deb/tests/functional/api/test_publish.py b/pulp_deb/tests/functional/api/test_publish.py index a37fba462..c98dfe691 100644 --- a/pulp_deb/tests/functional/api/test_publish.py +++ b/pulp_deb/tests/functional/api/test_publish.py @@ -3,39 +3,47 @@ import unittest from random import choice -from requests.exceptions import HTTPError - -from pulp_smash import api, config -from pulp_smash.pulp3.utils import gen_repo, get_content, get_versions, modify_repo, sync +from pulp_smash import config +from pulp_smash.pulp3.utils import gen_repo, get_content, get_versions, modify_repo from pulp_deb.tests.functional.constants import ( DEB_GENERIC_CONTENT_NAME, DEB_PACKAGE_NAME, - DEB_PUBLICATION_PATH, - DEB_REMOTE_PATH, - DEB_REPO_PATH, - VERBATIM_PUBLICATION_PATH, ) +from pulp_deb.tests.functional.utils import set_up_module as setUpModule # noqa:F401 from pulp_deb.tests.functional.utils import ( - create_deb_publication, - create_verbatim_publication, gen_deb_remote, + monitor_task, + deb_apt_publication_api, + deb_remote_api, + deb_repository_api, + deb_verbatim_publication_api, ) -from pulp_deb.tests.functional.utils import set_up_module as setUpModule # noqa:F401 +from pulpcore.client.pulp_deb import ( + RepositorySyncURL, + DebDebPublication, + DebVerbatimPublication, +) +from pulpcore.client.pulp_deb.exceptions import ApiException -class PublishAnyRepoVersionTestCase(unittest.TestCase): - """Test whether a particular repository version can be published. + +class PublishAnyRepoVersionSimpleTestCase(unittest.TestCase): + """Test whether a particular repository version can be published simple. This test targets the following issues: * `Pulp #3324 `_ - * `Pulp Smash #897 `_ + * `Pulp Smash #897 `_ """ class Meta: - PUBLICATION_PATH = DEB_PUBLICATION_PATH - create_publication = create_deb_publication + publication_api = deb_apt_publication_api + + @staticmethod + def Publication(*args, **kwargs): + """Delegate Publication constructor.""" + return DebDebPublication(simple=True, *args, **kwargs) def test_all(self): """Test whether a particular repository version can be published. @@ -51,53 +59,102 @@ def test_all(self): repository versions to be published at same time. """ cfg = config.get_config() - client = api.Client(cfg, api.json_handler) + repo_api = deb_repository_api + remote_api = deb_remote_api + publication_api = self.Meta.publication_api body = gen_deb_remote() - remote = client.post(DEB_REMOTE_PATH, body) - self.addCleanup(client.delete, remote["pulp_href"]) + remote = remote_api.create(body) + self.addCleanup(remote_api.delete, remote.pulp_href) - repo = client.post(DEB_REPO_PATH, gen_repo()) - self.addCleanup(client.delete, repo["pulp_href"]) + repo = repo_api.create(gen_repo()) + self.addCleanup(repo_api.delete, repo.pulp_href) - sync(cfg, remote, repo) + repository_sync_data = RepositorySyncURL(remote=remote.pulp_href) + sync_response = repo_api.sync(repo.pulp_href, repository_sync_data) + monitor_task(sync_response.task) # Step 1 - repo = client.get(repo["pulp_href"]) - for deb_generic_content in get_content(repo)[DEB_GENERIC_CONTENT_NAME]: - modify_repo(cfg, repo, remove_units=[deb_generic_content]) - for deb_package in get_content(repo)[DEB_PACKAGE_NAME]: - modify_repo(cfg, repo, remove_units=[deb_package]) - version_hrefs = tuple(ver["pulp_href"] for ver in get_versions(repo)) + repo = repo_api.read(repo.pulp_href) + for deb_generic_content in get_content(repo.to_dict())[DEB_GENERIC_CONTENT_NAME]: + modify_repo(cfg, repo.to_dict(), remove_units=[deb_generic_content]) + for deb_package in get_content(repo.to_dict())[DEB_PACKAGE_NAME]: + modify_repo(cfg, repo.to_dict(), remove_units=[deb_package]) + version_hrefs = tuple(ver["pulp_href"] for ver in get_versions(repo.to_dict())) non_latest = choice(version_hrefs[:-1]) # Step 2 - publication = self.Meta.create_publication(cfg, repo) + publish_data = self.Meta.Publication(repository=repo.pulp_href) + publish_response = publication_api.create(publish_data) + created_resources = monitor_task(publish_response.task) + publication_href = created_resources[0] + self.addCleanup(publication_api.delete, publication_href) + publication = publication_api.read(publication_href) # Step 3 - self.assertEqual(publication["repository_version"], version_hrefs[-1]) + self.assertEqual(publication.repository_version, version_hrefs[-1]) # Step 4 - publication = self.Meta.create_publication(cfg, repo, non_latest) + publish_data = self.Meta.Publication(repository_version=non_latest) + publish_response = publication_api.create(publish_data) + created_resources = monitor_task(publish_response.task) + publication_href = created_resources[0] + publication = publication_api.read(publication_href) # Step 5 - self.assertEqual(publication["repository_version"], non_latest) + self.assertEqual(publication.repository_version, non_latest) # Step 6 - with self.assertRaises(HTTPError): - body = {"repository": repo["pulp_href"], "repository_version": non_latest} - client.post(self.Meta.PUBLICATION_PATH, body) + with self.assertRaises(ApiException): + body = {"repository": repo.pulp_href, "repository_version": non_latest} + publication_api.create(body) + + +class PublishAnyRepoVersionStructuredTestCase(PublishAnyRepoVersionSimpleTestCase): + """Test whether a particular repository version can be published structured. + + This test targets the following issues: + + * `Pulp #3324 `_ + * `Pulp Smash #897 `_ + """ + + class Meta: + publication_api = deb_apt_publication_api + + @staticmethod + def Publication(*args, **kwargs): + """Delegate Publication constructor.""" + return DebDebPublication(structured=True, *args, **kwargs) + + +class PublishAnyRepoVersionCombinedTestCase(PublishAnyRepoVersionSimpleTestCase): + """Test whether a particular repository version can be published both simple and structured. + + This test targets the following issues: + + * `Pulp #3324 `_ + * `Pulp Smash #897 `_ + """ + + class Meta: + publication_api = deb_apt_publication_api + + @staticmethod + def Publication(*args, **kwargs): + """Delegate Publication constructor.""" + return DebDebPublication(simple=True, structured=True, *args, **kwargs) -class VerbatimPublishAnyRepoVersionTestCase(PublishAnyRepoVersionTestCase): +class VerbatimPublishAnyRepoVersionTestCase(PublishAnyRepoVersionSimpleTestCase): """Test whether a particular repository version can be published verbatim. This test targets the following issues: * `Pulp #3324 `_ - * `Pulp Smash #897 `_ + * `Pulp Smash #897 `_ """ class Meta: - PUBLICATION_PATH = VERBATIM_PUBLICATION_PATH - create_publication = create_verbatim_publication + publication_api = deb_verbatim_publication_api + Publication = DebVerbatimPublication diff --git a/pulp_deb/tests/functional/api/test_sync.py b/pulp_deb/tests/functional/api/test_sync.py index 57e82d3ec..8fdb674e3 100644 --- a/pulp_deb/tests/functional/api/test_sync.py +++ b/pulp_deb/tests/functional/api/test_sync.py @@ -2,28 +2,34 @@ """Tests that sync deb plugin repositories.""" import unittest -from pulp_smash import api, cli, config, exceptions +from pulp_smash import cli, config from pulp_smash.pulp3.constants import MEDIA_PATH -from pulp_smash.pulp3.utils import gen_repo, get_content_summary, get_added_content_summary, sync +from pulp_smash.pulp3.utils import gen_repo, get_added_content_summary, get_content_summary from pulp_deb.tests.functional.constants import ( DEB_FIXTURE_SUMMARY, DEB_FULL_FIXTURE_SUMMARY, - DEB_REMOTE_PATH, - DEB_REPO_PATH, + DEB_INVALID_FIXTURE_URL, ) -from pulp_deb.tests.functional.utils import gen_deb_remote from pulp_deb.tests.functional.utils import set_up_module as setUpModule # noqa:F401 +from pulp_deb.tests.functional.utils import ( + gen_deb_remote, + monitor_task, + PulpTaskError, + deb_remote_api, + deb_repository_api, +) + +from pulpcore.client.pulp_deb import RepositorySyncURL class BasicSyncTestCase(unittest.TestCase): - """Sync repositories with the deb plugin.""" + """Sync a repository with the deb plugin.""" @classmethod def setUpClass(cls): """Create class-wide variables.""" cls.cfg = config.get_config() - cls.client = api.Client(cls.cfg, api.json_handler) def test_sync_small(self): """Test synching with deb content only.""" @@ -46,40 +52,53 @@ def do_sync(self, sync_udebs, fixture_summary): 2. Assert that repository version is None. 3. Sync the remote. 4. Assert that repository version is not None. - 5. Sync the remote one more time. - 6. Assert that repository version is the same as the previous one. + 5. Assert that the correct number of units were added and are present + in the repo. + 6. Sync the remote one more time. + 7. Assert that repository version is the same as the previous one. + 8. Assert that the same number of content units are present and that no + units were added. """ - repo = self.client.post(DEB_REPO_PATH, gen_repo()) - self.addCleanup(self.client.delete, repo["pulp_href"]) + repo_api = deb_repository_api + remote_api = deb_remote_api + + repo = repo_api.create(gen_repo()) + self.addCleanup(repo_api.delete, repo.pulp_href) body = gen_deb_remote(sync_udebs=sync_udebs) - remote = self.client.post(DEB_REMOTE_PATH, body) - self.addCleanup(self.client.delete, remote["pulp_href"]) + remote = remote_api.create(body) + self.addCleanup(remote_api.delete, remote.pulp_href) # Sync the repository. - self.assertEqual(repo["latest_version_href"], f"{repo['pulp_href']}versions/0/") - sync(self.cfg, remote, repo) - repo = self.client.get(repo["pulp_href"]) + self.assertEqual(repo.latest_version_href, f"{repo.pulp_href}versions/0/") + repository_sync_data = RepositorySyncURL(remote=remote.pulp_href) + sync_response = repo_api.sync(repo.pulp_href, repository_sync_data) + monitor_task(sync_response.task) + repo = repo_api.read(repo.pulp_href) - self.assertIsNotNone(repo["latest_version_href"]) - self.assertDictEqual(get_content_summary(repo), fixture_summary) - self.assertDictEqual(get_added_content_summary(repo), fixture_summary) + self.assertIsNotNone(repo.latest_version_href) + self.assertDictEqual(get_content_summary(repo.to_dict()), fixture_summary) + self.assertDictEqual(get_added_content_summary(repo.to_dict()), fixture_summary) # Sync the repository again. - latest_version_href = repo["latest_version_href"] - sync(self.cfg, remote, repo) - repo = self.client.get(repo["pulp_href"]) + latest_version_href = repo.latest_version_href + repository_sync_data = RepositorySyncURL(remote=remote.pulp_href) + sync_response = repo_api.sync(repo.pulp_href, repository_sync_data) + monitor_task(sync_response.task) + repo = repo_api.read(repo.pulp_href) - self.assertEqual(latest_version_href, repo["latest_version_href"]) - self.assertDictEqual(get_content_summary(repo), fixture_summary) + self.assertEqual(latest_version_href, repo.latest_version_href) + self.assertDictEqual(get_content_summary(repo.to_dict()), fixture_summary) def test_file_decriptors(self): """Test whether file descriptors are closed properly. This test targets the following issue: + `Pulp #4073 `_ Do the following: + 1. Check if 'lsof' is installed. If it is not, skip this test. 2. Create and sync a repo. 3. Run the 'lsof' command to verify that files in the @@ -87,43 +106,67 @@ def test_file_decriptors(self): 4. Assert that issued command returns `0` opened files. """ cli_client = cli.Client(self.cfg, cli.echo_handler) + repo_api = deb_repository_api + remote_api = deb_remote_api # check if 'lsof' is available if cli_client.run(("which", "lsof")).returncode != 0: raise unittest.SkipTest("lsof package is not present") - repo = self.client.post(DEB_REPO_PATH, gen_repo()) - self.addCleanup(self.client.delete, repo["pulp_href"]) + repo = repo_api.create(gen_repo()) + self.addCleanup(repo_api.delete, repo.pulp_href) - remote = self.client.post(DEB_REMOTE_PATH, gen_deb_remote()) - self.addCleanup(self.client.delete, remote["pulp_href"]) + remote = remote_api.create(gen_deb_remote()) + self.addCleanup(remote_api.delete, remote.pulp_href) - sync(self.cfg, remote, repo) + repository_sync_data = RepositorySyncURL(remote=remote.pulp_href) + sync_response = repo_api.sync(repo.pulp_href, repository_sync_data) + monitor_task(sync_response.task) cmd = "lsof -t +D {}".format(MEDIA_PATH).split() response = cli_client.run(cmd).stdout self.assertEqual(len(response), 0, response) -class SyncInvalidURLTestCase(unittest.TestCase): - """Sync a repository with an invalid url on the Remote.""" +class SyncInvalidTestCase(unittest.TestCase): + """Sync a repository with a given url on the remote.""" - def test_all(self): - """ - Sync a repository using a Remote url that does not exist. - - Test that we get a task failure. + def test_invalid_url(self): + """Sync a repository using a remote url that does not exist. + Test that we get a task failure. See :meth:`do_test`. """ - cfg = config.get_config() - client = api.Client(cfg, api.json_handler) - - repo = client.post(DEB_REPO_PATH, gen_repo()) - self.addCleanup(client.delete, repo["pulp_href"]) - - body = gen_deb_remote(url="http://i-am-an-invalid-url.com/invalid/") - remote = client.post(DEB_REMOTE_PATH, body) - self.addCleanup(client.delete, remote["pulp_href"]) - - with self.assertRaises(exceptions.TaskReportError): - sync(cfg, remote, repo) + with self.assertRaises(PulpTaskError) as exc: + self.do_test("http://i-am-an-invalid-url.com/invalid/") + error = exc.exception.task.error + self.assertIsNotNone(error["description"]) + + # Provide an invalid repository and specify keywords in the anticipated error message + @unittest.skip("FIXME: Plugin writer action required.") + def test_invalid_deb_content(self): + """Sync a repository using an invalid plugin_content repository. + + Assert that an exception is raised, and that error message has + keywords related to the reason of the failure. See :meth:`do_test`. + """ + with self.assertRaises(PulpTaskError) as exc: + self.do_test(DEB_INVALID_FIXTURE_URL) + error = exc.exception.task.error + for key in ("mismached", "empty"): + self.assertIn(key, error["description"]) + + def do_test(self, url): + """Sync a repository given ``url`` on the remote.""" + repo_api = deb_repository_api + remote_api = deb_remote_api + + repo = repo_api.create(gen_repo()) + self.addCleanup(repo_api.delete, repo.pulp_href) + + body = gen_deb_remote(url=url) + remote = remote_api.create(body) + self.addCleanup(remote_api.delete, remote.pulp_href) + + repository_sync_data = RepositorySyncURL(remote=remote.pulp_href) + sync_response = repo_api.sync(repo.pulp_href, repository_sync_data) + return monitor_task(sync_response.task) diff --git a/pulp_deb/tests/functional/api/test_upload.py b/pulp_deb/tests/functional/api/test_upload.py deleted file mode 100644 index 32f8a029f..000000000 --- a/pulp_deb/tests/functional/api/test_upload.py +++ /dev/null @@ -1,107 +0,0 @@ -# coding=utf-8 -"""Tests that verify upload of content to Pulp.""" -import hashlib -import unittest - -from pulp_smash import api, config, utils -from pulp_smash.pulp3.constants import ARTIFACTS_PATH -from pulp_smash.pulp3.utils import delete_orphans, gen_repo, sync - -from pulp_deb.tests.functional.constants import ( - DEB_PACKAGE_PATH, - DEB_PACKAGE_RELPATH, - DEB_PACKAGE_URL, - DEB_REMOTE_PATH, - DEB_REPO_PATH, -) -from pulp_deb.tests.functional.utils import gen_deb_remote -from pulp_deb.tests.functional.utils import set_up_module as setUpModule # noqa:F401 - - -class SingleRequestUploadTestCase(unittest.TestCase): - """Test whether one can upload a RPM using a single request. - - This test targets the following issues: - - `Pulp #4087 `_ - `Pulp #4285 `_ - """ - - @classmethod - def setUpClass(cls): - """Create class-wide variables.""" - cls.cfg = config.get_config() - cls.client = api.Client(cls.cfg, api.page_handler) - cls.file = {"file": utils.http_get(DEB_PACKAGE_URL)} - - @classmethod - def tearDownClass(cls): - """Clean up after all tests ran.""" - delete_orphans(cls.cfg) - - def setUp(self): - """Perform per test preparation.""" - delete_orphans(self.cfg) - - def single_request_upload(self, relative_path=None, repo=None): - """Create single request upload.""" - data = {} - if relative_path: - data["relative_path"] = relative_path - if repo: - data["repository"] = repo["pulp_href"] - return self.client.post(DEB_PACKAGE_PATH, files=self.file, data=data) - - def test_single_request_upload(self): - """Test single request upload.""" - repo = self.client.post(DEB_REPO_PATH, gen_repo()) - self.addCleanup(self.client.delete, repo["pulp_href"]) - self.single_request_upload(repo=repo) - repo = self.client.get(repo["pulp_href"]) - - # Assertion about repo version. - self.assertIsNotNone(repo["latest_version_href"], repo) - - # Assertions about artifacts. - artifact = self.client.get(ARTIFACTS_PATH) - self.assertEqual(len(artifact), 1, artifact) - self.assertEqual( - artifact[0]["sha256"], hashlib.sha256(self.file["file"]).hexdigest(), artifact - ) - - # Assertion about content unit. - content = self.client.get(DEB_PACKAGE_PATH) - self.assertEqual(len(content), 1, content) - - def test_duplicate_unit(self): - """Test single request upload for unit already present in Pulp.""" - repo = self.client.post(DEB_REPO_PATH, gen_repo()) - self.addCleanup(self.client.delete, repo["pulp_href"]) - # Hint this relative path is different from the one from a simple upload, because it - # contains the release - result = self.single_request_upload(relative_path=DEB_PACKAGE_RELPATH) - # Check for a content_unit in created_resources - task = self.client.get(result["task"]) - self.assertEqual(len(task["created_resources"]), 1) - self.assertRegex(task["created_resources"][0], r"^/pulp/api/v3/content/deb/packages/") - result = self.single_request_upload(repo=repo) - # check for a repository_version in created_resources - task = self.client.get(result["task"]) - self.assertEqual(len(task["created_resources"]), 2) - self.assertRegex( - " ".join(task["created_resources"]), r"/pulp/api/v3/repositories/.*/versions/" - ) - - def test_sync_interference(self): - """Test that uploading a file does not break a consecutive sync containing that file.""" - upload_repo = self.client.post(DEB_REPO_PATH, gen_repo()) - self.addCleanup(self.client.delete, upload_repo["pulp_href"]) - sync_repo = self.client.post(DEB_REPO_PATH, gen_repo()) - self.addCleanup(self.client.delete, sync_repo["pulp_href"]) - remote = self.client.post(DEB_REMOTE_PATH, gen_deb_remote()) - self.addCleanup(self.client.delete, remote["pulp_href"]) - - # upload a file into one repository - self.single_request_upload(repo=upload_repo) - # sync the other repository - sync(self.cfg, remote, sync_repo) diff --git a/pulp_deb/tests/functional/constants.py b/pulp_deb/tests/functional/constants.py index 724072b03..6eb849b6d 100644 --- a/pulp_deb/tests/functional/constants.py +++ b/pulp_deb/tests/functional/constants.py @@ -90,6 +90,9 @@ def _clean_dict(d): DEB_GENERIC_CONTENT_RELPATH = "dists/ragnarok/asgard/binary-armeb/Release" DEB_GENERIC_CONTENT_URL = urljoin(DEB_FIXTURE_URL, DEB_GENERIC_CONTENT_RELPATH) +# FIXME: replace this with your own fixture repository URL and metadata +DEB_INVALID_FIXTURE_URL = urljoin(PULP_FIXTURES_BASE_URL, "deb_invalid/") + # FIXME: replace this with your own fixture repository URL and metadata DEB_LARGE_FIXTURE_URL = urljoin(PULP_FIXTURES_BASE_URL, "deb_large/") diff --git a/pulp_deb/tests/functional/utils.py b/pulp_deb/tests/functional/utils.py index 0748d8f12..fa101982f 100644 --- a/pulp_deb/tests/functional/utils.py +++ b/pulp_deb/tests/functional/utils.py @@ -3,8 +3,10 @@ import os from functools import partial from unittest import SkipTest +from time import sleep +from tempfile import NamedTemporaryFile -from pulp_smash import api, selectors +from pulp_smash import api, selectors, utils from pulp_smash.pulp3.utils import ( gen_remote, gen_repo, @@ -31,6 +33,48 @@ VERBATIM_PUBLICATION_PATH, ) +from pulpcore.client.pulpcore import ( + ApiClient as CoreApiClient, + ArtifactsApi, + Configuration, + TasksApi, +) +from pulpcore.client.pulp_deb import ( + ApiClient as DebApiClient, + ContentGenericContentsApi, + ContentPackagesApi, + DistributionsAptApi, + PublicationsAptApi, + PublicationsVerbatimApi, + RemotesAptApi, + RepositoriesAptApi, +) + +skip_if = partial(selectors.skip_if, exc=SkipTest) # pylint:disable=invalid-name +"""The ``@skip_if`` decorator, customized for unittest. + +:func:`pulp_smash.selectors.skip_if` is test runner agnostic. This function is +identical, except that ``exc`` has been set to ``unittest.SkipTest``. +""" + +configuration = Configuration() +configuration.username = "admin" +configuration.password = "password" +configuration.safe_chars_for_path_param = "/" + +core_client = CoreApiClient(configuration) +artifact_api = ArtifactsApi(core_client) +task_api = TasksApi(core_client) + +deb_client = DebApiClient(configuration) +deb_generic_content_api = ContentGenericContentsApi(deb_client) +deb_package_api = ContentPackagesApi(deb_client) +deb_remote_api = RemotesAptApi(deb_client) +deb_repository_api = RepositoriesAptApi(deb_client) +deb_apt_publication_api = PublicationsAptApi(deb_client) +deb_verbatim_publication_api = PublicationsVerbatimApi(deb_client) +deb_distribution_api = DistributionsAptApi(deb_client) + def set_up_module(): """Skip tests Pulp 3 isn't under test or if pulp_deb isn't installed.""" @@ -38,6 +82,11 @@ def set_up_module(): require_pulp_plugins({"pulp_deb"}, SkipTest) +def gen_deb_client(): + """Return an OBJECT for deb client.""" + return deb_client + + def gen_deb_remote( url=DEB_FIXTURE_URL, distributions=DEB_FIXTURE_RELEASE, sync_udebs=False, **kwargs ): @@ -72,7 +121,7 @@ def _rel_path(package, component=""): "{}_{}_{}.deb".format(package["package"], package["version"], package["architecture"]), ) - content = get_content(repo, version_href) + content = get_content(repo.to_dict(), version_href) result = { DEB_PACKAGE_NAME: [ (content_unit["relative_path"], _rel_path(content_unit, "all")) @@ -108,15 +157,15 @@ def get_deb_verbatim_content_unit_paths(repo, version_href=None): return { DEB_RELEASE_FILE_NAME: [ (content_unit["relative_path"], content_unit["relative_path"]) - for content_unit in get_content(repo, version_href)[DEB_RELEASE_FILE_NAME] + for content_unit in get_content(repo.to_dict(), version_href)[DEB_RELEASE_FILE_NAME] ], DEB_PACKAGE_NAME: [ (content_unit["relative_path"], content_unit["relative_path"]) - for content_unit in get_content(repo, version_href)[DEB_PACKAGE_NAME] + for content_unit in get_content(repo.to_dict(), version_href)[DEB_PACKAGE_NAME] ], DEB_GENERIC_CONTENT_NAME: [ (content_unit["relative_path"], content_unit["relative_path"]) - for content_unit in get_content(repo, version_href)[DEB_GENERIC_CONTENT_NAME] + for content_unit in get_content(repo.to_dict(), version_href)[DEB_GENERIC_CONTENT_NAME] ], } @@ -222,9 +271,45 @@ def create_verbatim_publication(cfg, repo, version_href=None): return client.get(tasks[-1]["created_resources"][0]) -skip_if = partial(selectors.skip_if, exc=SkipTest) -"""The ``@skip_if`` decorator, customized for unittest. +def gen_artifact(url): + """Creates an artifact.""" + response = utils.http_get(url) + with NamedTemporaryFile() as temp_file: + temp_file.write(response) + temp_file.flush() + artifact = ArtifactsApi(core_client).create(file=temp_file.name) + return artifact.to_dict() -:func:`pulp_smash.selectors.skip_if` is test runner agnostic. This function is -identical, except that ``exc`` has been set to ``unittest.SkipTest``. -""" + +class PulpTaskError(Exception): + """Exception to describe task errors.""" + + def __init__(self, task): + """Provide task info to exception.""" + description = task.error["description"] + super().__init__(self, f"Pulp task failed ({description})") + self.task = task + + +def monitor_task(task_href): + """Polls the Task API until the task is in a completed state. + + Prints the task details and a success or failure message. Exits on failure. + + Args: + task_href(str): The href of the task to monitor + + Returns: + list[str]: List of hrefs that identify resource created by the task + + """ + completed = ["completed", "failed", "canceled"] + task = task_api.read(task_href) + while task.state not in completed: + sleep(2) + task = task_api.read(task_href) + + if task.state == "completed": + return task.created_resources + + raise PulpTaskError(task=task) diff --git a/pulp_deb/tests/unit/test_serializers.py b/pulp_deb/tests/unit/test_serializers.py new file mode 100644 index 000000000..86cade47f --- /dev/null +++ b/pulp_deb/tests/unit/test_serializers.py @@ -0,0 +1,40 @@ +import unittest +from django.test import TestCase + +from pulp_deb.app.serializers import GenericContentSerializer +from pulp_deb.app.models import GenericContent + +from pulpcore.plugin.models import Artifact + + +# Fill data with sufficient information to create DebContent +# Provide sufficient parameters to create the DebContent object +# Depending on the base class of the serializer, provide either "_artifact" or "_artifacts" +@unittest.skip("FIXME: plugin writer action required") +class TestGenericContentSerializer(TestCase): + """Test GenericContentSerializer.""" + + def setUp(self): + """Set up the GenericContentSerializer tests.""" + self.artifact = Artifact.objects.create( + md5="ec0df26316b1deb465d2d18af7b600f5", + sha1="cf6121b0425c2f2e3a2fcfe6f402d59730eb5661", + sha224="9a6297eb28d91fad5277c0833856031d0e940432ad807658bd2b60f4", + sha256="c8ddb3dcf8da48278d57b0b94486832c66a8835316ccf7ca39e143cbfeb9184f", + sha384="53a8a0cebcb7780ed7624790c9d9a4d09ba74b47270d397f5ed7bc1c46777a0fbe362aaf2bbe7f0966a350a12d76e28d", # noqa + sha512="a94a65f19b864d184a2a5e07fa29766f08c6d49b6f624b3dd3a36a98267b9137d9c35040b3e105448a869c23c2aec04c9e064e3555295c1b8de6515eed4da27d", # noqa + size=1024, + ) + + def test_valid_data(self): + """Test that the GenericContentSerializer accepts valid data.""" + data = {"_artifact": "/pulp/api/v3/artifacts/{}/".format(self.artifact.pk)} + serializer = GenericContentSerializer(data=data) + self.assertTrue(serializer.is_valid()) + + def test_duplicate_data(self): + """Test that the GenericContentSerializer does not accept data.""" + GenericContent.objects.create(artifact=self.artifact) + data = {"_artifact": "/pulp/api/v3/artifacts/{}/".format(self.artifact.pk)} + serializer = GenericContentSerializer(data=data) + self.assertFalse(serializer.is_valid())