Assert PRN support to advanced copy API

Fixes: #3853

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": <RepositoryVersion.[pulp_href|prn]>),
+            "dest_repo": <RpmRepository.[pulp_href|prn]>),
+            "content": [<Content.[pulp_href|prn]>, <Content.[pulp_href|prn]>]
+          }
+        ]
+        ```
+
+        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

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)
 

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."""