Add a test for filtering distributions by content

Required PR: pulp/pulpcore#2992

[noissue]

pulp_file/tests/functional/api/from_pulpcore/test_distributions.py

@@ -2,6 +2,7 @@
 import csv
 import hashlib
 import pytest
+import uuid
 from time import sleep
 import unittest
 from urllib.parse import urljoin
@@ -26,6 +27,7 @@
 from pulpcore.client.pulp_file import (
     ContentFilesApi,
     DistributionsFileApi,
+    FileFileDistribution,
     FileFilePublication,
     PublicationsFileApi,
     RemotesFileApi,
@@ -511,3 +513,63 @@ def do_range_request_download_test(self, range_header, expected_bytes):
         self.assertEqual(
             req1_reponse.headers["Content-Range"], req2_response.headers["Content-Range"]
         )
+
+
+def test_distribution_filtering(
+    file_content_api_client,
+    file_distro_api_client,
+    file_fixture_gen_remote,
+    file_random_content_unit,
+    file_repo_api_client,
+    file_pub_api_client,
+    gen_object_with_cleanup,
+    write_3_iso_file_fixture_data_factory,
+):
+    """Test distribution filtering based on the content exposed from the distribution."""
+
+    def generate_repo_with_content():
+        repo = gen_object_with_cleanup(file_repo_api_client, gen_repo())
+        repo_manifest_path = write_3_iso_file_fixture_data_factory(str(uuid.uuid4()))
+        remote = file_fixture_gen_remote(manifest_path=repo_manifest_path, policy="on_demand")
+        body = RepositorySyncURL(remote=remote.pulp_href)
+        task_response = file_repo_api_client.sync(repo.pulp_href, body).task
+        version_href = monitor_task(task_response).created_resources[0]
+        content = file_content_api_client.list(repository_version_added=version_href).results[0]
+        return repo, content
+
+    repo1, content1 = generate_repo_with_content()
+
+    publish_data = FileFilePublication(repository=repo1.pulp_href)
+    publication = gen_object_with_cleanup(file_pub_api_client, publish_data)
+
+    data = FileFileDistribution(name="pub1", base_path="pub1", publication=publication.pulp_href)
+    distribution_pub1 = gen_object_with_cleanup(file_distro_api_client, data)
+
+    results = file_distro_api_client.list(with_content=content1.pulp_href).results
+    assert [distribution_pub1] == results
+
+    data = FileFileDistribution(name="pub2", base_path="pub2", publication=publication.pulp_href)
+    distribution_pub2 = gen_object_with_cleanup(file_distro_api_client, data)
+
+    results = set(
+        d.pulp_href for d in file_distro_api_client.list(with_content=content1.pulp_href).results
+    )
+    assert {distribution_pub1.pulp_href, distribution_pub2.pulp_href} == results
+
+    repo2, content2 = generate_repo_with_content()
+
+    data = FileFileDistribution(name="repo", base_path="repo", repository=repo2.pulp_href)
+    distribution_repo = gen_object_with_cleanup(file_distro_api_client, data)
+
+    results = file_distro_api_client.list(with_content=content2.pulp_href).results
+    assert [distribution_repo] == results
+
+    results = file_distro_api_client.list(with_content=file_random_content_unit.pulp_href).results
+    assert [] == results
+
+    results = set(d.pulp_href for d in file_distro_api_client.list().results)
+    assert {
+        distribution_pub1.pulp_href,
+        distribution_pub2.pulp_href,
+        distribution_repo.pulp_href
+    } == results

pulp_file/tests/functional/utils.py

@@ -75,15 +75,6 @@ def gen_file_content_attrs(artifact):
     return {"artifact": artifact["pulp_href"], "relative_path": utils.uuid4()}
 
 
-def gen_file_content_upload_attrs():
-    """Generate a dict with content unit attributes without artifact for upload.
-
-    :param artifact: A dict of info about the artifact.
-    :returns: A semi-random dict for use in creating a content unit.
-    """
-    return {"relative_path": utils.uuid4()}
-
-
 def populate_pulp(cfg, url=FILE_FIXTURE_MANIFEST_URL):
     """Add file contents to Pulp.