From f0697eea3de3383634d069e9dbd4a98246d431b8 Mon Sep 17 00:00:00 2001 From: Pedro Brochado Date: Tue, 21 Jan 2025 18:25:58 -0300 Subject: [PATCH] Assert PRN support to advanced copy API Fixes: #3853 --- pulp_rpm/app/serializers/repository.py | 27 ++++++++++++++- pulp_rpm/tests/functional/api/test_copy.py | 38 ++++++++++++++++++---- pulp_rpm/tests/functional/api/test_prn.py | 7 ---- 3 files changed, 58 insertions(+), 14 deletions(-) diff --git a/pulp_rpm/app/serializers/repository.py b/pulp_rpm/app/serializers/repository.py index 2dc29757d..246fc54ca 100644 --- a/pulp_rpm/app/serializers/repository.py +++ b/pulp_rpm/app/serializers/repository.py @@ -37,6 +37,7 @@ ) from pulp_rpm.app.schema import COPY_CONFIG_SCHEMA from urllib.parse import urlparse +from textwrap import dedent class RpmRepositorySerializer(RepositorySerializer): @@ -543,7 +544,27 @@ class CopySerializer(ValidateFieldsMixin, serializers.Serializer): """ config = CustomJSONField( - help_text=_("A JSON document describing sources, destinations, and content to be copied"), + help_text=_( + dedent( + """\ + A JSON document describing sources, destinations, and content to be copied. + + Its a list of dictionaries with the following form: + + ```json + [ + { + "source_repo_version": ), + "dest_repo": ), + "content": [, ] + } + ] + ``` + + If domains are enabled, the refered pulp object must be part of the current domain. + """ + ) + ), ) dependency_solving = serializers.BooleanField( @@ -559,6 +580,10 @@ def validate(self, data): """ def check_domain(domain, href, name): + # PRNs dont have domain info. + # Has to be checked on the task context + if href.startswith("prn:"): + return # We're doing just a string-check here rather than checking objects # because there can be A LOT of objects, and this is happening in the view-layer # where we have strictly-limited timescales to work with diff --git a/pulp_rpm/tests/functional/api/test_copy.py b/pulp_rpm/tests/functional/api/test_copy.py index 040bd7caa..44b9e46ec 100644 --- a/pulp_rpm/tests/functional/api/test_copy.py +++ b/pulp_rpm/tests/functional/api/test_copy.py @@ -18,8 +18,26 @@ from pulpcore.client.pulp_rpm import Copy from pulpcore.client.pulp_rpm.exceptions import ApiException +import subprocess +def noop(uri): + return uri + + +def get_prn(uri): + """Utility to get prn without having to setup django. + TODO: This is a Copy-paste from pulpcore. Make it public there. + """ + commands = f"from pulpcore.app.util import get_prn; print(get_prn(uri='{uri}'));" + process = subprocess.run(["pulpcore-manager", "shell", "-c", commands], capture_output=True) + + assert process.returncode == 0 + prn = process.stdout.decode().strip() + return prn + + +@pytest.mark.parametrize("get_id", [noop, get_prn], ids=["without-prn", "with-prn"]) @pytest.mark.parallel def test_modular_static_context_copy( init_and_sync, @@ -28,13 +46,19 @@ def test_modular_static_context_copy( rpm_modulemd_api, rpm_repository_factory, rpm_repository_api, + get_id, ): """Test copying a static_context-using repo to an empty destination.""" src, _ = init_and_sync(url=RPM_MODULES_STATIC_CONTEXT_FIXTURE_URL) dest = rpm_repository_factory() data = Copy( - config=[{"source_repo_version": src.latest_version_href, "dest_repo": dest.pulp_href}], + config=[ + { + "source_repo_version": get_id(src.latest_version_href), + "dest_repo": get_id(dest.pulp_href), + } + ], dependency_solving=False, ) monitor_task(rpm_copy_api.copy_content(data).task) @@ -44,7 +68,7 @@ def test_modular_static_context_copy( assert get_content_summary(dest.to_dict()) == RPM_MODULAR_STATIC_FIXTURE_SUMMARY assert get_added_content_summary(dest.to_dict()) == RPM_MODULAR_STATIC_FIXTURE_SUMMARY - modules = rpm_modulemd_api.list(repository_version=dest.latest_version_href).results + modules = rpm_modulemd_api.list(repository_version=get_id(dest.latest_version_href)).results module_static_contexts = [ (module.name, module.version) for module in modules if module.static_context ] @@ -141,6 +165,7 @@ def test_invalid_config( ) rpm_copy_api.copy_content(data) + @pytest.mark.parametrize("get_id", [noop, get_prn], ids=["without-prn", "with-prn"]) def test_content( self, monitor_task, @@ -149,20 +174,21 @@ def test_content( rpm_repository_api, rpm_repository_factory, rpm_unsigned_repo_immediate, + get_id, ): """Test the content parameter.""" src = rpm_unsigned_repo_immediate content = rpm_advisory_api.list(repository_version=src.latest_version_href).results - content_to_copy = (content[0].pulp_href, content[1].pulp_href) + content_to_copy = (get_id(content[0].pulp_href), get_id(content[1].pulp_href)) dest = rpm_repository_factory() data = Copy( config=[ { - "source_repo_version": src.latest_version_href, - "dest_repo": dest.pulp_href, + "source_repo_version": get_id(src.latest_version_href), + "dest_repo": get_id(dest.pulp_href), "content": content_to_copy, } ], @@ -172,7 +198,7 @@ def test_content( dest = rpm_repository_api.read(dest.pulp_href) dc = rpm_advisory_api.list(repository_version=dest.latest_version_href) - dest_content = [c.pulp_href for c in dc.results] + dest_content = [get_id(c.pulp_href) for c in dc.results] assert sorted(content_to_copy) == sorted(dest_content) diff --git a/pulp_rpm/tests/functional/api/test_prn.py b/pulp_rpm/tests/functional/api/test_prn.py index e117f45f5..ddbaf0941 100644 --- a/pulp_rpm/tests/functional/api/test_prn.py +++ b/pulp_rpm/tests/functional/api/test_prn.py @@ -1,13 +1,6 @@ import pytest -import requests - -# @pytest.fixture(scope="session") -# def pulp_openapi_schema_rpm(pulp_api_v3_url): -# COMPONENT="rpm" -# return requests.get(f"{pulp_api_v3_url}docs/api.json?bindings&component={COMPONENT}").json() - @pytest.mark.parallel def test_prn_schema(pulp_openapi_schema): """Test that PRN is a part of every serializer with a pulp_href."""