From dabe3bb55f5e39073204f04b6ea565b9706c37f7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pablo=20M=C3=A9ndez=20Hern=C3=A1ndez?= Date: Wed, 26 Aug 2026 12:35:14 +0200 Subject: [PATCH 1/2] fix: keep replicate TLS temp files alive Replication over HTTPS wrote the CA, client cert, and client key to NamedTemporaryFiles but only kept their paths, so CPython deleted earlier files before pulp-glue could load the upstream API with TLS verification. Keep those files alive for the duration of replicate(), pass the CA bundle path directly as verify_ssl, add regression coverage for the TLS handoff, and clean up the temp files on exit. Assisted-By: Cursor Co-authored-by: Cursor --- CHANGES/+replicate-ssl-tempfiles.bugfix | 1 + pulpcore/app/tasks/replica.py | 226 +++++++++++++----------- pulpcore/tests/unit/test_replica.py | 155 ++++++++++++++++ 3 files changed, 277 insertions(+), 105 deletions(-) create mode 100644 CHANGES/+replicate-ssl-tempfiles.bugfix create mode 100644 pulpcore/tests/unit/test_replica.py diff --git a/CHANGES/+replicate-ssl-tempfiles.bugfix b/CHANGES/+replicate-ssl-tempfiles.bugfix new file mode 100644 index 00000000000..3c9642aef77 --- /dev/null +++ b/CHANGES/+replicate-ssl-tempfiles.bugfix @@ -0,0 +1 @@ +Fixed replicate() deleting temporary TLS files before pulp-glue could use them, and stopped leaking PULP_CA_BUNDLE into later worker tasks. diff --git a/pulpcore/app/tasks/replica.py b/pulpcore/app/tasks/replica.py index 60510bc9d72..19de2bd60ed 100644 --- a/pulpcore/app/tasks/replica.py +++ b/pulpcore/app/tasks/replica.py @@ -1,6 +1,7 @@ import os import platform import sys +from contextlib import contextmanager from tempfile import NamedTemporaryFile from django.db import transaction @@ -28,115 +29,130 @@ def user_agent(): return f"pulpcore/{pulp_version} ({python}, {system}) (pulp-glue {pulp_glue_version})" +@contextmanager +def _ssl_temp_files(server): + """Write UpstreamPulp TLS material to temp files that live for this context.""" + ssl_files = {} + try: + for key in ["ca_cert", "client_cert", "client_key"]: + if value := getattr(server, key): + suffix = ".key" if key == "client_key" else ".pem" + with NamedTemporaryFile( + dir=".", mode="w", encoding="utf-8", delete=False, suffix=suffix + ) as f: + f.write(value) + f.flush() + ssl_files[key] = f.name + yield ssl_files + finally: + for path in ssl_files.values(): + try: + os.unlink(path) + except FileNotFoundError: + pass + + def replicate_distributions(server_pk, q_select=None, **kwargs): server = UpstreamPulp.objects.get(pk=server_pk) + with _ssl_temp_files(server) as ssl_files: + verify_ssl = ( + ssl_files["ca_cert"] + if server.tls_validation and "ca_cert" in ssl_files + else server.tls_validation + ) + ctx = ReplicaContext.from_config( + { + "base_url": server.base_url, + "api_root": server.api_root, + "domain": server.domain, + "username": server.username, + "password": server.password, + "cert": ssl_files.get("client_cert"), + "key": ssl_files.get("client_key"), + "user_agent": user_agent(), + "verify_ssl": verify_ssl, + "dry_run": True, # We only want to read from upstream anyway. + } + ) - # Write out temporary files related to SSL - ssl_files = {} - for key in ["ca_cert", "client_cert", "client_key"]: - if value := getattr(server, key): - f = NamedTemporaryFile(dir=".") - f.write(bytes(value, "utf-8")) - f.flush() - ssl_files[key] = f.name - - if "ca_cert" in ssl_files: - os.environ["PULP_CA_BUNDLE"] = ssl_files["ca_cert"] - - ctx = ReplicaContext.from_config( - { - "base_url": server.base_url, - "api_root": server.api_root, - "domain": server.domain, - "username": server.username, - "password": server.password, - "cert": ssl_files.get("client_cert"), - "key": ssl_files.get("client_key"), - "user_agent": user_agent(), - "verify_ssl": server.tls_validation, - "dry_run": True, # We only want to read from upstream anyway. + remote_settings = { + "ca_cert": server.ca_cert, + "tls_validation": server.tls_validation, + "client_cert": server.client_cert, + "client_key": server.client_key, + "download_concurrency": server.download_concurrency, + "max_retries": server.max_retries, + "total_timeout": server.total_timeout, + "connect_timeout": server.connect_timeout, + "sock_connect_timeout": server.sock_connect_timeout, + "sock_read_timeout": server.sock_read_timeout, } - ) - - remote_settings = { - "ca_cert": server.ca_cert, - "tls_validation": server.tls_validation, - "client_cert": server.client_cert, - "client_key": server.client_key, - "download_concurrency": server.download_concurrency, - "max_retries": server.max_retries, - "total_timeout": server.total_timeout, - "connect_timeout": server.connect_timeout, - "sock_connect_timeout": server.sock_connect_timeout, - "sock_read_timeout": server.sock_read_timeout, - } - - try: - task_group = TaskGroup.current() - supported_replicators = [] - # Load all the available replicators - for config in pulp_plugin_configs(): - if config.replicator_classes: - for replicator_class in config.replicator_classes: - req = PluginRequirement( - config.label, specifier=replicator_class.required_version - ) - if ctx.has_plugin(req): - replicator = replicator_class(ctx, task_group, remote_settings, server) - supported_replicators.append(replicator) - - effective_q_select = q_select if q_select is not None else server.q_select - distro_repo_pairs = [] - for replicator in supported_replicators: - distro_names = [] - pending_distributions = [] - distros = replicator.upstream_distributions(q=effective_q_select) - for distro in distros: - # Create remote - remote = replicator.create_or_update_remote(upstream_distribution=distro) - if not remote: - # The upstream distribution is not serving any content, - # let it fall through the cracks and be cleaned up below. - continue - # Check if there is already a repository - repository = replicator.create_or_update_repository(remote=remote) - if not repository: - # No update occurred because server.policy==LABELED and there was - # an already existing local repository with the same name - continue - - # Dispatch a sync task if needed - if replicator.requires_syncing(distro): - replicator.sync(repository, remote) - - # Add name to the list of known distribution names - distro_names.append(distro["name"]) - distro_repo_pairs.append((distro["name"], str(repository.pk))) - pending_distributions.append((repository, distro)) - - # Get or create distributions BEFORE remove_missing so that - # create_or_update_distribution can synchronously rename any existing - # distribution matched by base_path. remove_missing then sees the - # updated name in the DB and won't schedule it for deletion. - for repository, distro in pending_distributions: - replicator.create_or_update_distribution(repository, distro) - - # When a per-request q_select override is used, this is a selective sync - # of a subset of distributions. Skipping remove_missing avoids deleting - # distributions that simply weren't included in the filter — but it also - # means that distributions removed from upstream won't be cleaned up until - # a full (non-overridden) replication runs. - if q_select is None: - replicator.remove_missing(distro_names) - except GluePulpException as e: - raise ExternalServiceError(service_name=server.base_url, details=str(e)) - - dispatch( - finalize_replication, - task_group=task_group, - exclusive_resources=[server, distros_lock_uri(server.pulp_domain_id)], - args=[server.pk, distro_repo_pairs], - ) + try: + task_group = TaskGroup.current() + supported_replicators = [] + # Load all the available replicators + for config in pulp_plugin_configs(): + if config.replicator_classes: + for replicator_class in config.replicator_classes: + req = PluginRequirement( + config.label, specifier=replicator_class.required_version + ) + if ctx.has_plugin(req): + replicator = replicator_class(ctx, task_group, remote_settings, server) + supported_replicators.append(replicator) + + effective_q_select = q_select if q_select is not None else server.q_select + distro_repo_pairs = [] + for replicator in supported_replicators: + distro_names = [] + pending_distributions = [] + distros = replicator.upstream_distributions(q=effective_q_select) + for distro in distros: + # Create remote + remote = replicator.create_or_update_remote(upstream_distribution=distro) + if not remote: + # The upstream distribution is not serving any content, + # let it fall through the cracks and be cleaned up below. + continue + # Check if there is already a repository + repository = replicator.create_or_update_repository(remote=remote) + if not repository: + # No update occurred because server.policy==LABELED and there was + # an already existing local repository with the same name + continue + + # Dispatch a sync task if needed + if replicator.requires_syncing(distro): + replicator.sync(repository, remote) + + # Add name to the list of known distribution names + distro_names.append(distro["name"]) + distro_repo_pairs.append((distro["name"], str(repository.pk))) + pending_distributions.append((repository, distro)) + + # Get or create distributions BEFORE remove_missing so that + # create_or_update_distribution can synchronously rename any existing + # distribution matched by base_path. remove_missing then sees the + # updated name in the DB and won't schedule it for deletion. + for repository, distro in pending_distributions: + replicator.create_or_update_distribution(repository, distro) + + # When a per-request q_select override is used, this is a selective sync + # of a subset of distributions. Skipping remove_missing avoids deleting + # distributions that simply weren't included in the filter — but it also + # means that distributions removed from upstream won't be cleaned up until + # a full (non-overridden) replication runs. + if q_select is None: + replicator.remove_missing(distro_names) + except GluePulpException as e: + raise ExternalServiceError(service_name=server.base_url, details=str(e)) + + dispatch( + finalize_replication, + task_group=task_group, + exclusive_resources=[server, distros_lock_uri(server.pulp_domain_id)], + args=[server.pk, distro_repo_pairs], + ) def finalize_replication(server_pk, distro_repo_pairs, **kwargs): diff --git a/pulpcore/tests/unit/test_replica.py b/pulpcore/tests/unit/test_replica.py new file mode 100644 index 00000000000..e5f56ed2e73 --- /dev/null +++ b/pulpcore/tests/unit/test_replica.py @@ -0,0 +1,155 @@ +import os +from types import SimpleNamespace + +from pulpcore.app.tasks import replica +from pulpcore.app.tasks.replica import _ssl_temp_files + + +def test_ssl_temp_files_keep_all_certs_until_context_exits(tmp_path, monkeypatch): + """Katello replicate() sends ca_cert + client_cert + client_key together. + + The old loop stored only filenames, so the CA NamedTemporaryFile was + garbage-collected (and unlinked) before pulp-glue opened it. + """ + monkeypatch.chdir(tmp_path) + server = SimpleNamespace( + ca_cert="-----BEGIN CA-----\nca\n-----END CA-----", + client_cert="-----BEGIN CERT-----\ncert\n-----END CERT-----", + client_key="-----BEGIN KEY-----\nkey\n-----END KEY-----", + ) + + with _ssl_temp_files(server) as ssl_files: + for key, expected in ( + ("ca_cert", server.ca_cert), + ("client_cert", server.client_cert), + ("client_key", server.client_key), + ): + path = ssl_files[key] + assert os.path.exists(path) + with open(path, encoding="utf-8") as f: + assert f.read() == expected + ca_path = ssl_files["ca_cert"] + cert_path = ssl_files["client_cert"] + key_path = ssl_files["client_key"] + + assert not os.path.exists(ca_path) + assert not os.path.exists(cert_path) + assert not os.path.exists(key_path) + + +def test_ssl_temp_files_skips_missing_material(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + server = SimpleNamespace(ca_cert="ca", client_cert=None, client_key=None) + + with _ssl_temp_files(server) as ssl_files: + assert set(ssl_files) == {"ca_cert"} + assert os.path.exists(ssl_files["ca_cert"]) + + +def test_replicate_distributions_passes_ca_path_as_verify_ssl(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + captured = {} + server = SimpleNamespace( + base_url="https://example.com", + api_root="/pulp/", + domain="default", + username="user", + password="pass", + ca_cert="ca", + client_cert="cert", + client_key="key", + tls_validation=True, + download_concurrency=10, + max_retries=3, + total_timeout=30, + connect_timeout=5, + sock_connect_timeout=5, + sock_read_timeout=5, + q_select=None, + pulp_domain_id="domain-id", + pk="server-pk", + ) + + class DummyContext: + def has_plugin(self, req): + assert os.path.exists(captured["config"]["verify_ssl"]) + assert os.path.exists(captured["config"]["cert"]) + assert os.path.exists(captured["config"]["key"]) + return False + + class DummyReplicator: + required_version = ">=0" + + def fake_from_config(config): + captured["config"] = config + assert os.path.exists(config["verify_ssl"]) + assert os.path.exists(config["cert"]) + assert os.path.exists(config["key"]) + return DummyContext() + + monkeypatch.setattr(replica.UpstreamPulp.objects, "get", lambda pk: server) + monkeypatch.setattr(replica.ReplicaContext, "from_config", fake_from_config) + monkeypatch.setattr( + replica, + "pulp_plugin_configs", + lambda: [SimpleNamespace(label="core", replicator_classes=[DummyReplicator])], + ) + monkeypatch.setattr(replica.TaskGroup, "current", lambda: "task-group") + monkeypatch.setattr(replica, "dispatch", lambda *args, **kwargs: None) + + replica.replicate_distributions(server.pk) + + assert isinstance(captured["config"]["verify_ssl"], str) + + +def test_replicate_distributions_uses_false_verify_ssl_when_disabled(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + captured = {} + server = SimpleNamespace( + base_url="https://example.com", + api_root="/pulp/", + domain="default", + username="user", + password="pass", + ca_cert="ca", + client_cert="cert", + client_key="key", + tls_validation=False, + download_concurrency=10, + max_retries=3, + total_timeout=30, + connect_timeout=5, + sock_connect_timeout=5, + sock_read_timeout=5, + q_select=None, + pulp_domain_id="domain-id", + pk="server-pk", + ) + + class DummyContext: + def has_plugin(self, req): + return False + + class DummyReplicator: + required_version = ">=0" + + def fake_from_config(config): + captured["config"] = config + assert config["verify_ssl"] is False + assert os.path.exists(config["cert"]) + assert os.path.exists(config["key"]) + return DummyContext() + + monkeypatch.setattr(replica.UpstreamPulp.objects, "get", lambda pk: server) + monkeypatch.setattr(replica.ReplicaContext, "from_config", fake_from_config) + monkeypatch.setattr( + replica, + "pulp_plugin_configs", + lambda: [SimpleNamespace(label="core", replicator_classes=[DummyReplicator])], + ) + monkeypatch.setattr(replica.TaskGroup, "current", lambda: "task-group") + monkeypatch.setattr(replica, "dispatch", lambda *args, **kwargs: None) + + replica.replicate_distributions(server.pk) + + assert captured["config"]["verify_ssl"] is False From b45c5d5f78f25bf167c7e3cbddf2d0c29ba8985b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pablo=20M=C3=A9ndez=20Hern=C3=A1ndez?= Date: Thu, 27 Aug 2026 10:07:45 +0200 Subject: [PATCH 2/2] Add UpstreamPulp.remote_policy for remotes created during replication replicate() never set Remote.policy, so new remotes defaulted to immediate and downloaded all artifacts. Let UpstreamPulp carry the intended download policy so Capsules can replicate with on_demand. Assisted-By: Cursor Grok 4.6 Co-authored-by: Cursor --- CHANGES/+remote-policy.feature | 1 + docs/user/guides/replication.md | 4 +- .../0157_upstreampulp_remote_policy.py | 34 +++++++++ pulpcore/app/models/replica.py | 3 + pulpcore/app/serializers/replica.py | 13 +++- pulpcore/app/tasks/replica.py | 33 ++++++--- .../tests/functional/api/test_replication.py | 74 +++++++++++++++++++ .../tests/unit/test_replica_remote_policy.py | 35 +++++++++ 8 files changed, 183 insertions(+), 14 deletions(-) create mode 100644 CHANGES/+remote-policy.feature create mode 100644 pulpcore/app/migrations/0157_upstreampulp_remote_policy.py create mode 100644 pulpcore/tests/unit/test_replica_remote_policy.py diff --git a/CHANGES/+remote-policy.feature b/CHANGES/+remote-policy.feature new file mode 100644 index 00000000000..712da8f4e85 --- /dev/null +++ b/CHANGES/+remote-policy.feature @@ -0,0 +1 @@ +Added `UpstreamPulp.remote_policy` so remotes created during replication can use `on_demand` or `streamed` instead of defaulting to `immediate`. diff --git a/docs/user/guides/replication.md b/docs/user/guides/replication.md index a8b627e3e42..7bf07c3a4c6 100644 --- a/docs/user/guides/replication.md +++ b/docs/user/guides/replication.md @@ -48,6 +48,7 @@ pulp upstream-pulp create \ | `tls_validation` | Whether to verify the upstream server's TLS certificate. Defaults to `True`. | | `q_select` | A filter expression to select which upstream distributions to replicate. See [Filtering Distributions](#filtering-distributions-with-q_select). | | `policy` | Controls how replication manages local objects. One of `all`, `labeled`, or `nodelete`. See [Replication Policies](#replication-policies). Defaults to `all`. | +| `remote_policy` | Download policy for remotes created during replication. One of `immediate`, `on_demand`, or `streamed`. Distinct from `policy`. When unset, remotes use Pulp's default (`immediate`). | ## Running Replication @@ -151,7 +152,8 @@ pulp upstream-pulp replicate --upstream-pulp "my-upstream" ## Replication Policies The `policy` field controls how replication handles local objects, particularly when upstream -distributions are removed or no longer match a `q_select` filter. +distributions are removed or no longer match a `q_select` filter. It is not the same as a remote's +download policy (`immediate`, `on_demand`, or `streamed`); set that with `remote_policy`. ### `all` (default) diff --git a/pulpcore/app/migrations/0157_upstreampulp_remote_policy.py b/pulpcore/app/migrations/0157_upstreampulp_remote_policy.py new file mode 100644 index 00000000000..2c6ddc91c1f --- /dev/null +++ b/pulpcore/app/migrations/0157_upstreampulp_remote_policy.py @@ -0,0 +1,34 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("core", "0156_alter_contentartifact_relative_path_and_more"), + ] + + operations = [ + migrations.AddField( + model_name="upstreampulp", + name="remote_policy", + field=models.TextField( + choices=[ + ("immediate", "When syncing, download all metadata and content now."), + ( + "on_demand", + "When syncing, download metadata, but do not download content now. " + "Instead, download content as clients request it, and save it in Pulp " + "to be served for future client requests.", + ), + ( + "streamed", + "When syncing, download metadata, but do not download content now. " + "Instead,download content as clients request it, but never save it in " + "Pulp. This causes future requests for that same content to have to be " + "downloaded again.", + ), + ], + null=True, + ), + ), + ] diff --git a/pulpcore/app/models/replica.py b/pulpcore/app/models/replica.py index 29b32c4d5c0..4bbd7fa73a4 100644 --- a/pulpcore/app/models/replica.py +++ b/pulpcore/app/models/replica.py @@ -11,6 +11,8 @@ from pulpcore.app.util import get_domain_pk from pulpcore.plugin.models import AutoAddObjPermsMixin, BaseModel, EncryptedTextField +from .repository import Remote + class UpstreamPulp(BaseModel, AutoAddObjPermsMixin): ALL = "all" @@ -59,6 +61,7 @@ class UpstreamPulp(BaseModel, AutoAddObjPermsMixin): sock_read_timeout = models.FloatField( null=True, validators=[MinValueValidator(0.0, "Timeout must be >= 0")] ) + remote_policy = models.TextField(choices=Remote.POLICY_CHOICES, null=True) q_select = models.TextField(null=True) policy = models.TextField(choices=POLICY_CHOICES, default=ALL) diff --git a/pulpcore/app/serializers/replica.py b/pulpcore/app/serializers/replica.py index 7d13425865f..e130ac8afee 100644 --- a/pulpcore/app/serializers/replica.py +++ b/pulpcore/app/serializers/replica.py @@ -3,7 +3,7 @@ from rest_framework import serializers from rest_framework.validators import UniqueValidator -from pulpcore.app.models import UpstreamPulp +from pulpcore.app.models import Remote, UpstreamPulp from pulpcore.app.serializers import ( HiddenFieldsMixin, IdentityField, @@ -122,6 +122,16 @@ class UpstreamPulpSerializer(ModelSerializer, HiddenFieldsMixin): ), min_value=0.0, ) + remote_policy = serializers.ChoiceField( + choices=Remote.POLICY_CHOICES, + help_text=_( + "Download policy for remotes created during replication. One of 'immediate', " + "'on_demand', or 'streamed'. Distinct from 'policy', which controls how replicate " + "manages local objects. Defaults to the Remote default ('immediate') when unset." + ), + required=False, + allow_null=True, + ) pulp_last_updated = serializers.DateTimeField( help_text="Timestamp of the most recent update of the remote.", read_only=True @@ -178,6 +188,7 @@ class Meta: "connect_timeout", "sock_connect_timeout", "sock_read_timeout", + "remote_policy", "pulp_last_updated", "hidden_fields", "q_select", diff --git a/pulpcore/app/tasks/replica.py b/pulpcore/app/tasks/replica.py index 19de2bd60ed..d8aad2f0471 100644 --- a/pulpcore/app/tasks/replica.py +++ b/pulpcore/app/tasks/replica.py @@ -52,6 +52,26 @@ def _ssl_temp_files(server): pass +def _build_remote_settings(server): + """Build fields copied onto remotes created during replication.""" + remote_settings = { + "ca_cert": server.ca_cert, + "tls_validation": server.tls_validation, + "client_cert": server.client_cert, + "client_key": server.client_key, + "download_concurrency": server.download_concurrency, + "max_retries": server.max_retries, + "total_timeout": server.total_timeout, + "connect_timeout": server.connect_timeout, + "sock_connect_timeout": server.sock_connect_timeout, + "sock_read_timeout": server.sock_read_timeout, + } + # Omit policy when unset so new remotes keep Remote.policy's default (immediate). + if (remote_policy := getattr(server, "remote_policy", None)) is not None: + remote_settings["policy"] = remote_policy + return remote_settings + + def replicate_distributions(server_pk, q_select=None, **kwargs): server = UpstreamPulp.objects.get(pk=server_pk) with _ssl_temp_files(server) as ssl_files: @@ -75,18 +95,7 @@ def replicate_distributions(server_pk, q_select=None, **kwargs): } ) - remote_settings = { - "ca_cert": server.ca_cert, - "tls_validation": server.tls_validation, - "client_cert": server.client_cert, - "client_key": server.client_key, - "download_concurrency": server.download_concurrency, - "max_retries": server.max_retries, - "total_timeout": server.total_timeout, - "connect_timeout": server.connect_timeout, - "sock_connect_timeout": server.sock_connect_timeout, - "sock_read_timeout": server.sock_read_timeout, - } + remote_settings = _build_remote_settings(server) try: task_group = TaskGroup.current() supported_replicators = [] diff --git a/pulpcore/tests/functional/api/test_replication.py b/pulpcore/tests/functional/api/test_replication.py index f208a4be082..a06d49fe3af 100644 --- a/pulpcore/tests/functional/api/test_replication.py +++ b/pulpcore/tests/functional/api/test_replication.py @@ -281,6 +281,80 @@ def test_replication_remote_settings_propagation( assert remote.max_retries == 2 +@pytest.mark.parallel +def test_replication_remote_policy( + domain_factory, + bindings_cfg, + pulpcore_bindings, + file_bindings, + monitor_task, + monitor_task_group, + pulp_settings, + gen_object_with_cleanup, + file_distribution_factory, + file_publication_factory, + file_repository_factory, + tmp_path, + add_domain_objects_to_cleanup, +): + """Remotes created by replicate() inherit UpstreamPulp.remote_policy when set.""" + source_domain = domain_factory() + add_domain_objects_to_cleanup(source_domain) + + repository = file_repository_factory(pulp_domain=source_domain.name) + file_path = tmp_path / "file.txt" + file_path.write_text("DEADBEEF") + monitor_task( + file_bindings.ContentFilesApi.create( + file=str(file_path), + relative_path="file.txt", + repository=repository.pulp_href, + pulp_domain=source_domain.name, + ).task + ) + publication = file_publication_factory( + pulp_domain=source_domain.name, repository=repository.pulp_href + ) + file_distribution_factory(pulp_domain=source_domain.name, publication=publication.pulp_href) + + replica_domain = domain_factory() + add_domain_objects_to_cleanup(replica_domain) + + upstream_pulp_body = { + "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, + "remote_policy": "on_demand", + } + upstream_pulp = gen_object_with_cleanup( + pulpcore_bindings.UpstreamPulpsApi, upstream_pulp_body, pulp_domain=replica_domain.name + ) + + response = pulpcore_bindings.UpstreamPulpsApi.replicate( + upstream_pulp.pulp_href, pulpcore_bindings.module.UpstreamPulpReplicate() + ) + monitor_task_group(response.task_group) + + result = file_bindings.RemotesFileApi.list(pulp_domain=replica_domain.name) + assert result.count == 1 + remote = result.results[0] + assert remote.policy == "on_demand" + + pulpcore_bindings.UpstreamPulpsApi.partial_update( + upstream_pulp.pulp_href, {"remote_policy": "streamed"} + ) + response = pulpcore_bindings.UpstreamPulpsApi.replicate( + upstream_pulp.pulp_href, pulpcore_bindings.module.UpstreamPulpReplicate() + ) + monitor_task_group(response.task_group) + + remote = file_bindings.RemotesFileApi.list(pulp_domain=replica_domain.name).results[0] + assert remote.policy == "streamed" + + @pytest.mark.parallel def test_replication_with_repo_based_distribution( domain_factory, diff --git a/pulpcore/tests/unit/test_replica_remote_policy.py b/pulpcore/tests/unit/test_replica_remote_policy.py new file mode 100644 index 00000000000..f5eea3d335a --- /dev/null +++ b/pulpcore/tests/unit/test_replica_remote_policy.py @@ -0,0 +1,35 @@ +from types import SimpleNamespace + +from pulpcore.app.tasks.replica import _build_remote_settings + + +def _server(**overrides): + base = { + "ca_cert": "api-ca", + "tls_validation": True, + "client_cert": "api-cert", + "client_key": "api-key", + "download_concurrency": 10, + "max_retries": 3, + "total_timeout": 30, + "connect_timeout": 5, + "sock_connect_timeout": 5, + "sock_read_timeout": 5, + "remote_policy": None, + } + base.update(overrides) + return SimpleNamespace(**base) + + +def test_build_remote_settings_omits_policy_when_unset(): + settings = _build_remote_settings(_server()) + + assert "policy" not in settings + assert settings["ca_cert"] == "api-ca" + assert settings["download_concurrency"] == 10 + + +def test_build_remote_settings_includes_policy_when_set(): + settings = _build_remote_settings(_server(remote_policy="on_demand")) + + assert settings["policy"] == "on_demand"