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: #3853

CHANGES/3853.bugfix

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

pulp_rpm/app/serializers/repository.py

@@ -3,7 +3,13 @@
 
 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,
+    RepositoryVersion,
+)
 from pulpcore.plugin.serializers import (
     DetailRelatedField,
     DistributionSerializer,
@@ -37,6 +43,7 @@
 )
 from pulp_rpm.app.schema import COPY_CONFIG_SCHEMA
 from urllib.parse import urlparse
+from textwrap import dedent
 
 
 class RpmRepositorySerializer(RepositorySerializer):
@@ -543,7 +550,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,29 +590,83 @@ 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
-            if href and domain not in href:
+        def raise_validation(field, domain, id=""):
+            id = f"\n{id}" if id else ""
+            raise serializers.ValidationError(
+                _("The field {} contains object(s) not in {} domain.{}".format(field, domain, id))
+            )
+
+        def parse_reference(ref) -> tuple[str, str, bool]:
+            """Extract info from prn/href to enable checking domains.
+
+            This is used for:
+            1. In case of HREFS, avoid expensive extract_pk(href) to get pks.
+            2. HREF and PRNs have different information hardcoded available.
+               E.g: RepositoryVerseion HREF has its Repository pk; PRNs have the RepoVer pk.
+
+            Returns a tuple with (pk, class_name, is_prn)
+            """
+            if ref.startswith("prn:"):
+                ref_class, pk = resolve_prn(ref)
+                return (pk, ref_class, True)
+            # content:    ${BASE}/content/rpm/packages/${UUID}/
+            # repository: ${BASE}/repositories/rpm/rpm/${UUID}/
+            # repover:    ${BASE}/repositories/rpm/rpm/${UUID}/versions/0/
+            url = urlparse(ref).path.strip("/").split("/")
+            ref_class = RpmRepository if "/repositories/" in ref else Content
+            is_repover_href = url[-1].isdigit() and url[-2] == "versions"
+            uuid = url[-3] if is_repover_href else url[-1]
+            if len(uuid) < 32:
                 raise serializers.ValidationError(
-                    _("{} must be a part of the {} domain.").format(name, domain)
+                    _("The href path should end with a uuid pk: {}".format(ref))
                 )
+            return (uuid, ref_class, False)
+
+        def check_domain(entry, name, curr_domain):
+            href_or_prn = entry[name]
+            resource_pk, ref_class, is_prn = parse_reference(href_or_prn)
+            try:
+                if ref_class is RepositoryVersion and is_prn:
+                    resource_domain_pk = (
+                        RepositoryVersion.objects.select_related("repository")
+                        .values("repository__pulp_domain")
+                        .get(pk=resource_pk)["repository__pulp_domain"]
+                    )
+                else:
+                    resource_domain_pk = RpmRepository.objects.values("pulp_domain").get(
+                        pk=resource_pk
+                    )["pulp_domain"]
+            except RepositoryVersion.DoesNotExit as e:
+                raise serializers.ValidationError from e
+            except RpmRepository.DoesNotExit as e:
+                raise serializers.ValidationError from e
+
+            if resource_domain_pk != curr_domain.pk:
+                raise_validation(name, curr_domain.name, resource_domain_pk)
 
         def check_cross_domain_config(cfg):
             """Check that all config-elts are in 'our' domain."""
             # copy-cfg is a list of dictionaries.
             # source_repo_version and dest_repo are required fields.
             # Insure curr-domain exists in src/dest/dest_base_version/content-list hrefs
-            curr_domain_name = get_domain().name
+            curr_domain = get_domain()
             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")
+                # Check required fields individually
+                check_domain(entry, "source_repo_version", curr_domain)
+                check_domain(entry, "dest_repo", curr_domain)
+
+                # Check content generically to avoid timeout of multiple calls
+                content_list = entry.get("content", None)
+                if content_list:
+                    content_list = [parse_reference(v)[0] for v in content_list]
+                    distinct = (
+                        Content.objects.filter(pk__in=content_list).values("pulp_domain").distinct()
+                    )
+                    domain_ok = (
+                        len(distinct) == 1 and distinct.first()["pulp_domain"] == curr_domain.pk
+                    )
+                    if not domain_ok:
+                        raise_validation("content", 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."""