[PULP-276] Add support for prns

CHANGES/3853.bugfix

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

pulp_rpm/app/serializers/distribution.py

@@ -150,6 +150,7 @@ class Meta:
         model = DistributionTree
         fields = (
             "pulp_href",
+            "prn",
             "header_version",
             "release_name",
             "release_short",

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,30 @@ 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.
+
+        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(
@@ -559,6 +583,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/app/tasks/copy.py

@@ -1,5 +1,6 @@
 from django.db import transaction
 from django.db.models import Q
+from gettext import gettext as _
 
 from pulpcore.plugin.models import Content, RepositoryVersion
 from pulpcore.plugin.util import get_domain_pk
@@ -13,6 +14,7 @@
     PackageGroup,
     RpmRepository,
     Modulemd,
+    Domain,
 )
 
 
@@ -140,6 +142,16 @@ def copy_content(config, dependency_solving):
             criteria MUST be validated before being passed to this task.
         content_pks: a list of content pks to copy from source to destination
     """
+    current_domain_pk = get_domain_pk()
+
+    def check_cross_domain_invariance(content_qs, original_content_count):
+        in_domain_count = content_qs.count()
+
+        if in_domain_count < original_content_count:
+            domain_name = Domain.objects.get(pk=current_domain_pk)
+            raise ValueError(
+                _("All content units must belong to the {} domain.").format(domain_name.name)
+            )
 
     def process_entry(entry):
         source_repo_version = RepositoryVersion.objects.get(pk=entry["source_repo_version"])
@@ -151,18 +163,21 @@ def process_entry(entry):
         else:
             dest_repo_version = dest_repo.latest_version()
 
-        if entry.get("content") is not None:
-            content_filter = Q(pk__in=entry.get("content"))
+        if (content := entry.get("content")) is not None:
+            content_filter = Q(pk__in=content)
+            content_count = len(content)
         else:
             content_filter = Q()
+            content_count = 0
 
-        content_filter &= Q(pulp_domain=get_domain_pk())
+        content_filter &= Q(pulp_domain=current_domain_pk)
 
         return (
             source_repo_version,
             dest_repo_version,
             dest_repo,
             content_filter,
+            content_count,
             dest_version_provided,
         )
 
@@ -175,10 +190,12 @@ def process_entry(entry):
                 dest_repo_version,
                 dest_repo,
                 content_filter,
+                content_count,
                 dest_version_provided,
             ) = process_entry(entry)
 
             content_to_copy = source_repo_version.content.filter(content_filter)
+            check_cross_domain_invariance(content_to_copy, content_count)
             content_to_copy |= find_children_of_content(content_to_copy, source_repo_version)
 
             base_version = dest_repo_version if dest_version_provided else None

pulp_rpm/app/viewsets/repository.py

@@ -714,6 +714,7 @@ def create(self, request):
 
         config, shared_repos, exclusive_repos = self._process_config(config)
 
+        # The task must check that objects belong to the current domain
         async_result = dispatch(
             tasks.copy_content,
             shared_resources=shared_repos,

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

@@ -0,0 +1,17 @@
+import pytest
+
+
+@pytest.mark.parallel
+def test_prn_schema(pulp_openapi_schema):
+    """Test that PRN is a part of every serializer with a pulp_href."""
+    failed = []
+    for name, schema in pulp_openapi_schema["components"]["schemas"].items():
+        if name.endswith("Response"):
+            if "pulp_href" in schema["properties"]:
+                if "prn" in schema["properties"]:
+                    prn_schema = schema["properties"]["prn"]
+                    if prn_schema["type"] == "string" and prn_schema["readOnly"]:
+                        continue
+                failed.append(name)
+
+    assert len(failed) == 0