Allow content to be copied by href

fixes #6019
https://pulp.plan.io/issues/6019

CHANGES/6019.feature

@@ -0,0 +1 @@
+Add a content parameter to the copy api that accepts a list of hrefs to be copied.

pulp_rpm/app/serializers.py

@@ -19,7 +19,6 @@
     PublicationSerializer,
     RepositorySyncURLSerializer,
     RelatedField,
-    DetailRelatedField,
     RemoteSerializer,
     RepositorySerializer,
     SingleArtifactContentUploadSerializer,
@@ -617,6 +616,11 @@ class CopySerializer(serializers.Serializer):
         required=False,
     )
 
+    content = serializers.ListField(
+        help_text=_("A list of content unit hrefs to copy from one repo to the other"),
+        required=False,
+    )
+
     dependency_solving = serializers.BooleanField(
         help_text=_('Also copy dependencies of the content being copied.'),
         default=True
@@ -636,7 +640,13 @@ def validate(self, data):
         if hasattr(self, 'initial_data'):
             validate_unknown_fields(self.initial_data, self.fields)
 
+        if 'criteria' in data and 'content' in data:
+            raise serializers.ValidationError(
+                _("Criteria and content fields cannot both be set.")
+            )
+
         criteria = data.get('criteria')
+
         if criteria:
             validator = Draft7Validator(COPY_CRITERIA_SCHEMA)
 

pulp_rpm/app/tasks/copy.py

@@ -6,29 +6,33 @@
 from pulp_rpm.app.models import RpmRepository
 
 
-def _filter_content(content, criteria):
+def _filter_content(content, criteria, content_pks):
     """
     Filter content in the source repository version by criteria.
 
     Args:
         content: a queryset of content to filter
         criteria: a validated dict that maps content type to a list of filter criteria
+        content_pks: a whitelist of content_pks to filter content
     """
-    if not criteria:
+    if not criteria and not content_pks:
+        # if we have neither criteria and content pks, we're copying everything
         return content
 
-    content_pks = []
-    for content_type in RpmRepository.CONTENT_TYPES:
-        if criteria.get(content_type.TYPE):
-            filters = Q()
-            for filter in criteria[content_type.TYPE]:
-                filters |= Q(**filter)
-            content_pks += content_type.objects.filter(filters).values_list("pk", flat=True)
+    if criteria:
+        # find the content_pks based on criteria
+        content_pks = []
+        for content_type in RpmRepository.CONTENT_TYPES:
+            if criteria.get(content_type.TYPE):
+                filters = Q()
+                for filter in criteria[content_type.TYPE]:
+                    filters |= Q(**filter)
+                content_pks += content_type.objects.filter(filters).values_list("pk", flat=True)
 
     return content.filter(pk__in=content_pks)
 
 
-def copy_content(source_repo_version_pk, dest_repo_pk, criteria, dependency_solving):
+def copy_content(source_repo_version_pk, dest_repo_pk, criteria, content_pks, dependency_solving):
     """
     Copy content from one repo to another.
 
@@ -37,6 +41,7 @@ def copy_content(source_repo_version_pk, dest_repo_pk, criteria, dependency_solv
         dest_repo_pk: repository primary key to copy units into
         criteria: a dict that maps type to a list of criteria to filter content by. Note that this
             criteria MUST be validated before being passed to this task.
+        content_pks: a list of content pks to copy from source to destination
     """
     source_repo_version = RepositoryVersion.objects.get(pk=source_repo_version_pk)
     # source_repo = RpmRepository.objects.get(pk=source_repo_version.repository)
@@ -45,7 +50,7 @@ def copy_content(source_repo_version_pk, dest_repo_pk, criteria, dependency_solv
     dest_repo_version = dest_repo.latest_version()
 
     if not dependency_solving:
-        content_to_copy = _filter_content(source_repo_version.content, criteria)
+        content_to_copy = _filter_content(source_repo_version.content, criteria, content_pks)
 
         with dest_repo.new_version() as new_version:
             new_version.add_content(content_to_copy)
@@ -65,7 +70,7 @@ def copy_content(source_repo_version_pk, dest_repo_pk, criteria, dependency_solv
     for src in repo_mapping.keys():
         repo_name = solver.load_source_repo(src)
         libsolv_repo_names[repo_name] = src
-        content_to_copy[repo_name] = _filter_content(src.content, criteria)
+        content_to_copy[repo_name] = _filter_content(src.content, criteria, content_pks)
 
     for tgt in repo_mapping.values():
         solver.load_target_repo(tgt)

pulp_rpm/app/viewsets.py

@@ -4,11 +4,13 @@
 from rest_framework.decorators import action
 
 from pulpcore.plugin.actions import ModifyRepositoryActionMixin
+from pulpcore.plugin.models import Content
 from pulpcore.plugin.tasking import enqueue_with_reservation
 from pulpcore.plugin.serializers import AsyncOperationResponseSerializer
 from pulpcore.plugin.viewsets import (
     BaseDistributionViewSet,
     ContentFilter,
+    NamedModelViewSet,
     OperationPostponedResponse,
     PublicationViewSet,
     ReadOnlyContentViewSet,
@@ -257,13 +259,21 @@ def create(self, request):
         dest_repo = serializer.validated_data['dest_repo']
         criteria = serializer.validated_data.get('criteria', None)
         dependency_solving = serializer.validated_data['dependency_solving']
+        content_urls = serializer.validated_data.get('content', None)
+
+        content = []
+        if content_urls:
+            for url in content_urls:
+                c = NamedModelViewSet().get_resource(url, Content)
+                content.append(c.pk)
 
         source_repos = [source_repo]
         dest_repos = [dest_repo]
 
         async_result = enqueue_with_reservation(
             tasks.copy_content, [*source_repos, *dest_repos],
-            args=[source_repo.latest_version().pk, dest_repo.pk, criteria, dependency_solving],
+            args=[source_repo.latest_version().pk, dest_repo.pk, criteria, content,
+                  dependency_solving],
             kwargs={}
         )
         return OperationPostponedResponse(async_result, request)

pulp_rpm/tests/functional/api/test_copy.py

@@ -18,6 +18,7 @@
     RPM_FIXTURE_SUMMARY,
     RPM_REMOTE_PATH,
     RPM_REPO_PATH,
+    UPDATERECORD_CONTENT_PATH,
 )
 from pulp_rpm.tests.functional.utils import gen_rpm_remote, rpm_copy
 from pulp_rpm.tests.functional.utils import set_up_module as setUpModule  # noqa:F401
@@ -34,17 +35,15 @@ def setUpClass(cls):
 
         delete_orphans(cls.cfg)
 
-    def _do_test(self, criteria, expected_results):
-        """Test copying content units with the RPM plugin.
+    def _setup_repos(self):
+        """Prepare for a copy test by creating two repos and syncing.
 
         Do the following:
 
         1. Create two repositories and a remote.
         2. Sync the remote.
         3. Assert that repository version is not None.
         4. Assert that the correct number of units were added and are present in the repo.
-        5. Use the RPM copy API to units from the repo to the empty repo.
-        7. Assert that the correct number of units were added and are present in the dest repo.
         """
         source_repo = self.client.post(RPM_REPO_PATH, gen_repo())
         self.addCleanup(self.client.delete, source_repo['pulp_href'])
@@ -69,6 +68,18 @@ def _do_test(self, criteria, expected_results):
             get_added_content_summary(source_repo), RPM_FIXTURE_SUMMARY
         )
 
+        return source_repo, dest_repo
+
+    def _do_test(self, criteria, expected_results):
+        """Test copying content units with the RPM plugin.
+
+        Calls _setup_repos() which creates two repos and syncs to the source. Then:
+
+        1. Use the RPM copy API to units from the repo to the empty repo.
+        2. Assert that the correct number of units were added and are present in the dest repo.
+        """
+        source_repo, dest_repo = self._setup_repos()
+
         rpm_copy(self.cfg, source_repo, dest_repo, criteria)
         dest_repo = self.client.get(dest_repo['pulp_href'])
 
@@ -78,22 +89,11 @@ def _do_test(self, criteria, expected_results):
             get_added_content_summary(dest_repo), expected_results,
         )
 
-    def test_copy_none(self):
-        """Test copying none of the content."""
-        criteria = {}
-        results = {}
-        self._do_test(criteria, results)
-
     def test_copy_all(self):
         """Test copying all the content from one repo to another."""
         results = RPM_FIXTURE_SUMMARY
         self._do_test(None, results)
 
-    # def test_copy_by_href(self):
-    #     criteria = {}
-    #     results = {}
-    #     self._do_test(criteria, results)
-
     def test_copy_by_advisory_id(self):
         """Test copying an advisory by its id."""
         criteria = {
@@ -118,6 +118,34 @@ def test_invalid_criteria(self):
             }
             self._do_test(criteria, {})
 
+    def test_both_criteria_and_content(self):
+        """Test that specifying 'content' and 'criteria' is invalid."""
+        criteria = {
+            'advisory': [{'bad': 'field'}]
+        }
+        content = ["/pulp/api/v3/content/rpm/packages/6d0e1e29-ae58-4c70-8bcb-ae957250fc8f/"]
+        self._setup_repos()
+        source_repo, dest_repo = self._setup_repos()
+
+        with self.assertRaises(HTTPError):
+            rpm_copy(self.cfg, source_repo, dest_repo, criteria, content)
+
+    def test_content(self):
+        """Test the content parameter."""
+        source_repo, dest_repo = self._setup_repos()
+        latest_href = source_repo["latest_version_href"]
+
+        content = self.client.get(f"{UPDATERECORD_CONTENT_PATH}?repository_version={latest_href}")
+        content_to_copy = (content["results"][0]["pulp_href"], content["results"][1]["pulp_href"])
+
+        rpm_copy(self.cfg, source_repo, dest_repo, content=content_to_copy)
+
+        latest_href = dest_repo["pulp_href"] + "versions/1/"
+        dc = self.client.get(f"{UPDATERECORD_CONTENT_PATH}?repository_version={latest_href}")
+        dest_content = [c["pulp_href"] for c in dc["results"]]
+
+        self.assertEqual(sorted(content_to_copy), sorted(dest_content))
+
 
 class DependencySolvingTestCase(unittest.TestCase):
     """Copy units between repositories with the rpm plugin."""

pulp_rpm/tests/functional/constants.py

@@ -266,7 +266,7 @@
 SRPM_UNSIGNED_FIXTURE_URL = urljoin(PULP_FIXTURES_BASE_URL, 'srpm-unsigned/')
 """The URL to a repository with unsigned SRPM packages."""
 
-UPDATERECORD_CONTENT_PATH = urljoin(BASE_CONTENT_PATH, 'rpm/updates/')
+UPDATERECORD_CONTENT_PATH = urljoin(BASE_CONTENT_PATH, 'rpm/advisories/')
 """The location of RPM UpdateRecords on the content endpoint."""
 
 RPM_KICKSTART_FIXTURE_URL = urljoin(PULP_FIXTURES_BASE_URL, 'rpm-kickstart/')

pulp_rpm/tests/functional/utils.py

@@ -105,7 +105,7 @@ def populate_pulp(cfg, url=RPM_SIGNED_FIXTURE_URL):
     return client.get(RPM_CONTENT_PATH)['results']
 
 
-def rpm_copy(cfg, source_repo, dest_repo, criteria, recursive=False):
+def rpm_copy(cfg, source_repo, dest_repo, criteria=None, content=None, recursive=False):
     """Sync a repository.
 
     :param pulp_smash.config.PulpSmashConfig cfg: Information about the Pulp
@@ -121,13 +121,15 @@ def rpm_copy(cfg, source_repo, dest_repo, criteria, recursive=False):
     data = {
         'source_repo': source_repo['pulp_href'],
         'dest_repo': dest_repo['pulp_href'],
-        'criteria': criteria,
         'dependency_solving': recursive,
     }
 
     if criteria:
         data['criteria'] = criteria
 
+    if content:
+        data['content'] = content
+
     return client.post(RPM_COPY_PATH, data)