Assert PRN support to advanced copy API

* Add cross_domain validation for PRNs on copy API serializer. The new
  validation doesn't capture the first href/prn that failed anymore,
  as it relies on querying distinct domains in the bulk of references.

Fixes: pulp#3853

Author: pedro-psb

CHANGES/3853.bugfix

@@ -0,0 +1 @@
+Extended PRN support to Advanced Copy API and DistributionTree.

pulp_rpm/app/serializers/repository.py

@@ -3,7 +3,7 @@
 
 from django.conf import settings
 from jsonschema import Draft7Validator
-from pulpcore.plugin.models import AsciiArmoredDetachedSigningService, Publication, Remote
+from pulpcore.plugin.models import AsciiArmoredDetachedSigningService, Publication, Remote, Content
 from pulpcore.plugin.serializers import (
     DetailRelatedField,
     DistributionSerializer,
@@ -14,7 +14,7 @@
     RepositorySyncURLSerializer,
     ValidateFieldsMixin,
 )
-from pulpcore.plugin.util import get_domain
+from pulpcore.plugin.util import get_domain, extract_pk
 from rest_framework import serializers
 from pulp_rpm.app.fields import CustomJSONField
 
@@ -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,32 @@ 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 available fields:
+
+        ```json
+        [
+          {
+            "source_repo_version": <RepositoryVersion [pulp_href|prn]>,
+            "dest_repo": <RpmRepository [pulp_href|prn]>,
+            "dest_base_version": <int>,
+            "content": [<Content [pulp_href|prn]>, ...]
+          },
+          ...
+        ]
+        ```
+
+        If domains are enabled, the refered pulp objects must be part of the current domain.
+
+        For usage examples, refer to the advanced copy guide:
+        <https://pulpproject.org/pulp_rpm/docs/user/guides/modify/#advanced-copy-workflow>
+        """
+            )
+        ),
     )
 
     dependency_solving = serializers.BooleanField(
@@ -558,14 +584,24 @@ def validate(self, data):
         Check for cross-domain references (if domain-enabled).
         """
 
-        def check_domain(domain, href, name):
-            # 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
+        def _extract_pk(ref):
+            """Workaround for expensive extract_pk(href) calls."""
+            if ref.starswith("prn:"):
+                return extract_pk(ref)
+            else:
+                # This assumes ref is a non-nested href. E.g:
+                # /pulp/api/v3/content/rpm/packagegroups/0194a901-a9a6-705a-9436-ba6e19c98f40/
+                uuid = urlparse(ref).path.strip("/").split("/")[-1]
+                if len(uuid) < 32:
+                    raise serializers.ValidationError(
+                        _("The href path should end with a uuid pk: {}".format(ref))
+                    )
+                return uuid
+
+        def check_domain_ok(domain, href) -> bool:
             if href and domain not in href:
-                raise serializers.ValidationError(
-                    _("{} must be a part of the {} domain.").format(name, domain)
-                )
+                return False
+            return True
 
         def check_cross_domain_config(cfg):
             """Check that all config-elts are in 'our' domain."""
@@ -574,13 +610,25 @@ def check_cross_domain_config(cfg):
             # Insure curr-domain exists in src/dest/dest_base_version/content-list hrefs
             curr_domain_name = get_domain().name
             for entry in cfg:
-                check_domain(curr_domain_name, entry["source_repo_version"], "dest_repo")
-                check_domain(curr_domain_name, entry["dest_repo"], "dest_repo")
-                check_domain(
-                    curr_domain_name, entry.get("dest_base_version", None), "dest_base_version"
-                )
-                for content_href in entry.get("content", []):
-                    check_domain(curr_domain_name, content_href, "content")
+                domain_ok = True
+                ref_list = [entry["dest_version"]]
+
+                # Special case for source_repo_version, because the general method using extract_pk
+                # can't be used for nested hrefs (the pk is not in the href).
+                if (source_rv := entry["source_repo_version"]).startswith("prn:"):
+                    ref_list.append(source_rv)
+                else:
+                    domain_ok = domain_ok and check_domain_ok(curr_domain_name, source_rv)
+
+                # Check if all references belong to the correct domain without individual calls
+                ref_list.extend(entry.get("content", []))
+                pks_list = (_extract_pk(v) for v in ref_list)
+                distinct = Content.objects.filter(pk__in=pks_list).values("pulp_domain").distinct()
+                domain_ok = domain_ok and (len(distinct) == 1 and distinct.name == curr_domain_name)
+                if not domain_ok:
+                    raise serializers.ValidationError(
+                        _("All href/prns must be a part of the {} domain.").format(curr_domain_name)
+                    )
 
         super().validate(data)
         if "config" in data:

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