Add additional tests for replication

Generated-By: claude-opus-4.6

Author: dralley

pulpcore/app/replica.py

@@ -174,18 +174,10 @@ def distribution_extra_fields(self, repository, upstream_distribution):
         """
         Return the fields that need to be updated/cleared on distributions for idempotence.
 
-        Note: repository_version is computed here but filtered out for updates/creates.
-        It will be set atomically in finalize_replication after sync completes.
+        Note: repository, publication, and repository_version are NOT included here.
+        They are all updated atomically in finalize_replication after all syncs complete.
         """
-        latest = repository.latest_version()
-        if latest:
-            repo_version_href = get_url(repository) + "versions/{}/".format(latest.number)
-        else:
-            repo_version_href = None
         return {
-            "repository": None,
-            "repository_version": repo_version_href,
-            "publication": None,
             "base_path": upstream_distribution["base_path"],
         }
 
@@ -198,35 +190,22 @@ def create_or_update_distribution(self, repository, upstream_distribution):
             )
             if not self._is_managed(distro):
                 return None
-            # Don't update repository_version here — that happens atomically in
-            # finalize_replication after all syncs complete.
-            # Do clear repository and publication so they don't conflict.
-            update_data = {k: v for k, v in distribution_data.items() if k != "repository_version"}
-            needs_update = self.needs_update(update_data, distro)
+            needs_update = self.needs_update(distribution_data, distro)
             if needs_update:
-                # Update the distribution
                 dispatch(
                     ageneral_update,
                     task_group=self.task_group,
                     shared_resources=[repository, self.server],
                     exclusive_resources=self.distros_uris,
                     args=(distro.pk, self.app_label, self.distribution_serializer_name),
                     kwargs={
-                        "data": update_data,
+                        "data": distribution_data,
                         "partial": True,
                     },
                 )
         except self.distribution_model_cls.DoesNotExist:
-            # Dispatch a task to create the distribution
-            # Don't set repository_version for new distributions - it will be set in
-            # finalize_replication after sync completes.
-            create_data = {
-                k: v
-                for k, v in distribution_data.items()
-                if k not in ("repository_version", "repository", "publication")
-            }
+            create_data = dict(distribution_data)
             create_data["name"] = upstream_distribution["name"]
-            create_data["pulp_labels"] = distribution_data["pulp_labels"]
             dispatch(
                 general_create,
                 task_group=self.task_group,

pulpcore/app/tasks/replica.py

@@ -124,16 +124,24 @@ def finalize_replication(server_pk, distro_repo_pairs):
     if task_group.tasks.exclude(pk=task.pk).exclude(state=TASK_STATES.COMPLETED).exists():
         raise Exception("Replication failed.")
 
-    # Atomically update all managed distributions to point to their repo's latest version.
+    # Atomically update all managed distributions to point to their repo's latest version,
+    # clearing any previous repository or publication references.
     with transaction.atomic():
         for distro_name, repo_pk in distro_repo_pairs:
             distro = Distribution.objects.get(name=distro_name, pulp_domain=server.pulp_domain)
             repo = Repository.objects.get(pk=repo_pk)
             latest_version = repo.latest_version()
             if latest_version:
-                if distro.repository_version != latest_version:
+                needs_update = (
+                    distro.repository_version != latest_version
+                    or distro.repository is not None
+                    or distro.publication is not None
+                )
+                if needs_update:
+                    distro.repository = None
+                    distro.publication = None
                     distro.repository_version = latest_version
-                    distro.save(update_fields=["repository_version"])
+                    distro.save(update_fields=["repository", "publication", "repository_version"])
 
     # Record timestamp of last successful replication.
     started_at = task_group.tasks.aggregate(Min("started_at"))["started_at__min"]

pulpcore/tests/functional/api/test_replication.py

@@ -13,8 +13,8 @@ def test_replication(
     pulp_settings,
     gen_object_with_cleanup,
 ):
-    # This test assures that an Upstream Pulp can be created in a non-default domain and that this
-    # Upstream Pulp configuration can be used to execute the replicate task.
+    """This test assures that an Upstream Pulp can be created in a non-default domain and that this
+    Upstream Pulp configuration can be used to execute the replicate task."""
 
     # Create a non-default domain
     non_default_domain = domain_factory()
@@ -57,8 +57,8 @@ def test_replication_idempotence(
     tmp_path,
     add_domain_objects_to_cleanup,
 ):
-    # This test assures that an Upstream Pulp can be created in a non-default domain and that this
-    # Upstream Pulp configuration can be used to execute the replicate task.
+    """This test assures that an Upstream Pulp can be created in a non-default domain and that this
+    Upstream Pulp configuration can be used to execute the replicate task."""
 
     # Create a domain to replicate from
     source_domain = domain_factory()
@@ -168,6 +168,186 @@ def test_replication_idempotence(
     assert upstream_pulp2.prn.split(":")[-1] == new_remote.pulp_labels["UpstreamPulp"]
 
 
+@pytest.mark.parallel
+def test_replication_with_repo_based_distribution(
+    domain_factory,
+    bindings_cfg,
+    pulpcore_bindings,
+    file_bindings,
+    monitor_task,
+    monitor_task_group,
+    pulp_settings,
+    gen_object_with_cleanup,
+    file_distribution_factory,
+    file_repository_factory,
+    file_remote_factory,
+    basic_manifest_path,
+    add_domain_objects_to_cleanup,
+):
+    """Test replication when upstream distribution uses repository (not publication)."""
+    source_domain = domain_factory()
+    add_domain_objects_to_cleanup(source_domain)
+
+    # Create a repo with autopublish, sync it, and distribute via repository (not publication)
+    remote = file_remote_factory(
+        pulp_domain=source_domain.name, manifest_path=basic_manifest_path, policy="immediate"
+    )
+    repo = file_repository_factory(pulp_domain=source_domain.name, autopublish=True)
+    sync_data = file_bindings.module.RepositorySyncURL(remote=remote.pulp_href, mirror=True)
+    monitor_task(file_bindings.RepositoriesFileApi.sync(repo.pulp_href, sync_data).task)
+    _ = file_distribution_factory(pulp_domain=source_domain.name, repository=repo.pulp_href)
+
+    # Replicate
+    replica_domain = domain_factory()
+    add_domain_objects_to_cleanup(replica_domain)
+    upstream_pulp = gen_object_with_cleanup(
+        pulpcore_bindings.UpstreamPulpsApi,
+        {
+            "name": str(uuid.uuid4()),
+            "base_url": bindings_cfg.host,
+            "api_root": pulp_settings.API_ROOT,
+            "domain": source_domain.name,
+            "username": bindings_cfg.username,
+            "password": bindings_cfg.password,
+        },
+        pulp_domain=replica_domain.name,
+    )
+    response = pulpcore_bindings.UpstreamPulpsApi.replicate(upstream_pulp.pulp_href)
+    monitor_task_group(response.task_group)
+
+    # Verify replica distribution uses repository_version, not repository or publication
+    replica_distro = file_bindings.DistributionsFileApi.list(
+        pulp_domain=replica_domain.name
+    ).results[0]
+    assert replica_distro.repository is None
+    assert replica_distro.repository_version is not None
+    assert replica_distro.publication is None
+
+    # Verify content was replicated
+    source_repo = file_bindings.RepositoriesFileApi.read(repo.pulp_href)
+    source_present = file_bindings.RepositoriesFileVersionsApi.read(
+        source_repo.latest_version_href
+    ).content_summary.present
+    replica_present = file_bindings.RepositoriesFileVersionsApi.read(
+        replica_distro.repository_version
+    ).content_summary.present
+    assert source_present["file.file"]["count"] == replica_present["file.file"]["count"]
+
+
+@pytest.mark.parallel
+def test_replication_multi_distribution_content_update(
+    domain_factory,
+    bindings_cfg,
+    pulpcore_bindings,
+    file_bindings,
+    monitor_task,
+    monitor_task_group,
+    pulp_settings,
+    gen_object_with_cleanup,
+    file_distribution_factory,
+    file_repository_factory,
+    file_publication_factory,
+    add_domain_objects_to_cleanup,
+    tmp_path,
+):
+    """Test that all distributions are updated after re-replication with new content."""
+    source_domain = domain_factory()
+    add_domain_objects_to_cleanup(source_domain)
+
+    # Create 3 repos with content and publication-based distributions
+    distros = []
+    repos = []
+    for i in range(3):
+        repo = file_repository_factory(pulp_domain=source_domain.name)
+        repos.append(repo)
+        file_path = tmp_path / f"file_{i}.txt"
+        file_path.write_text(f"content_{i}")
+        monitor_task(
+            file_bindings.ContentFilesApi.create(
+                file=str(file_path),
+                relative_path=f"file_{i}.txt",
+                repository=repo.pulp_href,
+                pulp_domain=source_domain.name,
+            ).task
+        )
+        pub = file_publication_factory(pulp_domain=source_domain.name, repository=repo.pulp_href)
+        distros.append(
+            file_distribution_factory(pulp_domain=source_domain.name, publication=pub.pulp_href)
+        )
+
+    # Initial replication
+    replica_domain = domain_factory()
+    add_domain_objects_to_cleanup(replica_domain)
+    upstream_pulp = gen_object_with_cleanup(
+        pulpcore_bindings.UpstreamPulpsApi,
+        {
+            "name": str(uuid.uuid4()),
+            "base_url": bindings_cfg.host,
+            "api_root": pulp_settings.API_ROOT,
+            "domain": source_domain.name,
+            "username": bindings_cfg.username,
+            "password": bindings_cfg.password,
+        },
+        pulp_domain=replica_domain.name,
+    )
+    response = pulpcore_bindings.UpstreamPulpsApi.replicate(upstream_pulp.pulp_href)
+    monitor_task_group(response.task_group)
+
+    # Record initial versions
+    replica_distros = file_bindings.DistributionsFileApi.list(
+        pulp_domain=replica_domain.name
+    ).results
+    assert len(replica_distros) == 3
+    initial_versions = {}
+    for rd in replica_distros:
+        assert rd.repository is None
+        assert rd.repository_version is not None
+        assert rd.publication is None
+        initial_versions[rd.name] = rd.repository_version
+
+    # Add new content to all source repos and update publications
+    for i, repo in enumerate(repos):
+        file_path = tmp_path / f"file_{i}_v2.txt"
+        file_path.write_text(f"new_content_{i}")
+        monitor_task(
+            file_bindings.ContentFilesApi.create(
+                file=str(file_path),
+                relative_path=f"file_{i}_v2.txt",
+                repository=repo.pulp_href,
+                pulp_domain=source_domain.name,
+            ).task
+        )
+        repo = file_bindings.RepositoriesFileApi.read(repo.pulp_href)
+        pub = file_publication_factory(
+            pulp_domain=source_domain.name,
+            repository_version=repo.latest_version_href,
+        )
+        monitor_task(
+            file_bindings.DistributionsFileApi.partial_update(
+                distros[i].pulp_href, {"publication": pub.pulp_href}
+            ).task
+        )
+
+    # Re-replicate
+    response = pulpcore_bindings.UpstreamPulpsApi.replicate(upstream_pulp.pulp_href)
+    monitor_task_group(response.task_group)
+
+    # Verify all distributions were updated to new versions with new content
+    replica_distros = file_bindings.DistributionsFileApi.list(
+        pulp_domain=replica_domain.name
+    ).results
+    assert len(replica_distros) == 3
+    for rd in replica_distros:
+        assert rd.repository is None
+        assert rd.repository_version is not None
+        assert rd.publication is None
+        # Version should have changed
+        assert rd.repository_version != initial_versions[rd.name]
+        # Verify new content is present (original file + new file = 2)
+        version = file_bindings.RepositoriesFileVersionsApi.read(rd.repository_version)
+        assert version.content_summary.present["file.file"]["count"] == 2
+
+
 @pytest.mark.parallel
 def test_replication_with_wrong_ca_cert(
     domain_factory,
@@ -177,9 +357,9 @@ def test_replication_with_wrong_ca_cert(
     pulp_settings,
     gen_object_with_cleanup,
 ):
-    # This test assures that setting ca_cert on an Upstream Pulp causes that CA bundle to be used
-    # to verify the certificate presented by the Upstream Pulp's REST API. The replication tasks
-    # are expected to fail.
+    """This test assures that setting ca_cert on an Upstream Pulp causes that CA bundle to be used
+    to verify the certificate presented by the Upstream Pulp's REST API. The replication tasks
+    are expected to fail."""
 
     if not bindings_cfg.host.startswith("https"):
         pytest.skip("HTTPS is not enabled for Pulp's API.")