diff --git a/.github/workflows/scripts/func_test_script.sh b/.github/workflows/scripts/func_test_script.sh
new file mode 100755
index 00000000000..b42e73688f5
--- /dev/null
+++ b/.github/workflows/scripts/func_test_script.sh
@@ -0,0 +1,24 @@
+#!/usr/bin/env bash
+# Functional test runner with task-queue sampling right after parallel suites.
+# Sourced from script.sh via FUNC_TEST_SCRIPT.
+
+_nightly_args=()
+if [[ "${GITHUB_WORKFLOW:-}" =~ "Nightly" ]]; then
+ _nightly_args=(--nightly)
+fi
+
+_pytest_common=(-v --timeout=300 -r sx --color=yes --suppress-no-test-exit-code --durations=20)
+
+# Parallel suites first (xdist -n 8), then sample worker-queue waits before serial.
+cmd_user_prefix bash -c "pytest ${_pytest_common[*]} --pyargs pulpcore.tests.functional -m parallel -n 8 ${_nightly_args[*]}"
+cmd_user_prefix bash -c "pytest ${_pytest_common[*]} --pyargs pulp_file.tests.functional -m parallel -n 8 ${_nightly_args[*]}"
+cmd_user_prefix bash -c "pytest ${_pytest_common[*]} --pyargs pulp_certguard.tests.functional -m parallel -n 8 ${_nightly_args[*]}"
+
+echo "::group::Task queue wait stats (after parallel suites)"
+# Short window: capture tasks from the parallel phase before purge/serial dilute the signal.
+cmd_user_prefix pulpcore-manager task-queue-stats --hours 1 --top 20 || true
+echo "::endgroup::"
+
+cmd_user_prefix bash -c "pytest ${_pytest_common[*]} --pyargs pulpcore.tests.functional -m 'not parallel' ${_nightly_args[*]}"
+cmd_user_prefix bash -c "pytest ${_pytest_common[*]} --pyargs pulp_file.tests.functional -m 'not parallel' ${_nightly_args[*]}"
+cmd_user_prefix bash -c "pytest ${_pytest_common[*]} --pyargs pulp_certguard.tests.functional -m 'not parallel' ${_nightly_args[*]}"
diff --git a/CHANGES/+task-queue-stats.misc b/CHANGES/+task-queue-stats.misc
new file mode 100644
index 00000000000..f59c75f51dc
--- /dev/null
+++ b/CHANGES/+task-queue-stats.misc
@@ -0,0 +1 @@
+Add `pulpcore-manager task-queue-stats` to summarize worker-queue waits from `unblocked_at`.
diff --git a/CLAUDE.md b/CLAUDE.md
index fc276b00c41..6dbef3719fb 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -38,6 +38,8 @@ pulpcore & pulp-file functional tests require both client bindings to be install
**Always** use the `oci-env` to run the functional and unit tests.
+Do not UPDATE `Task.pulp_created`. A Postgres trigger (`on_update_timestamp_task`) raises `Updating pulp_created is not allowed.` Create rows in the desired order instead.
+
## Modifying template_config.yml
Use the `plugin-template` tool after any changes made to `template_config.yml`.
@@ -57,6 +59,6 @@ When patchback fails to cherry-pick a PR into an older branch, you need to manua
## Contributing
-All docs, code comments, and changelogs are in markdown format. Keep comments and changelogs short and concise. Try to keep changelogs to just one line.
+All docs, code comments, and changelogs are in markdown format. Keep comments and changelogs short and concise. Try to keep changelogs to just one line. Use single backticks for inline code (`like_this`), not RST double backticks (``like_this``).
When preparing to commit and create a PR you **must** follow our [PR checklist](https://pulpproject.org/pulpcore/docs/dev/guides/pull-request-walkthrough/) Important to note is the AI attribution requirement in our commit messages. Also, note that our changelog entries are markdown.
diff --git a/pulp_file/tests/functional/api/test_auto_publish.py b/pulp_file/tests/functional/api/test_auto_publish.py
index 1ad8cab3bc1..f242e97634f 100644
--- a/pulp_file/tests/functional/api/test_auto_publish.py
+++ b/pulp_file/tests/functional/api/test_auto_publish.py
@@ -1,5 +1,7 @@
"""Tests that sync file plugin repositories."""
+from uuid import uuid4
+
import pytest
from pulpcore.client.pulp_file import (
@@ -24,8 +26,10 @@ def test_auto_publish_and_distribution(
file_random_content_unit,
monitor_task,
has_pulp_plugin,
+ random_artifact_factory,
):
"""Tests auto-publish and auto-distribution"""
+ # Remote is only needed to assert mirror=True is rejected with autopublish.
remote = file_remote_ssl_factory(manifest_path=basic_manifest_path, policy="on_demand")
repo = file_bindings.RepositoriesFileApi.read(file_repo_with_auto_publish.pulp_href)
distribution = gen_object_with_cleanup(
@@ -44,12 +48,18 @@ def test_auto_publish_and_distribution(
)
assert distribution.publication is None
- # Check what content and artifacts are in the fixture repository
- expected_files = get_files_in_manifest(remote.url)
-
- # Sync from the remote
- body = FileRepositorySyncURL(remote=remote.pulp_href)
- monitor_task(file_bindings.RepositoriesFileApi.sync(repo.pulp_href, body).task)
+ # One content unit is enough for version 1; attaching it triggers autopublish.
+ artifact = random_artifact_factory()
+ relative_path = f"{uuid4()}.iso"
+ created = monitor_task(
+ file_bindings.ContentFilesApi.create(
+ artifact=artifact.pulp_href,
+ relative_path=relative_path,
+ repository=repo.pulp_href,
+ ).task
+ ).created_resources
+ content = file_bindings.ContentFilesApi.read(created[1] if len(created) > 1 else created[0])
+ expected_files = {(relative_path, content.sha256, str(artifact.size))}
repo = file_bindings.RepositoriesFileApi.read(repo.pulp_href)
# Assert that a new repository version was created and a publication was created
diff --git a/pulp_file/tests/functional/api/test_filesystem_export.py b/pulp_file/tests/functional/api/test_filesystem_export.py
index da92c1b471b..1caf5472691 100644
--- a/pulp_file/tests/functional/api/test_filesystem_export.py
+++ b/pulp_file/tests/functional/api/test_filesystem_export.py
@@ -174,6 +174,15 @@ def test_fsexport_by_version(
}
+def _filesystem_domain(pulpcore_bindings, gen_object_with_cleanup):
+ body = {
+ "name": str(uuid.uuid4()),
+ "storage_class": "pulpcore.app.models.storage.FileSystem",
+ "storage_settings": {"MEDIA_ROOT": "/var/lib/pulp/media/"},
+ }
+ return gen_object_with_cleanup(pulpcore_bindings.DomainsApi, body)
+
+
@pytest.mark.skipif(not settings.DOMAIN_ENABLED, reason="Domains not enabled.")
@pytest.mark.parallel
def test_fsexport_cross_domain(
@@ -181,40 +190,48 @@ def test_fsexport_cross_domain(
fs_export_factory,
gen_object_with_cleanup,
pulpcore_bindings,
- pub_and_repo,
+ file_bindings,
+ file_repository_factory,
+ file_publication_factory,
+ tmp_path,
+ monitor_task,
):
+ # Publication and versions live in source_domain; exporter lives in other_domain.
+ source_domain = _filesystem_domain(pulpcore_bindings, gen_object_with_cleanup)
+ other_domain = _filesystem_domain(pulpcore_bindings, gen_object_with_cleanup)
+
+ src = tmp_path / "file.dat"
+ src.write_text("x")
+ content_href = file_bindings.ContentFilesApi.upload(
+ relative_path="0.dat", file=str(src), pulp_domain=source_domain.name
+ ).pulp_href
+ repository = file_repository_factory(pulp_domain=source_domain.name)
+ monitor_task(
+ file_bindings.RepositoriesFileApi.modify(
+ repository.pulp_href, {"add_content_units": [content_href]}
+ ).task
+ )
+ repository = file_bindings.RepositoriesFileApi.read(repository.pulp_href)
+ publication = file_publication_factory(
+ repository=repository.pulp_href, pulp_domain=source_domain.name
+ )
+ latest = repository.latest_version_href
+ zeroth = latest.rsplit("/", 2)[0] + "/0/"
+ exporter = fs_exporter_factory(pulp_domain=other_domain.name)
- entities = [{}, {}]
- for e in entities:
- body = {
- "name": str(uuid.uuid4()),
- "storage_class": "pulpcore.app.models.storage.FileSystem",
- "storage_settings": {"MEDIA_ROOT": "/var/lib/pulp/media/"},
- }
- e["domain"] = gen_object_with_cleanup(pulpcore_bindings.DomainsApi, body)
- e["publication"], e["repository"] = pub_and_repo(pulp_domain=e["domain"].name)
- e["exporter"] = fs_exporter_factory(pulp_domain=e["domain"].name)
- body = {"publication": e["publication"].pulp_href}
- e["export"] = fs_export_factory(e["exporter"], body=body)
-
- latest = entities[0]["repository"].latest_version_href
- zeroth = latest.replace("/2/", "/0/")
-
- with pytest.raises(BadRequestException) as e:
- body = {"publication": entities[0]["publication"].pulp_href}
- fs_export_factory(entities[1]["exporter"], body=body)
+ with pytest.raises(BadRequestException):
+ fs_export_factory(exporter, body={"publication": publication.pulp_href})
- with pytest.raises(BadRequestException) as e:
- body = {"repository_version": latest}
- fs_export_factory(entities[1]["exporter"], body=body)
+ with pytest.raises(BadRequestException):
+ fs_export_factory(exporter, body={"repository_version": latest})
- with pytest.raises(BadRequestException) as e:
- body = {"repository_version": latest, "start_repository_version": zeroth}
- fs_export_factory(entities[1]["exporter"], body=body)
+ with pytest.raises(BadRequestException):
+ fs_export_factory(
+ exporter, body={"repository_version": latest, "start_repository_version": zeroth}
+ )
- with pytest.raises(BadRequestException) as e:
- body = {
- "publication": entities[0]["publication"].pulp_href,
- "start_repository_version": zeroth,
- }
- fs_export_factory(entities[1]["exporter"], body=body)
+ with pytest.raises(BadRequestException):
+ fs_export_factory(
+ exporter,
+ body={"publication": publication.pulp_href, "start_repository_version": zeroth},
+ )
diff --git a/pulp_file/tests/functional/api/test_mime_types.py b/pulp_file/tests/functional/api/test_mime_types.py
index d44ba9f5b7f..e39d94713fa 100644
--- a/pulp_file/tests/functional/api/test_mime_types.py
+++ b/pulp_file/tests/functional/api/test_mime_types.py
@@ -13,30 +13,36 @@ def test_content_types(
file_bindings,
distribution_base_url,
file_repo_with_auto_publish,
- file_content_unit_with_name_factory,
gen_object_with_cleanup,
monitor_task,
+ tmp_path,
):
"""Test if content-app correctly returns mime-types based on filenames."""
+ relative_paths = {
+ "tar.gz": f"{uuid.uuid4()}.tar.gz",
+ "xml.gz": f"{uuid.uuid4()}.xml.gz",
+ "xml.bz2": f"{uuid.uuid4()}.xml.bz2",
+ "xml.zstd": f"{uuid.uuid4()}.xml.zstd",
+ "xml.xz": f"{uuid.uuid4()}.xml.xz",
+ "json.zstd": f"{uuid.uuid4()}.json.zstd",
+ "json": f"{uuid.uuid4()}.json",
+ "txt": f"{uuid.uuid4()}.txt",
+ "xml": f"{uuid.uuid4()}.xml",
+ "jpg": f"{uuid.uuid4()}.jpg",
+ "JPG": f"{uuid.uuid4()}.JPG",
+ "halabala": f"{uuid.uuid4()}.halabala",
+ "noextension1": f"{uuid.uuid4()}.asd/.asd/a",
+ "noextension2": f"{uuid.uuid4()}.....f",
+ }
+
+ blob = tmp_path / "blob"
+ blob.write_bytes(b"mime-type-test")
files = {
- "tar.gz": file_content_unit_with_name_factory(f"{str(uuid.uuid4())}.tar.gz"),
- "xml.gz": file_content_unit_with_name_factory(f"{str(uuid.uuid4())}.xml.gz"),
- "xml.bz2": file_content_unit_with_name_factory(f"{str(uuid.uuid4())}.xml.bz2"),
- "xml.zstd": file_content_unit_with_name_factory(f"{str(uuid.uuid4())}.xml.zstd"),
- "xml.xz": file_content_unit_with_name_factory(f"{str(uuid.uuid4())}.xml.xz"),
- "json.zstd": file_content_unit_with_name_factory(f"{str(uuid.uuid4())}.json.zstd"),
- "json": file_content_unit_with_name_factory(f"{str(uuid.uuid4())}.json"),
- "txt": file_content_unit_with_name_factory(f"{str(uuid.uuid4())}.txt"),
- "xml": file_content_unit_with_name_factory(f"{str(uuid.uuid4())}.xml"),
- "jpg": file_content_unit_with_name_factory(f"{str(uuid.uuid4())}.jpg"),
- "JPG": file_content_unit_with_name_factory(f"{str(uuid.uuid4())}.JPG"),
- "halabala": file_content_unit_with_name_factory(f"{str(uuid.uuid4())}.halabala"),
- "noextension1": file_content_unit_with_name_factory(f"{str(uuid.uuid4())}.asd/.asd/a"),
- "noextension2": file_content_unit_with_name_factory(f"{str(uuid.uuid4())}.....f"),
+ extension: file_bindings.ContentFilesApi.upload(file=str(blob), relative_path=relative_path)
+ for extension, relative_path in relative_paths.items()
}
- units_to_add = list(map(lambda f: f.pulp_href, files.values()))
- data = RepositoryAddRemoveContent(add_content_units=units_to_add)
+ data = RepositoryAddRemoveContent(add_content_units=[f.pulp_href for f in files.values()])
monitor_task(
file_bindings.RepositoriesFileApi.modify(file_repo_with_auto_publish.pulp_href, data).task
)
@@ -49,18 +55,20 @@ def test_content_types(
distribution = gen_object_with_cleanup(file_bindings.DistributionsFileApi, data)
distribution_base_url = distribution_base_url(distribution.base_url)
- received_mimetypes = {}
- for extension, content_unit in files.items():
+ async def fetch_mimetypes():
+ async with aiohttp.ClientSession() as session:
- async def get_content_type():
- async with aiohttp.ClientSession() as session:
+ async def get_content_type(extension, content_unit):
url = urljoin(distribution_base_url, content_unit.relative_path)
async with session.get(url) as response:
- return response.headers.get("Content-Type")
+ return extension, response.headers.get("Content-Type")
- content_type = asyncio.run(get_content_type())
- received_mimetypes[extension] = content_type
+ pairs = await asyncio.gather(
+ *(get_content_type(ext, unit) for ext, unit in files.items())
+ )
+ return dict(pairs)
+ received_mimetypes = asyncio.run(fetch_mimetypes())
expected_mimetypes = {
"tar.gz": "application/gzip",
"xml.gz": "application/gzip",
diff --git a/pulp_file/tests/functional/api/test_pulp_export.py b/pulp_file/tests/functional/api/test_pulp_export.py
index 792c15812cb..7640a870a56 100644
--- a/pulp_file/tests/functional/api/test_pulp_export.py
+++ b/pulp_file/tests/functional/api/test_pulp_export.py
@@ -19,7 +19,7 @@
@pytest.fixture
def pulp_exporter_factory(
- tmpdir,
+ tmp_path_factory,
pulpcore_bindings,
gen_object_with_cleanup,
add_to_filesystem_cleanup,
@@ -31,7 +31,7 @@ def _pulp_exporter_factory(
if repositories is None:
repositories = []
name = str(uuid.uuid4())
- path = "{}/{}/".format(tmpdir, name)
+ path = "{}/{}/".format(tmp_path_factory.mktemp("exporter"), name)
body = {
"name": name,
"path": path,
@@ -82,7 +82,7 @@ def _pulp_export_factory(exporter, body=None):
return _pulp_export_factory
-@pytest.fixture
+@pytest.fixture(scope="class")
def three_synced_repositories(
file_bindings,
file_repository_factory,
@@ -101,7 +101,8 @@ def three_synced_repositories(
file_bindings.RepositoriesFileApi.sync(repository.pulp_href, {}).task
for repository in repositories
]
- [monitor_task(task) for task in sync_tasks]
+ for task in sync_tasks:
+ monitor_task(task)
repositories = [
file_bindings.RepositoriesFileApi.read(repository.pulp_href) for repository in repositories
]
@@ -132,13 +133,25 @@ def shallow_pulp_exporter(pulp_exporter_factory):
return pulp_exporter_factory()
-@pytest.fixture
+@pytest.fixture(scope="class")
def full_pulp_exporter(
- pulp_exporter_factory,
+ pulpcore_bindings,
+ tmp_path_factory,
+ gen_object_with_cleanup,
+ add_to_filesystem_cleanup,
three_synced_repositories,
):
- repositories = three_synced_repositories
- return pulp_exporter_factory(repositories=repositories)
+ """Build exporter inline so this class-scoped fixture need not depend on a function factory."""
+ name = str(uuid.uuid4())
+ path = "{}/{}/".format(tmp_path_factory.mktemp("full-exporter"), name)
+ body = {
+ "name": name,
+ "path": path,
+ "repositories": [r.pulp_href for r in three_synced_repositories],
+ }
+ exporter = gen_object_with_cleanup(pulpcore_bindings.ExportersPulpApi, body)
+ add_to_filesystem_cleanup(path)
+ return exporter
@pytest.mark.parallel
@@ -169,74 +182,143 @@ def test_crud_exporter(pulpcore_bindings, shallow_pulp_exporter, monitor_task):
pulpcore_bindings.ExportersPulpApi.read(exporter.pulp_href)
-@pytest.mark.parallel
-def test_export(pulpcore_bindings, pulp_export_factory, full_pulp_exporter, monitor_task):
- exporter = full_pulp_exporter
- assert len(exporter.repositories) == 3
+class TestSyncedRepoExport:
+ """Don't mark parallel, tests are shorter than setup."""
- # Test export
- export = pulp_export_factory(exporter)
+ def test_export(self, pulpcore_bindings, pulp_export_factory, full_pulp_exporter, monitor_task):
+ exporter = full_pulp_exporter
+ assert len(exporter.repositories) == 3
- # Test list and delete
- # export 2 more to test on
- export_href2, export_href3 = (
- monitor_task(
- pulpcore_bindings.ExportersPulpExportsApi.create(exporter.pulp_href, {}).task
- ).created_resources[0]
- for _ in range(2)
- )
- exports = pulpcore_bindings.ExportersPulpExportsApi.list(exporter.pulp_href).results
- assert len(exports) == 3
- pulpcore_bindings.ExportersPulpExportsApi.delete(export.pulp_href)
- pulpcore_bindings.ExportersPulpExportsApi.delete(export_href2)
- exports = pulpcore_bindings.ExportersPulpExportsApi.list(exporter.pulp_href).results
- assert len(exports) == 1
- pulpcore_bindings.ExportersPulpExportsApi.delete(export_href3)
- exports = pulpcore_bindings.ExportersPulpExportsApi.list(exporter.pulp_href).results
- assert len(exports) == 0
+ # Test export
+ export = pulp_export_factory(exporter)
+ # Test list and delete
+ # export 2 more to test on
+ export_href2, export_href3 = (
+ monitor_task(
+ pulpcore_bindings.ExportersPulpExportsApi.create(exporter.pulp_href, {}).task
+ ).created_resources[0]
+ for _ in range(2)
+ )
+ exports = pulpcore_bindings.ExportersPulpExportsApi.list(exporter.pulp_href).results
+ assert len(exports) == 3
+ pulpcore_bindings.ExportersPulpExportsApi.delete(export.pulp_href)
+ pulpcore_bindings.ExportersPulpExportsApi.delete(export_href2)
+ exports = pulpcore_bindings.ExportersPulpExportsApi.list(exporter.pulp_href).results
+ assert len(exports) == 1
+ pulpcore_bindings.ExportersPulpExportsApi.delete(export_href3)
+ exports = pulpcore_bindings.ExportersPulpExportsApi.list(exporter.pulp_href).results
+ assert len(exports) == 0
+
+ def test_export_by_version_and_chunked(
+ self,
+ pulp_exporter_factory,
+ pulp_export_factory,
+ three_synced_repositories,
+ ):
+ repositories = three_synced_repositories
+ latest_versions = [r.latest_version_href for r in repositories]
+ zeroth_versions = [v_href.replace("/1/", "/0/") for v_href in latest_versions]
-@pytest.mark.parallel
-def test_export_by_version_and_chunked(
- pulp_exporter_factory,
- pulp_export_factory,
- three_synced_repositories,
-):
- repositories = three_synced_repositories
- latest_versions = [r.latest_version_href for r in repositories]
- zeroth_versions = [v_href.replace("/1/", "/0/") for v_href in latest_versions]
-
- # exporter for one repo. specify one version
- exporter = pulp_exporter_factory(repositories=[repositories[0]])
- body = {"versions": [latest_versions[0]]}
- export = pulp_export_factory(exporter, body)
- assert export.exported_resources[0].endswith("/1/")
- body = {"versions": [zeroth_versions[0]]}
- export = pulp_export_factory(exporter, body)
- assert export.exported_resources[0].endswith("/0/")
-
- # exporter for one repo. specify one *wrong* version
- with pytest.raises(ApiException, match="must belong to"):
- body = {"versions": [latest_versions[1]]}
- pulp_export_factory(exporter, body)
+ # exporter for one repo. specify one version
+ exporter = pulp_exporter_factory(repositories=[repositories[0]])
+ body = {"versions": [latest_versions[0]]}
+ export = pulp_export_factory(exporter, body)
+ assert export.exported_resources[0].endswith("/1/")
+ body = {"versions": [zeroth_versions[0]]}
+ export = pulp_export_factory(exporter, body)
+ assert export.exported_resources[0].endswith("/0/")
+
+ # exporter for one repo. specify one *wrong* version
+ with pytest.raises(ApiException, match="must belong to"):
+ body = {"versions": [latest_versions[1]]}
+ pulp_export_factory(exporter, body)
+
+ # test chunked export
+ body = {"chunk_size": "250B"}
+ export = pulp_export_factory(exporter, body)
+ assert export.output_file_info is not None
+ assert len(export.output_file_info) > 1
+
+ # Create a new exporter with two repos
+ exporter = pulp_exporter_factory(repositories=[repositories[0], repositories[1]])
+ # exporter for two repos, specify one version
+ with pytest.raises(ApiException, match="does not match the number"):
+ body = {"versions": [latest_versions[0]]}
+ pulp_export_factory(exporter, body)
+
+ # exporter for two repos, specify one correct and one *wrong* version
+ with pytest.raises(ApiException, match="must belong to"):
+ body = {"versions": [latest_versions[0], latest_versions[2]]}
+ pulp_export_factory(exporter, body)
+
+ def test_export_with_meta(self, pulpcore_bindings, pulp_export_factory, full_pulp_exporter):
+ exporter = full_pulp_exporter
+ user_meta = {
+ "initiator": "ci",
+ "purpose": "export",
+ "checksum_type": "md5", # pulp should override only in TOC JSON
+ }
- # test chunked export
- body = {"chunk_size": "250B"}
- export = pulp_export_factory(exporter, body)
- assert export.output_file_info is not None
- assert len(export.output_file_info) > 1
+ export = pulp_export_factory(exporter, {"meta": user_meta})
- # Create a new exporter with two repos
- exporter = pulp_exporter_factory(repositories=[repositories[0], repositories[1]])
- # exporter for two repos, specify one version
- with pytest.raises(ApiException, match="does not match the number"):
- body = {"versions": [latest_versions[0]]}
- pulp_export_factory(exporter, body)
+ # toc_info contains exactly user meta (unmodified)
+ meta_info = export.toc_info.get("meta", {})
+ assert meta_info == user_meta
- # exporter for two repos, specify one correct and one *wrong* version
- with pytest.raises(ApiException, match="must belong to"):
- body = {"versions": [latest_versions[0], latest_versions[2]]}
- pulp_export_factory(exporter, body)
+ # Validate TOC JSON file content
+ toc_file_path = export.toc_info.get("file")
+ assert toc_file_path and isinstance(toc_file_path, str)
+
+ with open(toc_file_path, "r") as f:
+ toc_data = json.load(f)
+
+ meta_json = toc_data.get("meta", {})
+ assert meta_json.get("initiator") == "ci"
+ assert meta_json.get("purpose") == "export"
+ # overridden field check
+ assert meta_json.get("checksum_type") == "crc32"
+
+ def test_export_chunk_ordering_and_naming(
+ self,
+ pulp_exporter_factory,
+ pulp_export_factory,
+ three_synced_repositories,
+ ):
+ exporter = pulp_exporter_factory(repositories=[three_synced_repositories[0]])
+ chunk_size_bytes = 100
+ body = {"chunk_size": f"{chunk_size_bytes}B"}
+ export = pulp_export_factory(exporter, body)
+
+ all_paths = [Path(p) for p in export.output_file_info.keys()]
+ tar_chunks = [p for p in all_paths if ".tar." in p.name]
+
+ assert len(tar_chunks) > 1, f"Expected multiple chunks for {chunk_size_bytes}B limit."
+
+ for index, path in enumerate(tar_chunks):
+ expected_suffix = f"{index:04d}"
+
+ assert path.name.endswith(expected_suffix), (
+ f"Chunk {path} missing suffix {expected_suffix}"
+ )
+ assert path.exists(), f"Chunk file {path} was not found on disk."
+
+ if index < len(tar_chunks) - 1:
+ assert path.stat().st_size == chunk_size_bytes
+
+ toc_path = Path(export.toc_info["file"])
+ with toc_path.open("r", encoding="utf-8") as f:
+ toc_data = json.load(f)
+
+ toc_filenames = list(toc_data["files"].keys())
+ expected_filenames = [p.name for p in tar_chunks]
+
+ assert toc_filenames == expected_filenames, (
+ f"TOC order mismatch.\nExpected: {expected_filenames}\nActual: {toc_filenames}"
+ )
+
+ assert toc_data["meta"]["chunk_size"] == chunk_size_bytes
+ assert toc_data["meta"]["checksum_type"] == "crc32"
@pytest.mark.parallel
@@ -307,65 +389,66 @@ def test_export_incremental(
@pytest.mark.skipif(not settings.DOMAIN_ENABLED, reason="Domains not enabled.")
@pytest.mark.parallel
def test_cross_domain_exporter(
- basic_manifest_path,
file_bindings,
- file_remote_factory,
+ file_repository_factory,
gen_object_with_cleanup,
pulpcore_bindings,
pulp_export_factory,
pulp_exporter_factory,
monitor_task,
+ tmp_path,
):
- # Create two domains
- # In each, create and sync a repository, create and export an exporter
- # Attempt to create an exporter using the *other domain's* repo
- # Attempt to update the exporter using the *other domain's* repo and last_export
- # Use the exporter and attempt to export the *other domain's* repo-versions
-
- entities = [{}, {}]
- for e in entities:
+ # Source domain: one uploaded file, exporter, and export (needed for last_export).
+ # Target domain: empty repo + exporter. Same-domain sync/export in the target is unused.
+
+ def _domain():
body = {
"name": str(uuid.uuid4()),
"storage_class": "pulpcore.app.models.storage.FileSystem",
"storage_settings": {"MEDIA_ROOT": "/var/lib/pulp/media/"},
}
- e["domain"] = gen_object_with_cleanup(pulpcore_bindings.DomainsApi, body)
- remote = file_remote_factory(
- manifest_path=basic_manifest_path, policy="immediate", pulp_domain=e["domain"].name
- )
+ return gen_object_with_cleanup(pulpcore_bindings.DomainsApi, body)
+
+ source_domain = _domain()
+ target_domain = _domain()
+
+ src = tmp_path / "file.txt"
+ src.write_text("x")
+ content_href = file_bindings.ContentFilesApi.upload(
+ relative_path="file.txt", file=str(src), pulp_domain=source_domain.name
+ ).pulp_href
+ source_repo = file_repository_factory(pulp_domain=source_domain.name)
+ monitor_task(
+ file_bindings.RepositoriesFileApi.modify(
+ source_repo.pulp_href, {"add_content_units": [content_href]}
+ ).task
+ )
+ source_repo = file_bindings.RepositoriesFileApi.read(source_repo.pulp_href)
+ source_exporter = pulp_exporter_factory(repositories=[source_repo], pulp_domain=source_domain)
+ source_export = pulp_export_factory(source_exporter)
- repo_body = {"name": str(uuid.uuid4()), "remote": remote.pulp_href}
- e["repository"] = gen_object_with_cleanup(
- file_bindings.RepositoriesFileApi, repo_body, pulp_domain=e["domain"].name
- )
- task = file_bindings.RepositoriesFileApi.sync(e["repository"].pulp_href, {}).task
- monitor_task(task)
- e["repository"] = file_bindings.RepositoriesFileApi.read(e["repository"].pulp_href)
- e["exporter"] = pulp_exporter_factory(
- repositories=[e["repository"]], pulp_domain=e["domain"]
- )
- e["export"] = pulp_export_factory(e["exporter"])
+ other_repo = file_repository_factory(pulp_domain=target_domain.name)
+ other_exporter = pulp_exporter_factory(repositories=[other_repo], pulp_domain=target_domain)
- target_domain = entities[1]["domain"]
# cross-create
with pytest.raises(BadRequestException) as e:
- pulp_exporter_factory(repositories=[entities[0]["repository"]], pulp_domain=target_domain)
+ pulp_exporter_factory(repositories=[source_repo], pulp_domain=target_domain)
assert e.value.status == 400
assert json.loads(e.value.body) == {
"non_field_errors": [f"Objects must all be a part of the {target_domain.name} domain."]
}
# cross-update
- body = {"repositories": [entities[0]["repository"].pulp_href]}
+ body = {"repositories": [source_repo.pulp_href]}
with pytest.raises(BadRequestException) as e:
- pulpcore_bindings.ExportersPulpApi.partial_update(entities[1]["exporter"].pulp_href, body)
+ pulpcore_bindings.ExportersPulpApi.partial_update(other_exporter.pulp_href, body)
assert e.value.status == 400
assert json.loads(e.value.body) == {
"non_field_errors": [f"Objects must all be a part of the {target_domain.name} domain."]
}
- body = {"last_export": entities[0]["export"].pulp_href}
+ body = {"last_export": source_export.pulp_href}
with pytest.raises(BadRequestException) as e:
- pulpcore_bindings.ExportersPulpApi.partial_update(entities[1]["exporter"].pulp_href, body)
+ pulpcore_bindings.ExportersPulpApi.partial_update(other_exporter.pulp_href, body)
assert e.value.status == 400
assert json.loads(e.value.body) == {
"non_field_errors": [f"Objects must all be a part of the {target_domain.name} domain."]
@@ -373,14 +456,14 @@ def test_cross_domain_exporter(
# cross-export
with pytest.raises(BadRequestException) as e:
- latest_v = entities[0]["repository"].latest_version_href
- zero_v = latest_v.replace("/1/", "/0/")
+ latest_v = source_repo.latest_version_href
+ zero_v = latest_v.rsplit("/", 2)[0] + "/0/"
body = {
"start_versions": [latest_v],
"versions": [zero_v],
"full": False,
}
- pulp_export_factory(entities[1]["exporter"], body)
+ pulp_export_factory(other_exporter, body)
assert e.value.status == 400
msgs = json.loads(e.value.body)
assert "versions" in msgs
@@ -391,72 +474,3 @@ def test_cross_domain_exporter(
assert msgs["start_versions"] == [
"Requested RepositoryVersions must belong to the Repositories named by the Exporter!"
]
-
-
-@pytest.mark.parallel
-def test_export_with_meta(pulpcore_bindings, pulp_export_factory, full_pulp_exporter):
- exporter = full_pulp_exporter
- user_meta = {
- "initiator": "ci",
- "purpose": "export",
- "checksum_type": "md5", # pulp should override only in TOC JSON
- }
-
- export = pulp_export_factory(exporter, {"meta": user_meta})
-
- # toc_info contains exactly user meta (unmodified)
- meta_info = export.toc_info.get("meta", {})
- assert meta_info == user_meta
-
- # Validate TOC JSON file content
- toc_file_path = export.toc_info.get("file")
- assert toc_file_path and isinstance(toc_file_path, str)
-
- with open(toc_file_path, "r") as f:
- toc_data = json.load(f)
-
- meta_json = toc_data.get("meta", {})
- assert meta_json.get("initiator") == "ci"
- assert meta_json.get("purpose") == "export"
- # overridden field check
- assert meta_json.get("checksum_type") == "crc32"
-
-
-@pytest.mark.parallel
-def test_export_chunk_ordering_and_naming(
- pulp_exporter_factory,
- pulp_export_factory,
- three_synced_repositories,
-):
- exporter = pulp_exporter_factory(repositories=[three_synced_repositories[0]])
- chunk_size_bytes = 100
- body = {"chunk_size": f"{chunk_size_bytes}B"}
- export = pulp_export_factory(exporter, body)
-
- all_paths = [Path(p) for p in export.output_file_info.keys()]
- tar_chunks = [p for p in all_paths if ".tar." in p.name]
-
- assert len(tar_chunks) > 1, f"Expected multiple chunks for {chunk_size_bytes}B limit."
-
- for index, path in enumerate(tar_chunks):
- expected_suffix = f"{index:04d}"
-
- assert path.name.endswith(expected_suffix), f"Chunk {path} missing suffix {expected_suffix}"
- assert path.exists(), f"Chunk file {path} was not found on disk."
-
- if index < len(tar_chunks) - 1:
- assert path.stat().st_size == chunk_size_bytes
-
- toc_path = Path(export.toc_info["file"])
- with toc_path.open("r", encoding="utf-8") as f:
- toc_data = json.load(f)
-
- toc_filenames = list(toc_data["files"].keys())
- expected_filenames = [p.name for p in tar_chunks]
-
- assert toc_filenames == expected_filenames, (
- f"TOC order mismatch.\nExpected: {expected_filenames}\nActual: {toc_filenames}"
- )
-
- assert toc_data["meta"]["chunk_size"] == chunk_size_bytes
- assert toc_data["meta"]["checksum_type"] == "crc32"
diff --git a/pulpcore/app/management/commands/task-queue-stats.py b/pulpcore/app/management/commands/task-queue-stats.py
new file mode 100644
index 00000000000..ecab536ef22
--- /dev/null
+++ b/pulpcore/app/management/commands/task-queue-stats.py
@@ -0,0 +1,253 @@
+"""
+Report task queue wait statistics from `unblocked_at` / `started_at`.
+
+Worker-queue wait is `started_at - unblocked_at`: the task was ready to run
+(resources free) but no worker had picked it up yet. Resource wait is
+`unblocked_at - pulp_created`.
+
+Only meaningful for `WORKER_TYPE=pulpcore`. Redis workers do not use the
+unblock mechanism, so those fields are not a reliable congestion signal there.
+"""
+
+import json
+from datetime import timedelta
+from gettext import gettext as _
+
+from django.conf import settings
+from django.core.management import BaseCommand
+from django.db.models import F, FloatField, Func
+from django.utils import timezone
+from django.utils.dateparse import parse_datetime
+
+from pulpcore.app.models import AppStatus, Task
+from pulpcore.constants import TASK_STATES
+
+
+class EpochSeconds(Func):
+ """PostgreSQL `EXTRACT(EPOCH FROM …)` as a float (subsecond precision)."""
+
+ function = "EXTRACT"
+ template = "%(function)s(EPOCH FROM %(expressions)s)"
+ output_field = FloatField()
+
+
+def _percentile(sorted_values, pct):
+ if not sorted_values:
+ return None
+ if len(sorted_values) == 1:
+ return sorted_values[0]
+ idx = min(len(sorted_values) - 1, max(0, round(pct / 100 * (len(sorted_values) - 1))))
+ return sorted_values[idx]
+
+
+def _fmt_seconds(value):
+ if value is None:
+ return "n/a"
+ return f"{value:.3f}s"
+
+
+def _summarize(values):
+ if not values:
+ return {"n": 0, "mean": None, "p50": None, "p90": None, "p99": None, "max": None}
+ return {
+ "n": len(values),
+ "mean": sum(values) / len(values),
+ "p50": _percentile(values, 50),
+ "p90": _percentile(values, 90),
+ "p99": _percentile(values, 99),
+ "max": values[-1],
+ }
+
+
+class Command(BaseCommand):
+ help = _("Summarize how long completed tasks waited for a worker after becoming unblocked.")
+
+ def add_arguments(self, parser):
+ parser.add_argument(
+ "--hours",
+ type=float,
+ default=None,
+ help=_("Only include tasks created in the last N hours."),
+ )
+ parser.add_argument(
+ "--since",
+ type=str,
+ default=None,
+ help=_("Only include tasks created at or after this ISO-8601 timestamp."),
+ )
+ parser.add_argument(
+ "--top",
+ type=int,
+ default=15,
+ help=_("Show the N task names with the highest mean worker wait (default: 15)."),
+ )
+ parser.add_argument(
+ "--min-worker-wait",
+ type=float,
+ default=0.0,
+ help=_("Only include tasks whose worker wait is at least this many seconds."),
+ )
+ parser.add_argument(
+ "--json",
+ action="store_true",
+ help=_("Emit machine-readable JSON instead of a text report."),
+ )
+
+ def handle(self, *args, **options):
+ worker_type = getattr(settings, "WORKER_TYPE", "pulpcore")
+ online_workers = AppStatus.objects.online().filter(app_type="worker").count()
+
+ qs = Task.objects.filter(
+ state=TASK_STATES.COMPLETED,
+ unblocked_at__isnull=False,
+ started_at__isnull=False,
+ finished_at__isnull=False,
+ )
+
+ since = None
+ if options["since"]:
+ since = parse_datetime(options["since"])
+ if since is None:
+ self.stderr.write(self.style.ERROR(f"Invalid --since value: {options['since']}"))
+ return
+ if timezone.is_naive(since):
+ since = timezone.make_aware(since, timezone.get_current_timezone())
+ elif options["hours"] is not None:
+ since = timezone.now() - timedelta(hours=options["hours"])
+
+ if since is not None:
+ qs = qs.filter(pulp_created__gte=since)
+
+ qs = qs.annotate(
+ worker_wait=EpochSeconds(F("started_at") - F("unblocked_at")),
+ resource_wait=EpochSeconds(F("unblocked_at") - F("pulp_created")),
+ total_wait=EpochSeconds(F("started_at") - F("pulp_created")),
+ runtime=EpochSeconds(F("finished_at") - F("started_at")),
+ )
+
+ min_wait = options["min_worker_wait"]
+ if min_wait:
+ qs = qs.filter(worker_wait__gte=min_wait)
+
+ rows = list(qs.values_list("name", "worker_wait", "resource_wait", "total_wait", "runtime"))
+
+ worker_waits = sorted(r[1] for r in rows)
+ resource_waits = sorted(r[2] for r in rows)
+ total_waits = sorted(r[3] for r in rows)
+ runtimes = sorted(r[4] for r in rows)
+
+ by_name = {}
+ for name, worker_wait, resource_wait, total_wait, runtime in rows:
+ bucket = by_name.setdefault(
+ name, {"worker": [], "resource": [], "total": [], "runtime": []}
+ )
+ bucket["worker"].append(worker_wait)
+ bucket["resource"].append(resource_wait)
+ bucket["total"].append(total_wait)
+ bucket["runtime"].append(runtime)
+
+ top_n = options["top"]
+ top_names = sorted(
+ (
+ {
+ "name": name,
+ "n": len(stats["worker"]),
+ "worker_wait": _summarize(sorted(stats["worker"])),
+ "resource_wait": _summarize(sorted(stats["resource"])),
+ "total_wait": _summarize(sorted(stats["total"])),
+ "runtime": _summarize(sorted(stats["runtime"])),
+ }
+ for name, stats in by_name.items()
+ ),
+ key=lambda item: (
+ item["worker_wait"]["mean"] if item["worker_wait"]["mean"] is not None else -1,
+ item["n"],
+ ),
+ reverse=True,
+ )[:top_n]
+
+ null_unblocked = Task.objects.filter(state=TASK_STATES.COMPLETED, unblocked_at__isnull=True)
+ if since is not None:
+ null_unblocked = null_unblocked.filter(pulp_created__gte=since)
+ null_unblocked_count = null_unblocked.count()
+
+ currently_waiting_unblocked = None
+ currently_waiting_blocked = None
+ if worker_type == "pulpcore":
+ currently_waiting_unblocked = Task.objects.filter(
+ state=TASK_STATES.WAITING, unblocked_at__isnull=False
+ ).count()
+ currently_waiting_blocked = Task.objects.filter(
+ state=TASK_STATES.WAITING, unblocked_at__isnull=True
+ ).count()
+
+ report = {
+ "worker_type": worker_type,
+ "online_workers": online_workers,
+ "since": since.isoformat() if since else None,
+ "min_worker_wait": min_wait,
+ "completed_with_null_unblocked_at": null_unblocked_count,
+ "currently_waiting_unblocked": currently_waiting_unblocked,
+ "currently_waiting_blocked": currently_waiting_blocked,
+ "worker_wait": _summarize(worker_waits),
+ "resource_wait": _summarize(resource_waits),
+ "total_wait": _summarize(total_waits),
+ "runtime": _summarize(runtimes),
+ "top_by_mean_worker_wait": top_names,
+ "notes": [],
+ }
+
+ if worker_type != "pulpcore":
+ report["notes"].append(
+ "WORKER_TYPE is not pulpcore; unblocked_at is not maintained by Redis workers, "
+ "so worker_wait is not a reliable congestion signal."
+ )
+ if null_unblocked_count:
+ report["notes"].append(
+ f"{null_unblocked_count} completed task(s) have null unblocked_at "
+ "(common under Redis workers)."
+ )
+
+ if options["json"]:
+ self.stdout.write(json.dumps(report, indent=2, default=str))
+ return
+
+ self.stdout.write("Task queue wait stats")
+ self.stdout.write(f" worker_type: {worker_type}")
+ self.stdout.write(f" online_workers: {online_workers}")
+ self.stdout.write(f" since: {since.isoformat() if since else 'all completed tasks'}")
+ if min_wait:
+ self.stdout.write(f" min_worker_wait filter: {min_wait}s")
+ self.stdout.write(
+ f" currently waiting (unblocked/blocked): "
+ f"{currently_waiting_unblocked}/{currently_waiting_blocked}"
+ )
+ self.stdout.write(f" completed with null unblocked_at: {null_unblocked_count}")
+ self.stdout.write("")
+
+ def print_summary(label, summary):
+ self.stdout.write(
+ f"{label}: n={summary['n']} mean={_fmt_seconds(summary['mean'])} "
+ f"p50={_fmt_seconds(summary['p50'])} p90={_fmt_seconds(summary['p90'])} "
+ f"p99={_fmt_seconds(summary['p99'])} max={_fmt_seconds(summary['max'])}"
+ )
+
+ print_summary("WORKER_WAIT (started_at - unblocked_at)", report["worker_wait"])
+ print_summary("RESOURCE_WAIT(unblocked_at - pulp_created)", report["resource_wait"])
+ print_summary("TOTAL_WAIT (started_at - pulp_created)", report["total_wait"])
+ print_summary("RUNTIME (finished_at - started_at)", report["runtime"])
+
+ if top_names:
+ self.stdout.write("")
+ self.stdout.write(f"Top {len(top_names)} task names by mean worker wait:")
+ for item in top_names:
+ ww = item["worker_wait"]
+ short = item["name"].rsplit(".", 1)[-1]
+ self.stdout.write(
+ f" mean={_fmt_seconds(ww['mean']):>8} p90={_fmt_seconds(ww['p90']):>8} "
+ f"max={_fmt_seconds(ww['max']):>8} n={ww['n']:<5} {short} ({item['name']})"
+ )
+
+ for note in report["notes"]:
+ self.stdout.write("")
+ self.stdout.write(self.style.WARNING(f"Note: {note}"))
diff --git a/pulpcore/tests/functional/api/test_api_root_rewrite.py b/pulpcore/tests/functional/api/test_api_root_rewrite.py
index 69c8acc17ce..11cc557f62a 100644
--- a/pulpcore/tests/functional/api/test_api_root_rewrite.py
+++ b/pulpcore/tests/functional/api/test_api_root_rewrite.py
@@ -1,6 +1,8 @@
+import asyncio
import json
import uuid
+import aiohttp
import pytest
"""
@@ -49,14 +51,24 @@ def auth_headers(bindings):
def test_list_endpoints(file_bindings, proxy_rewrite_set, pulp_api_v3_path):
"""Check that ALL rewritten API_ROOT endpoints are accessible."""
API_ROOT = pulp_api_v3_path.encode("utf-8")
- for endpoint, url in proxy_rewrite_set.items():
- headers = auth_headers(file_bindings)
- response = file_bindings.client.rest_client.request("GET", url, headers=headers)
- assert response.status == 200
+ headers = auth_headers(file_bindings)
+
+ async def fetch_all():
+ async with aiohttp.ClientSession(headers=headers) as session:
+
+ async def get_one(endpoint, url):
+ async with session.get(url, params={"limit": 1}, ssl=False) as response:
+ return endpoint, url, response.status, await response.read()
+
+ return await asyncio.gather(
+ *(get_one(endpoint, url) for endpoint, url in proxy_rewrite_set.items())
+ )
+ for endpoint, url, status, body in asyncio.run(fetch_all()):
+ assert status == 200, f"failed on {endpoint}:{url}"
if endpoint != "tasks":
# Tasks reserved resources can have original API_ROOT
- assert API_ROOT not in response.response.data, f"failed on {endpoint}:{url}"
+ assert API_ROOT not in body, f"failed on {endpoint}:{url}"
@pytest.mark.parallel
diff --git a/pulpcore/tests/functional/api/test_replication.py b/pulpcore/tests/functional/api/test_replication.py
index 17ab460363f..f208a4be082 100644
--- a/pulpcore/tests/functional/api/test_replication.py
+++ b/pulpcore/tests/functional/api/test_replication.py
@@ -293,21 +293,26 @@ def test_replication_with_repo_based_distribution(
gen_object_with_cleanup,
file_distribution_factory,
file_repository_factory,
- file_remote_factory,
- basic_manifest_path,
add_domain_objects_to_cleanup,
+ tmp_path,
):
"""Test replication when upstream distribution uses repository (not publication)."""
source_domain = domain_factory()
add_domain_objects_to_cleanup(source_domain)
- # Create a repo, sync it w/ mirror=True, and distribute via repository (not publication)
- remote = file_remote_factory(
- pulp_domain=source_domain.name, manifest_path=basic_manifest_path, policy="immediate"
+ src = tmp_path / "file.txt"
+ src.write_text("repo-based")
+ content_href = file_bindings.ContentFilesApi.upload(
+ relative_path="file.txt",
+ file=str(src),
+ pulp_domain=source_domain.name,
+ ).pulp_href
+ repo = file_repository_factory(pulp_domain=source_domain.name, autopublish=True)
+ monitor_task(
+ file_bindings.RepositoriesFileApi.modify(
+ repo.pulp_href, {"add_content_units": [content_href]}
+ ).task
)
- repo = file_repository_factory(pulp_domain=source_domain.name)
- sync_data = file_bindings.module.FileRepositorySyncURL(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
@@ -369,22 +374,28 @@ def test_replication_multi_distribution_content_update(
source_domain = domain_factory()
add_domain_objects_to_cleanup(source_domain)
- # Create 3 repos with content and publication-based distributions
+ # Create 2 repos with content and publication-based distributions
distros = []
repos = []
- for i in range(3):
+ modify_tasks = []
+ for i in range(2):
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,
+ content_href = file_bindings.ContentFilesApi.upload(
+ file=str(file_path),
+ relative_path=f"file_{i}.txt",
+ pulp_domain=source_domain.name,
+ ).pulp_href
+ modify_tasks.append(
+ file_bindings.RepositoriesFileApi.modify(
+ repo.pulp_href, {"add_content_units": [content_href]}
).task
)
+ for task in modify_tasks:
+ monitor_task(task)
+ for repo in repos:
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)
@@ -414,7 +425,7 @@ def test_replication_multi_distribution_content_update(
replica_distros = file_bindings.DistributionsFileApi.list(
pulp_domain=replica_domain.name
).results
- assert len(replica_distros) == 3
+ assert len(replica_distros) == 2
initial_versions = {}
for rd in replica_distros:
assert rd.repository is None
@@ -423,17 +434,23 @@ def test_replication_multi_distribution_content_update(
initial_versions[rd.name] = rd.repository_version
# Add new content to all source repos and update publications
+ modify_tasks = []
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,
+ content_href = file_bindings.ContentFilesApi.upload(
+ file=str(file_path),
+ relative_path=f"file_{i}_v2.txt",
+ pulp_domain=source_domain.name,
+ ).pulp_href
+ modify_tasks.append(
+ file_bindings.RepositoriesFileApi.modify(
+ repo.pulp_href, {"add_content_units": [content_href]}
).task
)
+ for task in modify_tasks:
+ monitor_task(task)
+ for i, repo in enumerate(repos):
repo = file_bindings.RepositoriesFileApi.read(repo.pulp_href)
pub = file_publication_factory(
pulp_domain=source_domain.name,
@@ -455,7 +472,7 @@ def test_replication_multi_distribution_content_update(
replica_distros = file_bindings.DistributionsFileApi.list(
pulp_domain=replica_domain.name
).results
- assert len(replica_distros) == 3
+ assert len(replica_distros) == 2
for rd in replica_distros:
assert rd.repository is None
assert rd.repository_version is not None
@@ -561,10 +578,8 @@ def test_replication_optimization(
pulp_settings,
file_bindings,
file_repository_factory,
- file_remote_factory,
file_distribution_factory,
file_publication_factory,
- basic_manifest_path,
monitor_task,
gen_object_with_cleanup,
tmp_path,
@@ -583,19 +598,20 @@ def test_replication_optimization(
pulpcore_bindings.UpstreamPulpsApi, upstream_pulp_body, pulp_domain=non_default_domain.name
)
- # sync a repository on the "remote" Pulp instance
- upstream_remote = file_remote_factory(
- pulp_domain=source_domain.name, manifest_path=basic_manifest_path, policy="immediate"
- )
+ # One content unit on the "remote" Pulp instance is enough to test skip-sync
+ src = tmp_path / "file.txt"
+ src.write_text("replica")
+ content_href = file_bindings.ContentFilesApi.upload(
+ relative_path="file.txt",
+ file=str(src),
+ pulp_domain=source_domain.name,
+ ).pulp_href
upstream_repository = file_repository_factory(pulp_domain=source_domain.name)
-
- repository_sync_data = file_bindings.module.FileRepositorySyncURL(
- remote=upstream_remote.pulp_href, mirror=True
- )
- response = file_bindings.RepositoriesFileApi.sync(
- upstream_repository.pulp_href, repository_sync_data
+ monitor_task(
+ file_bindings.RepositoriesFileApi.modify(
+ upstream_repository.pulp_href, {"add_content_units": [content_href]}
+ ).task
)
- monitor_task(response.task)
upstream_repository = file_bindings.RepositoriesFileApi.read(upstream_repository.pulp_href)
upstream_publication = file_publication_factory(
pulp_domain=source_domain.name, repository_version=upstream_repository.latest_version_href
@@ -1057,23 +1073,29 @@ def populate_upstream(
domain_factory,
file_bindings,
file_repository_factory,
- file_remote_factory,
file_distribution_factory,
- write_3_iso_file_fixture_data_factory,
monitor_task,
+ tmp_path,
):
def _populate_upstream(number, prefix=""):
upstream_domain = domain_factory()
+ src = tmp_path / f"{uuid.uuid4()}.txt"
+ src.write_text("replica")
+ content_href = file_bindings.ContentFilesApi.upload(
+ relative_path="file.txt",
+ file=str(src),
+ pulp_domain=upstream_domain.name,
+ ).pulp_href
tasks = []
for i in range(number):
repo = file_repository_factory(pulp_domain=upstream_domain.name, autopublish=True)
- name = f"{prefix}{i}"
- fix = write_3_iso_file_fixture_data_factory(name)
- remote = file_remote_factory(pulp_domain=upstream_domain.name, manifest_path=fix)
- body = {"remote": remote.pulp_href}
- tasks.append(file_bindings.RepositoriesFileApi.sync(repo.pulp_href, body).task)
+ tasks.append(
+ file_bindings.RepositoriesFileApi.modify(
+ repo.pulp_href, {"add_content_units": [content_href]}
+ ).task
+ )
file_distribution_factory(
- name=name,
+ name=f"{prefix}{i}",
pulp_domain=upstream_domain.name,
repository=repo.pulp_href,
pulp_labels={"upstream": str(i), "even" if i % 2 == 0 else "odd": ""},
@@ -1097,7 +1119,7 @@ def test_replicate_with_basic_q_select(
add_domain_objects_to_cleanup,
):
"""Test basic label select replication."""
- source_domain = populate_upstream(6)
+ source_domain = populate_upstream(4)
dest_domain = domain_factory()
upstream_body = {
"name": str(uuid.uuid4()),
@@ -1110,14 +1132,14 @@ def test_replicate_with_basic_q_select(
upstream = gen_object_with_cleanup(
pulpcore_bindings.UpstreamPulpsApi, upstream_body, pulp_domain=dest_domain.name
)
- # Run the replicate task and assert that all 6 repos got synced
+ # Run the replicate task and assert that all repos got synced
response = pulpcore_bindings.UpstreamPulpsApi.replicate(
upstream.pulp_href, pulpcore_bindings.module.UpstreamPulpReplicate()
)
monitor_task_group(response.task_group)
add_domain_objects_to_cleanup(dest_domain)
result = pulpcore_bindings.DistributionsApi.list(pulp_domain=dest_domain.name)
- assert result.count == 6
+ assert result.count == 4
# Update q_select to sync only 'even' repos
body = {"q_select": "pulp_label_select='even'"}
@@ -1127,11 +1149,11 @@ def test_replicate_with_basic_q_select(
)
monitor_task_group(response.task_group)
result = pulpcore_bindings.DistributionsApi.list(pulp_domain=dest_domain.name)
- assert result.count == 3
- assert {d.name for d in result.results} == {"0", "2", "4"}
+ assert result.count == 2
+ assert {d.name for d in result.results} == {"0", "2"}
# Update q_select to sync one 'upstream' repo
- body["q_select"] = "pulp_label_select='upstream=4'"
+ body["q_select"] = "pulp_label_select='upstream=2'"
pulpcore_bindings.UpstreamPulpsApi.partial_update(upstream.pulp_href, body)
response = pulpcore_bindings.UpstreamPulpsApi.replicate(
upstream.pulp_href, pulpcore_bindings.module.UpstreamPulpReplicate()
@@ -1139,7 +1161,7 @@ def test_replicate_with_basic_q_select(
monitor_task_group(response.task_group)
result = pulpcore_bindings.DistributionsApi.list(pulp_domain=dest_domain.name)
assert result.count == 1
- assert result.results[0].name == "4"
+ assert result.results[0].name == "2"
# Show that basic label select is ANDed together
body["q_select"] = "pulp_label_select='even,upstream=0'"
@@ -1165,7 +1187,7 @@ def test_replicate_with_per_request_q_select(
add_domain_objects_to_cleanup,
):
"""Test that q_select can be passed per-request to the replicate action."""
- source_domain = populate_upstream(6)
+ source_domain = populate_upstream(4)
dest_domain = domain_factory()
add_domain_objects_to_cleanup(dest_domain)
@@ -1192,8 +1214,8 @@ def test_replicate_with_per_request_q_select(
)
monitor_task_group(response.task_group)
result = pulpcore_bindings.DistributionsApi.list(pulp_domain=dest_domain.name)
- assert result.count == 3
- assert {d.name for d in result.results} == {"0", "2", "4"}
+ assert result.count == 2
+ assert {d.name for d in result.results} == {"0", "2"}
# Selective replicate of 'odd' should NOT delete the 'even' ones (remove_missing skipped)
replicate_body = pulpcore_bindings.module.UpstreamPulpReplicate(
@@ -1204,8 +1226,8 @@ def test_replicate_with_per_request_q_select(
)
monitor_task_group(response.task_group)
result = pulpcore_bindings.DistributionsApi.list(pulp_domain=dest_domain.name)
- assert result.count == 6
- assert {d.name for d in result.results} == {"0", "1", "2", "3", "4", "5"}
+ assert result.count == 4
+ assert {d.name for d in result.results} == {"0", "1", "2", "3"}
# Full replicate (no per-request q_select) should still work and run remove_missing
response = pulpcore_bindings.UpstreamPulpsApi.replicate(
@@ -1213,7 +1235,7 @@ def test_replicate_with_per_request_q_select(
)
monitor_task_group(response.task_group)
result = pulpcore_bindings.DistributionsApi.list(pulp_domain=dest_domain.name)
- assert result.count == 6
+ assert result.count == 4
@pytest.mark.parallel
@@ -1228,7 +1250,7 @@ def test_replicate_with_complex_q_select(
add_domain_objects_to_cleanup,
):
"""Test complex q_select replication."""
- source_domain = populate_upstream(6)
+ source_domain = populate_upstream(4)
dest_domain = domain_factory()
add_domain_objects_to_cleanup(dest_domain)
upstream_body = {
@@ -1252,16 +1274,16 @@ def test_replicate_with_complex_q_select(
assert result.count == 2
assert {d.name for d in result.results} == {"1", "2"}
- # Test odds but not five
- body = {"q_select": "pulp_label_select='odd' AND NOT pulp_label_select='upstream=5'"}
+ # Test odds but not three
+ body = {"q_select": "pulp_label_select='odd' AND NOT pulp_label_select='upstream=3'"}
pulpcore_bindings.UpstreamPulpsApi.partial_update(upstream.pulp_href, body)
response = pulpcore_bindings.UpstreamPulpsApi.replicate(
upstream.pulp_href, pulpcore_bindings.module.UpstreamPulpReplicate()
)
monitor_task_group(response.task_group)
result = pulpcore_bindings.DistributionsApi.list(pulp_domain=dest_domain.name)
- assert result.count == 2
- assert {d.name for d in result.results} == {"1", "3"}
+ assert result.count == 1
+ assert {d.name for d in result.results} == {"1"}
# Test we error when trying to provide an invalid q expression
body["q_select"] = "invalid='testing'"
@@ -1301,9 +1323,9 @@ def _add_domain_to_cleanup(domain):
@pytest.mark.parametrize(
"policy,results",
[
- ("nodelete", [{"b0", "b1", "a0", "a1", "a2"}, {"b0", "b1", "a0", "a1", "a2"}]),
- ("labeled", [{"b0", "b1", "a0", "a1", "a2"}, {"b0", "b1", "a0"}]),
- ("all", [{"a0", "a1", "a2"}, {"a0"}]),
+ ("nodelete", [{"b0", "a0", "a1"}, {"b0", "a0", "a1"}]),
+ ("labeled", [{"b0", "a0", "a1"}, {"b0", "a0"}]),
+ ("all", [{"a0", "a1"}, {"a0"}]),
],
)
def test_replicate_policy(
@@ -1320,8 +1342,8 @@ def test_replicate_policy(
gen_object_with_cleanup,
):
"""Test replicate delete_policy."""
- a_domain = populate_upstream(3, prefix="a")
- b_domain = populate_upstream(2, prefix="b")
+ a_domain = populate_upstream(2, prefix="a")
+ b_domain = populate_upstream(1, prefix="b")
upstream_body = {
"name": str(uuid.uuid4()),
"base_url": bindings_cfg.host,
@@ -1345,10 +1367,10 @@ def test_replicate_policy(
assert result.count == len(results[0])
assert {r.name for r in result.results} == results[0]
- # delete a1, a2
+ # delete a1
result = pulpcore_bindings.DistributionsApi.list(pulp_domain=a_domain.name)
- monitor_task(file_bindings.DistributionsFileApi.delete(result.results[0].pulp_href).task)
- monitor_task(file_bindings.DistributionsFileApi.delete(result.results[1].pulp_href).task)
+ a1 = next(d for d in result.results if d.name == "a1")
+ monitor_task(file_bindings.DistributionsFileApi.delete(a1.pulp_href).task)
result = pulpcore_bindings.DistributionsApi.list(pulp_domain=a_domain.name)
assert result.count == 1
assert result.results[0].name == "a0"
diff --git a/pulpcore/tests/functional/api/test_tasking.py b/pulpcore/tests/functional/api/test_tasking.py
index d9e95a56a0c..4b8135882d4 100644
--- a/pulpcore/tests/functional/api/test_tasking.py
+++ b/pulpcore/tests/functional/api/test_tasking.py
@@ -9,7 +9,6 @@
import pytest
from aiohttp import BasicAuth
-from django.conf import settings
from pulpcore.client.pulpcore import ApiException
from pulpcore.constants import IMMEDIATE_TIMEOUT
@@ -93,6 +92,10 @@ def test_worker_cleanup_on_missing_worker(dispatch_task, monitor_task, pulpcore_
Test that when a worker dies unexpectedly while executing a task,
the worker cleanup process marks the task as failed and releases its locks,
allowing subsequent tasks requiring the same resource to execute.
+
+ Prefer the unit test pulpcore.tests.unit.tasking.test_missing_worker_cleanup
+ for routine coverage of the cleanup path. This e2e test is long_running
+ (skipped when --timeout < 600) but still runs in nightly CI.
"""
# Use a unique resource identifier to avoid conflicts with other tests
resource = str(uuid4())
@@ -731,54 +734,3 @@ def test_times_out_on_task_worker(
)
monitor_task(task_href)
assert "timed out after" in ctx.value.task.error["description"]
-
-
-@pytest.mark.parallel
-@pytest.mark.skipif(
- settings.WORKER_TYPE != "redis",
- reason="Only runs with WORKER_TYPE=redis",
-)
-def test_fetch_task_beyond_initial_batch(dispatch_task, monitor_task, pulpcore_bindings):
- """Test that tasks beyond the initial fetch batch are still processed.
-
- When more than FETCH_TASK_LIMIT tasks are blocked on the same exclusive resource,
- the RedisWorker should double the fetch limit and find runnable tasks further
- down the queue.
- """
- blocker_resource = str(uuid4())
- other_resource = str(uuid4())
-
- # Dispatch a long-running task that holds the blocker resource
- blocker_href = dispatch_task(
- "pulpcore.app.tasks.test.sleep",
- args=(60,),
- exclusive_resources=[blocker_resource],
- )
- time.sleep(2)
-
- # Dispatch 25 tasks that all need the same blocked resource
- blocked_hrefs = []
- for _ in range(25):
- href = dispatch_task(
- "pulpcore.app.tasks.test.sleep",
- args=(0,),
- exclusive_resources=[blocker_resource],
- )
- blocked_hrefs.append(href)
-
- # Dispatch a task that uses a completely different resource (position 27 in the queue)
- unblocked_href = dispatch_task(
- "pulpcore.app.tasks.test.sleep",
- args=(0,),
- exclusive_resources=[other_resource],
- )
-
- # The unblocked task should complete even though 25 tasks ahead of it are blocked
- unblocked_task = monitor_task(unblocked_href)
- assert unblocked_task.state == "completed"
-
- # Cancel the blocker so blocked tasks can drain
- try:
- pulpcore_bindings.TasksApi.tasks_cancel(blocker_href, {"state": "canceled"})
- except ApiException:
- pass
diff --git a/pulpcore/tests/functional/api/using_plugin/test_checkpoint.py b/pulpcore/tests/functional/api/using_plugin/test_checkpoint.py
index b3844d7e699..305b650071f 100644
--- a/pulpcore/tests/functional/api/using_plugin/test_checkpoint.py
+++ b/pulpcore/tests/functional/api/using_plugin/test_checkpoint.py
@@ -2,7 +2,7 @@
import re
import uuid
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
from time import sleep
from urllib.parse import urlparse
@@ -13,25 +13,47 @@
from pulpcore.content.handler import Handler
+def _wait_until_checkpoint_ts_advances(previous_created):
+ """Block until wall-clock formats to a later checkpoint timestamp than previous_created.
+
+ Checkpoint URLs are second-resolution, so consecutive publications need distinct
+ seconds. Fixed sleep(1) is wasteful when create/publish already crossed a second.
+ """
+ previous_ts = Handler._format_checkpoint_timestamp(previous_created)
+ while Handler._format_checkpoint_timestamp(datetime.now(timezone.utc)) == previous_ts:
+ sleep(0.05)
+
+
@pytest.fixture(scope="class")
-def content_factory(tmp_path_factory, file_bindings, monitor_task):
+def content_factory(tmp_path_factory, file_bindings):
def _content_factory(name):
file = tmp_path_factory.mktemp("content") / name
file.write_text(str(uuid.uuid4()))
- return monitor_task(
- file_bindings.ContentFilesApi.create(relative_path=name, file=str(file)).task
- ).created_resources[0]
+ return file_bindings.ContentFilesApi.upload(relative_path=name, file=str(file)).pulp_href
+
+ def _precreate(names):
+ return [_content_factory(name) for name in names]
+ _content_factory.precreate = _precreate
return _content_factory
@pytest.fixture(scope="class")
def create_publication(content_factory, file_bindings, monitor_task):
counter = [0]
+ content_queue = []
+
+ def precreate(n):
+ names = []
+ for _ in range(n):
+ names.append(str(counter[0]))
+ counter[0] += 1
+ content_queue.extend(content_factory.precreate(names))
def _create_publication(repo, checkpoint):
- content_href = content_factory(f"{counter[0]}")
- counter[0] += 1
+ if not content_queue:
+ precreate(1)
+ content_href = content_queue.pop(0)
monitor_task(
file_bindings.RepositoriesFileApi.modify(
@@ -46,6 +68,7 @@ def _create_publication(repo, checkpoint):
)
return file_bindings.PublicationsFileApi.read(response.created_resources[0])
+ _create_publication.precreate = precreate
return _create_publication
@@ -58,16 +81,14 @@ def setup(
repo = file_repository_factory()
distribution = file_distribution_factory(repository=repo.pulp_href, checkpoint=True)
+ # Five publications: content creates overlap; only wait between pubs when needed
+ # for distinct second-resolution checkpoint timestamps.
+ create_publication.precreate(5)
pubs = []
- pubs.append(create_publication(repo, False))
- sleep(1)
- pubs.append(create_publication(repo, True))
- sleep(1)
- pubs.append(create_publication(repo, False))
- sleep(1)
- pubs.append(create_publication(repo, True))
- sleep(1)
- pubs.append(create_publication(repo, False))
+ for checkpoint in (False, True, False, True, False):
+ if pubs:
+ _wait_until_checkpoint_ts_advances(pubs[-1].pulp_created)
+ pubs.append(create_publication(repo, checkpoint))
return pubs, distribution
@@ -82,7 +103,8 @@ def _checkpoint_url(distribution, timestamp):
class TestCheckpointDistribution:
- @pytest.mark.parallel
+ """Don't mark parallel, tests are shorter than setup."""
+
def test_base_path_lists_checkpoints(self, setup, http_get, distribution_base_url):
pubs, distribution = setup
@@ -93,7 +115,6 @@ def test_base_path_lists_checkpoints(self, setup, http_get, distribution_base_ur
assert Handler._format_checkpoint_timestamp(pubs[1].pulp_created) in checkpoints_ts
assert Handler._format_checkpoint_timestamp(pubs[3].pulp_created) in checkpoints_ts
- @pytest.mark.parallel
def test_distro_root_no_trailing_slash_is_redirected(
self,
setup,
@@ -112,7 +133,6 @@ def test_distro_root_no_trailing_slash_is_redirected(
assert Handler._format_checkpoint_timestamp(pubs[1].pulp_created) in checkpoints_ts
assert Handler._format_checkpoint_timestamp(pubs[3].pulp_created) in checkpoints_ts
- @pytest.mark.parallel
def test_timestamped_checkpoint_no_trailing_slash_is_redirected(
self,
setup,
@@ -128,7 +148,6 @@ def test_timestamped_checkpoint_no_trailing_slash_is_redirected(
assert f"Index of {urlparse(pub_1_url).path}
" in response
- @pytest.mark.parallel
def test_exact_timestamp_is_served(self, setup, http_get, checkpoint_url):
pubs, distribution = setup
@@ -137,7 +156,6 @@ def test_exact_timestamp_is_served(self, setup, http_get, checkpoint_url):
assert f"Index of {urlparse(pub_1_url).path}
" in response
- @pytest.mark.parallel
def test_invalid_timestamp_returns_404(self, setup, http_get, distribution_base_url):
_, distribution = setup
with pytest.raises(ClientResponseError) as exc:
@@ -150,7 +168,6 @@ def test_invalid_timestamp_returns_404(self, setup, http_get, distribution_base_
assert exc.value.status == 404
- @pytest.mark.parallel
def test_checkpoint_artifact_is_served(self, setup, http_get, checkpoint_url):
pubs, distribution = setup
pub_1_url = checkpoint_url(distribution, pubs[1].pulp_created)
@@ -164,7 +181,6 @@ def test_checkpoint_artifact_is_served(self, setup, http_get, checkpoint_url):
artifact_names = {line.split(",")[0] for line in lines}
assert artifact_names == {"0", "1"}
- @pytest.mark.parallel
def test_non_checkpoint_timestamp_is_redirected(self, setup, http_get, checkpoint_url):
pubs, distribution = setup
# Using a non-checkpoint publication timestamp
@@ -178,7 +194,6 @@ def test_non_checkpoint_timestamp_is_redirected(self, setup, http_get, checkpoin
response = http_get(pub_4_url[:-1]).decode("utf-8")
assert f"Index of {urlparse(pub_3_url).path}
" in response
- @pytest.mark.parallel
def test_arbitrary_timestamp_is_redirected(self, setup, http_get, checkpoint_url):
pubs, distribution = setup
pub_1_url = checkpoint_url(distribution, pubs[1].pulp_created)
@@ -191,7 +206,6 @@ def test_arbitrary_timestamp_is_redirected(self, setup, http_get, checkpoint_url
response = http_get(arbitrary_url[:-1]).decode("utf-8")
assert f"Index of {urlparse(pub_1_url).path}
" in response
- @pytest.mark.parallel
def test_current_timestamp_serves_latest_checkpoint(self, setup, http_get, checkpoint_url):
pubs, distribution = setup
pub_3_url = checkpoint_url(distribution, pubs[3].pulp_created)
@@ -201,7 +215,6 @@ def test_current_timestamp_serves_latest_checkpoint(self, setup, http_get, check
assert f"Index of {urlparse(pub_3_url).path}
" in response
- @pytest.mark.parallel
def test_before_first_timestamp_returns_404(self, setup, http_get, checkpoint_url):
pubs, distribution = setup
pub_0_url = checkpoint_url(distribution, pubs[0].pulp_created)
@@ -211,7 +224,6 @@ def test_before_first_timestamp_returns_404(self, setup, http_get, checkpoint_ur
assert exc.value.status == 404
- @pytest.mark.parallel
def test_future_timestamp_returns_404(self, setup, http_get, checkpoint_url):
_, distribution = setup
url = checkpoint_url(distribution, datetime.now() + timedelta(days=1))
@@ -240,7 +252,6 @@ def test_checkpoint_publication_with_repository_version_fails(
def test_checkpoint_retention(
file_bindings,
file_repository_factory,
- file_distribution_factory,
create_publication,
monitor_task,
):
@@ -250,35 +261,24 @@ def test_checkpoint_retention(
retain their checkpoint=True flag. Older ones get their checkpoint flag cleared.
"""
repo = file_repository_factory()
- file_distribution_factory(repository=repo.pulp_href, checkpoint=True)
- # Create 4 checkpoint publications
- checkpoint_pubs = []
- for _ in range(4):
- checkpoint_pubs.append(create_publication(repo, True))
+ create_publication.precreate(3)
+ checkpoint_pubs = [create_publication(repo, True) for _ in range(2)]
- # Verify all 4 publications are checkpoints
for pub in checkpoint_pubs:
assert file_bindings.PublicationsFileApi.read(pub.pulp_href).checkpoint is True
- # Set retain_checkpoints=2 — should clear checkpoint flag on the 2 oldest
+ # Set retain_checkpoints=1 — should clear checkpoint flag on the oldest
task = file_bindings.RepositoriesFileApi.partial_update(
- repo.pulp_href, {"retain_checkpoints": 2}
+ repo.pulp_href, {"retain_checkpoints": 1}
).task
monitor_task(task)
- # Verify the 2 oldest had their flag cleared
- for pub in checkpoint_pubs[:2]:
- assert file_bindings.PublicationsFileApi.read(pub.pulp_href).checkpoint is False
-
- # Verify the 2 most recent still have checkpoint=True
- for pub in checkpoint_pubs[2:]:
- assert file_bindings.PublicationsFileApi.read(pub.pulp_href).checkpoint is True
+ assert file_bindings.PublicationsFileApi.read(checkpoint_pubs[0].pulp_href).checkpoint is False
+ assert file_bindings.PublicationsFileApi.read(checkpoint_pubs[1].pulp_href).checkpoint is True
# Create another checkpoint — should trigger steady-state cleanup
new_pub = create_publication(repo, True)
- # checkpoint_pubs[2] should now be cleared too
- assert file_bindings.PublicationsFileApi.read(checkpoint_pubs[2].pulp_href).checkpoint is False
- assert file_bindings.PublicationsFileApi.read(checkpoint_pubs[3].pulp_href).checkpoint is True
+ assert file_bindings.PublicationsFileApi.read(checkpoint_pubs[1].pulp_href).checkpoint is False
assert file_bindings.PublicationsFileApi.read(new_pub.pulp_href).checkpoint is True
diff --git a/pulpcore/tests/functional/api/using_plugin/test_content_directory.py b/pulpcore/tests/functional/api/using_plugin/test_content_directory.py
index a3fd6bab5c7..3dcf6957b02 100644
--- a/pulpcore/tests/functional/api/using_plugin/test_content_directory.py
+++ b/pulpcore/tests/functional/api/using_plugin/test_content_directory.py
@@ -3,8 +3,8 @@
@pytest.mark.parallel
def test_hidden_distros(file_distribution_factory, pulp_content_url, http_get):
- visible = [file_distribution_factory() for _ in range(5)]
- hidden = [file_distribution_factory(hidden=True) for _ in range(5)]
+ visible = [file_distribution_factory() for _ in range(2)]
+ hidden = [file_distribution_factory(hidden=True) for _ in range(2)]
content = http_get(pulp_content_url).decode("utf-8")
diff --git a/pulpcore/tests/functional/api/using_plugin/test_distributions.py b/pulpcore/tests/functional/api/using_plugin/test_distributions.py
index 5d7f57c1e59..2f3ebf78846 100644
--- a/pulpcore/tests/functional/api/using_plugin/test_distributions.py
+++ b/pulpcore/tests/functional/api/using_plugin/test_distributions.py
@@ -245,28 +245,33 @@ def has_base_path_lock(task):
@pytest.mark.parallel
def test_distribution_filtering(
file_bindings,
- file_remote_factory,
file_random_content_unit,
file_repository_factory,
gen_object_with_cleanup,
- write_3_iso_file_fixture_data_factory,
monitor_task,
+ tmp_path,
):
"""Test distribution filtering based on the content exposed from the distribution."""
- def generate_repo_with_content():
+ content_units = []
+ for name in ("1.bin", "2.bin"):
+ path = tmp_path / name
+ path.write_bytes(name.encode())
+ content_units.append(
+ file_bindings.ContentFilesApi.upload(file=str(path), relative_path=f"{uuid4()}.iso")
+ )
+ content1, content2 = content_units
+
+ def generate_repo_with_content(content):
repo = file_repository_factory()
- repo_manifest_path = write_3_iso_file_fixture_data_factory(str(uuid4()))
- remote = file_remote_factory(manifest_path=repo_manifest_path, policy="on_demand")
- body = file_bindings.FileRepositorySyncURL(remote=remote.pulp_href)
- task_response = file_bindings.RepositoriesFileApi.sync(repo.pulp_href, body).task
- version_href = monitor_task(task_response).created_resources[0]
- content = file_bindings.ContentFilesApi.list(repository_version_added=version_href).results[
- 0
- ]
+ monitor_task(
+ file_bindings.RepositoriesFileApi.modify(
+ repo.pulp_href, {"add_content_units": [content.pulp_href]}
+ ).task
+ )
return repo, content
- repo1, content1 = generate_repo_with_content()
+ repo1, content1 = generate_repo_with_content(content1)
publish_data = file_bindings.FileFilePublication(repository=repo1.pulp_href)
publication = gen_object_with_cleanup(file_bindings.PublicationsFileApi, publish_data)
@@ -303,7 +308,7 @@ def generate_repo_with_content():
)
assert {distribution_pub1.pulp_href, distribution_repopub.pulp_href} == results
- repo2, content2 = generate_repo_with_content()
+ repo2, content2 = generate_repo_with_content(content2)
# add new content to the first repository to see whether the distribution filtering correctly
# traverses to the latest publication concerning the repository under the question that should
@@ -376,12 +381,11 @@ def _get_manifest_from_distribution(distribution, distribution_base_url):
def test_distribution_serves_publication_content(
file_bindings,
file_repo,
- file_remote_ssl_factory,
- basic_manifest_path,
gen_object_with_cleanup,
file_distribution_factory,
distribution_base_url,
monitor_task,
+ tmp_path,
):
"""Test that publication, repository, and repository_version distributions serve
correct content.
@@ -392,18 +396,27 @@ def test_distribution_serves_publication_content(
- A distribution with ``repository`` serves the latest publication (for the latest version).
- A distribution with ``repository_version`` serves the latest publication for that version.
"""
- # Sync to create version 1 (3 files)
- remote = file_remote_ssl_factory(manifest_path=basic_manifest_path, policy="immediate")
- body = file_bindings.FileRepositorySyncURL(remote=remote.pulp_href)
- monitor_task(file_bindings.RepositoriesFileApi.sync(file_repo.pulp_href, body).task)
+ content_hrefs = []
+ for i in range(3):
+ path = tmp_path / f"{i}.bin"
+ path.write_bytes(f"{i}".encode())
+ content_hrefs.append(
+ file_bindings.ContentFilesApi.upload(
+ file=str(path), relative_path=f"{uuid4()}.iso"
+ ).pulp_href
+ )
+ monitor_task(
+ file_bindings.RepositoriesFileApi.modify(
+ file_repo.pulp_href, {"add_content_units": content_hrefs}
+ ).task
+ )
file_repo = file_bindings.RepositoriesFileApi.read(file_repo.pulp_href)
v1_href = file_repo.latest_version_href
# Remove one content unit to create version 2 (2 files)
- v1_content = file_bindings.ContentFilesApi.list(repository_version=v1_href).results
monitor_task(
file_bindings.RepositoriesFileApi.modify(
- file_repo.pulp_href, {"remove_content_units": [v1_content[0].pulp_href]}
+ file_repo.pulp_href, {"remove_content_units": [content_hrefs[0]]}
).task
)
file_repo = file_bindings.RepositoriesFileApi.read(file_repo.pulp_href)
diff --git a/pulpcore/tests/functional/api/using_plugin/test_pagination.py b/pulpcore/tests/functional/api/using_plugin/test_pagination.py
index 333a7d3bccf..9b12b05f13a 100644
--- a/pulpcore/tests/functional/api/using_plugin/test_pagination.py
+++ b/pulpcore/tests/functional/api/using_plugin/test_pagination.py
@@ -6,38 +6,50 @@
@pytest.mark.parallel
def test_repo_version_pagination(
file_bindings,
- file_content_unit_with_name_factory,
file_repo,
monitor_task,
+ tmp_path,
):
- # Create 20 new repository versions (21 in total)
- for i in range(20):
- content_unit = file_content_unit_with_name_factory(f"{i}.iso")
+ # Create several content units, then add them one-by-one to produce enough
+ # repository versions for pagination.
+ # limit=2 with 5 versions total (initial + 4) covers first/middle/last pages.
+ page_size = 2
+ versions_to_add = 4
+
+ content_hrefs = []
+ for i in range(versions_to_add):
+ path = tmp_path / f"{i}.iso"
+ path.write_bytes(f"{i}".encode())
+ content_hrefs.append(
+ file_bindings.ContentFilesApi.upload(file=str(path), relative_path=f"{i}.iso").pulp_href
+ )
+
+ for content_href in content_hrefs:
monitor_task(
file_bindings.RepositoriesFileApi.modify(
- file_repo.pulp_href, {"add_content_units": [content_unit.pulp_href]}
+ file_repo.pulp_href, {"add_content_units": [content_href]}
).task
)
- # Assert that the limit of 10 items per page of results is respected.
+ # Assert that the requested limit is respected on the first page.
first_page = file_bindings.RepositoriesFileVersionsApi.list(
- file_repo.pulp_href, limit=10, offset=0
+ file_repo.pulp_href, limit=page_size, offset=0
)
- assert len(first_page.results) == 10
+ assert len(first_page.results) == page_size
assert first_page.previous is None
assert first_page.next is not None
- # Assert that a limit and an offset are respected.
+ # Assert that limit and offset are respected on a middle page.
second_page = file_bindings.RepositoriesFileVersionsApi.list(
- file_repo.pulp_href, limit=10, offset=10
+ file_repo.pulp_href, limit=page_size, offset=page_size
)
- assert len(second_page.results) == 10
+ assert len(second_page.results) == page_size
assert second_page.previous is not None
assert second_page.next is not None
- # Assert that the limit and offset are respected for the last page of results.
+ # Assert the last (partial) page has previous and no next.
third_page = file_bindings.RepositoriesFileVersionsApi.list(
- file_repo.pulp_href, limit=10, offset=20
+ file_repo.pulp_href, limit=page_size, offset=page_size * 2
)
assert len(third_page.results) == 1
assert third_page.previous is not None
diff --git a/pulpcore/tests/functional/api/using_plugin/test_prn.py b/pulpcore/tests/functional/api/using_plugin/test_prn.py
index 55164c8f7d2..4927ddc9542 100644
--- a/pulpcore/tests/functional/api/using_plugin/test_prn.py
+++ b/pulpcore/tests/functional/api/using_plugin/test_prn.py
@@ -50,7 +50,7 @@ def test_create_and_filter_with_prn(
):
"""Test that we can use PRNs to refer to any object"""
# Creation tests
- remote = file_remote_factory(basic_manifest_path)
+ remote = file_remote_factory(basic_manifest_path, policy="on_demand")
task = file_bindings.RepositoriesFileApi.sync(file_repo.pulp_href, {"remote": remote.prn}).task
task = monitor_task(task)
assert len(task.created_resources) == 1
diff --git a/pulpcore/tests/functional/api/using_plugin/test_pulpimport.py b/pulpcore/tests/functional/api/using_plugin/test_pulpimport.py
index 349d6781375..7865a2aef78 100644
--- a/pulpcore/tests/functional/api/using_plugin/test_pulpimport.py
+++ b/pulpcore/tests/functional/api/using_plugin/test_pulpimport.py
@@ -29,8 +29,7 @@
]
-@pytest.fixture
-def import_export_repositories(
+def _create_import_export_repositories(
file_bindings,
file_repository_factory,
file_remote_ssl_factory,
@@ -39,35 +38,88 @@ def import_export_repositories(
):
import_repos = []
export_repos = []
- for r in range(NUM_REPOS):
+ sync_tasks = []
+ for _ in range(NUM_REPOS):
import_repo = file_repository_factory()
export_repo = file_repository_factory()
remote = file_remote_ssl_factory(manifest_path=basic_manifest_path, policy="immediate")
repository_sync_data = FileRepositorySyncURL(remote=remote.pulp_href)
- sync_response = file_bindings.RepositoriesFileApi.sync(
- export_repo.pulp_href, repository_sync_data
+ sync_tasks.append(
+ file_bindings.RepositoriesFileApi.sync(export_repo.pulp_href, repository_sync_data).task
)
- monitor_task(sync_response.task)
-
- export_repo = file_bindings.RepositoriesFileApi.read(export_repo.pulp_href)
-
export_repos.append(export_repo)
import_repos.append(import_repo)
+ for task in sync_tasks:
+ monitor_task(task)
+
+ export_repos = [file_bindings.RepositoriesFileApi.read(repo.pulp_href) for repo in export_repos]
return import_repos, export_repos
-@pytest.fixture
-def exporter(pulpcore_bindings, tmpdir, gen_object_with_cleanup, import_export_repositories):
+def _make_importer_factory(pulpcore_bindings, gen_object_with_cleanup, import_export_repositories):
+ def _importer_factory(name=None, exported_repos=None, mapping=None):
+ """Create an importer."""
+ _import_repos, _exported_repos = import_export_repositories
+ if not name:
+ name = str(uuid.uuid4())
+
+ if not mapping:
+ mapping = {}
+ if not exported_repos:
+ exported_repos = _exported_repos
+
+ for idx, repo in enumerate(exported_repos):
+ mapping[repo.name] = _import_repos[idx].name
+
+ body = {
+ "name": name,
+ "repo_mapping": mapping,
+ }
+
+ importer = gen_object_with_cleanup(pulpcore_bindings.ImportersPulpApi, body)
+
+ return importer
+
+ return _importer_factory
+
+
+@pytest.fixture(scope="class")
+def import_export_repositories(
+ file_bindings,
+ file_repository_factory,
+ file_remote_ssl_factory,
+ basic_manifest_path,
+ monitor_task,
+):
+ return _create_import_export_repositories(
+ file_bindings,
+ file_repository_factory,
+ file_remote_ssl_factory,
+ basic_manifest_path,
+ monitor_task,
+ )
+
+
+@pytest.fixture(scope="class")
+def exporter(
+ pulpcore_bindings, tmp_path_factory, gen_object_with_cleanup, import_export_repositories
+):
_, export_repos = import_export_repositories
body = {
"name": str(uuid.uuid4()),
"repositories": [r.pulp_href for r in export_repos],
- "path": str(tmpdir),
+ "path": str(tmp_path_factory.mktemp("exporter")),
}
- exporter = gen_object_with_cleanup(pulpcore_bindings.ExportersPulpApi, body)
- return exporter
+ return gen_object_with_cleanup(pulpcore_bindings.ExportersPulpApi, body)
+
+
+@pytest.fixture(scope="class")
+def importer_factory(pulpcore_bindings, gen_object_with_cleanup, import_export_repositories):
+ return _make_importer_factory(
+ pulpcore_bindings, gen_object_with_cleanup, import_export_repositories
+ )
@pytest.fixture
@@ -95,34 +147,6 @@ def import_check_directory(tmp_path):
os.chmod(f"{tmp_path}/nowritedir", 0o755)
-@pytest.fixture
-def importer_factory(pulpcore_bindings, gen_object_with_cleanup, import_export_repositories):
- def _importer_factory(name=None, exported_repos=None, mapping=None):
- """Create an importer."""
- _import_repos, _exported_repos = import_export_repositories
- if not name:
- name = str(uuid.uuid4())
-
- if not mapping:
- mapping = {}
- if not exported_repos:
- exported_repos = _exported_repos
-
- for idx, repo in enumerate(exported_repos):
- mapping[repo.name] = _import_repos[idx].name
-
- body = {
- "name": name,
- "repo_mapping": mapping,
- }
-
- importer = gen_object_with_cleanup(pulpcore_bindings.ImportersPulpApi, body)
-
- return importer
-
- return _importer_factory
-
-
def _find_toc(chunked_export):
filenames = [f for f in list(chunked_export.output_file_info.keys()) if f.endswith("json")]
return filenames[0]
@@ -134,7 +158,24 @@ def _find_path(created_export):
@pytest.fixture
-def perform_import(pulpcore_bindings, exporter, generate_export, monitor_task_group):
+def generate_export(pulpcore_bindings, monitor_task):
+ """Create and read back an export for the specified PulpExporter."""
+
+ def _generate_export(exporter, body=None):
+ if body is None:
+ body = {}
+
+ export_response = pulpcore_bindings.ExportersPulpExportsApi.create(exporter.pulp_href, body)
+ export_href = monitor_task(export_response.task).created_resources[0]
+ export = pulpcore_bindings.ExportersPulpExportsApi.read(export_href)
+
+ return export
+
+ return _generate_export
+
+
+@pytest.fixture
+def perform_import(pulpcore_bindings, monitor_task_group):
def _perform_import(importer, export, chunked=False, body=None):
"""Perform an import with importer."""
if body is None:
@@ -155,87 +196,150 @@ def _perform_import(importer, export, chunked=False, body=None):
return _perform_import
-@pytest.mark.parallel
-def test_importer_create(pulpcore_bindings, importer_factory):
- """Test creating an importer."""
- name = str(uuid.uuid4())
- importer = importer_factory(name)
- assert importer.name == name
-
- importer = pulpcore_bindings.ImportersPulpApi.read(importer.pulp_href)
- assert importer.name == name
-
-
-@pytest.mark.parallel
-def test_importer_delete(pulpcore_bindings, importer_factory):
- """Test deleting an importer."""
- importer = importer_factory()
-
- pulpcore_bindings.ImportersPulpApi.delete(importer.pulp_href)
-
- with pytest.raises(ApiException) as ae:
- pulpcore_bindings.ImportersPulpApi.read(importer.pulp_href)
- assert 404 == ae.value.status
-
-
-@pytest.mark.parallel
-def test_import(
- file_bindings,
- exporter,
- generate_export,
- importer_factory,
- import_export_repositories,
- perform_import,
-):
- """Test an import."""
- import_repos, exported_repos = import_export_repositories
- importer = importer_factory()
- export = generate_export(exporter)
- task_group = perform_import(importer, export)
- assert (len(import_repos) + 1) == task_group.completed
-
- for report in task_group.group_progress_reports:
- if report.code == "import.repo.versions":
- assert report.done == len(import_repos)
-
- for repo in import_repos:
- repo = file_bindings.RepositoriesFileApi.read(repo.pulp_href)
- assert f"{repo.pulp_href}versions/1/" == repo.latest_version_href
-
-
-@pytest.mark.parallel
-@pytest.mark.parametrize("chunk_size", ["1KB", "5KB"])
-def test_chunked_import(
- file_bindings,
- chunk_size,
- exporter,
- generate_export,
- importer_factory,
- import_export_repositories,
- perform_import,
-):
- """Test an import."""
- import_repos, exported_repos = import_export_repositories
- importer = importer_factory()
- export = generate_export(exporter, body={"chunk_size": chunk_size})
- task_group = perform_import(importer, export, chunked=True)
- assert (len(import_repos) + 1) == task_group.completed
-
- for repo in import_repos:
- repo = file_bindings.RepositoriesFileApi.read(repo.pulp_href)
- assert f"{repo.pulp_href}versions/1/" == repo.latest_version_href
-
+class TestPulpImport:
+ """Don't mark parallel, tests are shorter than setup."""
+
+ def test_importer_create(self, pulpcore_bindings, importer_factory):
+ """Test creating an importer."""
+ name = str(uuid.uuid4())
+ importer = importer_factory(name)
+ assert importer.name == name
+
+ importer = pulpcore_bindings.ImportersPulpApi.read(importer.pulp_href)
+ assert importer.name == name
+
+ def test_importer_delete(self, pulpcore_bindings, importer_factory):
+ """Test deleting an importer."""
+ importer = importer_factory()
+
+ pulpcore_bindings.ImportersPulpApi.delete(importer.pulp_href)
+
+ with pytest.raises(ApiException) as ae:
+ pulpcore_bindings.ImportersPulpApi.read(importer.pulp_href)
+ assert 404 == ae.value.status
+
+ def test_import(
+ self,
+ file_bindings,
+ exporter,
+ generate_export,
+ importer_factory,
+ import_export_repositories,
+ perform_import,
+ ):
+ """Test an import."""
+ import_repos, exported_repos = import_export_repositories
+ importer = importer_factory()
+ export = generate_export(exporter)
+ task_group = perform_import(importer, export)
+ assert (len(import_repos) + 1) == task_group.completed
+
+ for report in task_group.group_progress_reports:
+ if report.code == "import.repo.versions":
+ assert report.done == len(import_repos)
+
+ for repo in import_repos:
+ repo = file_bindings.RepositoriesFileApi.read(repo.pulp_href)
+ assert f"{repo.pulp_href}versions/1/" == repo.latest_version_href
+
+ @pytest.mark.parametrize("chunk_size", ["1KB", "5KB"])
+ def test_chunked_import(
+ self,
+ file_bindings,
+ chunk_size,
+ exporter,
+ generate_export,
+ importer_factory,
+ import_export_repositories,
+ perform_import,
+ ):
+ """Test an import."""
+ import_repos, exported_repos = import_export_repositories
+ importer = importer_factory()
+ export = generate_export(exporter, body={"chunk_size": chunk_size})
+ task_group = perform_import(importer, export, chunked=True)
+ assert (len(import_repos) + 1) == task_group.completed
+
+ for repo in import_repos:
+ repo = file_bindings.RepositoriesFileApi.read(repo.pulp_href)
+ assert f"{repo.pulp_href}versions/1/" == repo.latest_version_href
+
+ def test_import_mapping_missing_repos(self, importer_factory, import_export_repositories):
+ import_repos, exported_repos = import_export_repositories
+ a_map = {"foo": "bar"}
+ for repo in import_repos:
+ a_map[repo.name] = repo.name
+ a_map["blech"] = "bang"
+
+ with pytest.raises(ApiException, match="['bar', 'bang']"):
+ importer_factory(mapping=a_map)
+
+ def test_double_import(
+ self,
+ pulpcore_bindings,
+ file_bindings,
+ exporter,
+ generate_export,
+ importer_factory,
+ import_export_repositories,
+ perform_import,
+ ):
+ """Test two imports of our export."""
+ import_repos, exported_repos = import_export_repositories
+ export = generate_export(exporter)
+
+ importer = importer_factory()
+ perform_import(importer, export)
+ perform_import(importer, export)
+
+ imports = pulpcore_bindings.ImportersPulpImportsApi.list(importer.pulp_href).results
+ assert len(imports) == 2
+
+ for repo in import_repos:
+ repo = file_bindings.RepositoriesFileApi.read(repo.pulp_href)
+ # still only one version as pulp won't create a new version if nothing changed
+ assert f"{repo.pulp_href}versions/1/" == repo.latest_version_href
+
+ def test_import_check_valid_path(self, pulpcore_bindings, exporter, generate_export):
+ created_export = generate_export(exporter)
+ body = {"path": _find_path(created_export)}
+ result = pulpcore_bindings.ImportersPulpImportCheckApi.pulp_import_check_post(body)
+ assert result.path.context == _find_path(created_export)
+ assert result.path.is_valid
+ assert len(result.path.messages) == 0
+ assert result.toc is None
+ assert result.repo_mapping is None
+
+ def test_import_check_valid_toc(self, pulpcore_bindings, exporter, generate_export):
+ chunked_export = generate_export(exporter, body={"chunk_size": "5KB"})
+ body = {"toc": _find_toc(chunked_export)}
+ result = pulpcore_bindings.ImportersPulpImportCheckApi.pulp_import_check_post(body)
+ assert result.toc.context == _find_toc(chunked_export)
+ assert result.toc.is_valid
+ assert len(result.toc.messages) == 0
+ assert result.path is None
+ assert result.repo_mapping is None
+
+ def test_import_check_all_valid(self, pulpcore_bindings, exporter, generate_export):
+ created_export = generate_export(exporter)
+ chunked_export = generate_export(exporter, body={"chunk_size": "5KB"})
+ body = {
+ "path": _find_path(created_export),
+ "toc": _find_toc(chunked_export),
+ "repo_mapping": json.dumps({"foo": "bar"}),
+ }
+ result = pulpcore_bindings.ImportersPulpImportCheckApi.pulp_import_check_post(body)
+ assert result.path.context == _find_path(created_export)
+ assert result.toc.context == _find_toc(chunked_export)
+ assert result.repo_mapping.context == json.dumps({"foo": "bar"})
-@pytest.fixture
-def test_import_mapping_missing_repos(importer_factory, import_export_repositories):
- import_repos, exported_repos = import_export_repositories
- a_map = {"foo": "bar"}
- for repo in import_repos:
- a_map[repo.name] = repo.name
- a_map["blech"] = "bang"
+ assert result.path.is_valid
+ assert result.toc.is_valid
+ assert result.repo_mapping.is_valid
- with pytest.raises(ApiException, match="['bar', 'bang']"):
- importer_factory(mapping=a_map)
+ assert len(result.path.messages) == 0
+ assert len(result.toc.messages) == 0
+ assert len(result.repo_mapping.messages) == 0
@pytest.mark.parallel
@@ -249,7 +353,7 @@ def test_import_auto_repo_creation(
generate_export,
monitor_task,
perform_import,
- tmpdir,
+ tmp_path_factory,
):
"""Test the automatic repository creation feature where users do not ."""
# 1. create and sync a new repository
@@ -271,7 +375,7 @@ def test_import_auto_repo_creation(
body = {
"name": str(uuid.uuid4()),
"repositories": [export_repo.pulp_href],
- "path": str(tmpdir),
+ "path": str(tmp_path_factory.mktemp("auto-export")),
}
exporter = gen_object_with_cleanup(pulpcore_bindings.ExportersPulpApi, body)
export = generate_export(exporter)
@@ -301,57 +405,6 @@ def test_import_auto_repo_creation(
monitor_task(file_bindings.RepositoriesFileApi.delete(imported_repo.pulp_href).task)
-@pytest.mark.parallel
-def test_double_import(
- pulpcore_bindings,
- file_bindings,
- exporter,
- generate_export,
- importer_factory,
- import_export_repositories,
- perform_import,
-):
- """Test two imports of our export."""
- import_repos, exported_repos = import_export_repositories
- export = generate_export(exporter)
-
- importer = importer_factory()
- perform_import(importer, export)
- perform_import(importer, export)
-
- imports = pulpcore_bindings.ImportersPulpImportsApi.list(importer.pulp_href).results
- assert len(imports) == 2
-
- for repo in import_repos:
- repo = file_bindings.RepositoriesFileApi.read(repo.pulp_href)
- # still only one version as pulp won't create a new version if nothing changed
- assert f"{repo.pulp_href}versions/1/" == repo.latest_version_href
-
-
-@pytest.mark.parallel
-def test_import_check_valid_path(pulpcore_bindings, exporter, generate_export):
- created_export = generate_export(exporter)
- body = {"path": _find_path(created_export)}
- result = pulpcore_bindings.ImportersPulpImportCheckApi.pulp_import_check_post(body)
- assert result.path.context == _find_path(created_export)
- assert result.path.is_valid
- assert len(result.path.messages) == 0
- assert result.toc is None
- assert result.repo_mapping is None
-
-
-@pytest.mark.parallel
-def test_import_check_valid_toc(pulpcore_bindings, exporter, generate_export):
- chunked_export = generate_export(exporter, body={"chunk_size": "5KB"})
- body = {"toc": _find_toc(chunked_export)}
- result = pulpcore_bindings.ImportersPulpImportCheckApi.pulp_import_check_post(body)
- assert result.toc.context == _find_toc(chunked_export)
- assert result.toc.is_valid
- assert len(result.toc.messages) == 0
- assert result.path is None
- assert result.repo_mapping is None
-
-
@pytest.mark.parallel
def test_import_check_repo_mapping(pulpcore_bindings):
body = {"repo_mapping": json.dumps({"foo": "bar"})}
@@ -401,29 +454,6 @@ def test_import_check_no_file(pulpcore_bindings):
assert any("file /tmp/idonotexist does not exist" in s for s in result.toc.messages)
-@pytest.mark.parallel
-def test_import_check_all_valid(pulpcore_bindings, exporter, generate_export):
- created_export = generate_export(exporter)
- chunked_export = generate_export(exporter, body={"chunk_size": "5KB"})
- body = {
- "path": _find_path(created_export),
- "toc": _find_toc(chunked_export),
- "repo_mapping": json.dumps({"foo": "bar"}),
- }
- result = pulpcore_bindings.ImportersPulpImportCheckApi.pulp_import_check_post(body)
- assert result.path.context == _find_path(created_export)
- assert result.toc.context == _find_toc(chunked_export)
- assert result.repo_mapping.context == json.dumps({"foo": "bar"})
-
- assert result.path.is_valid
- assert result.toc.is_valid
- assert result.repo_mapping.is_valid
-
- assert len(result.path.messages) == 0
- assert len(result.toc.messages) == 0
- assert len(result.repo_mapping.messages) == 0
-
-
@pytest.mark.parallel
def test_import_check_multiple_errors(pulpcore_bindings, import_check_directory):
body = {
@@ -451,37 +481,47 @@ def test_import_check_multiple_errors(pulpcore_bindings, import_check_directory)
assert result.repo_mapping.messages[0] == "invalid JSON"
+# Standalone chain for test_import_not_latest_version (mutates repos differently).
@pytest.fixture
-def generate_export(pulpcore_bindings, monitor_task):
- """Create and read back an export for the specified PulpExporter."""
-
- def _generate_export(exporter, body=None):
- if body is None:
- body = {}
-
- export_response = pulpcore_bindings.ExportersPulpExportsApi.create(exporter.pulp_href, body)
- export_href = monitor_task(export_response.task).created_resources[0]
- export = pulpcore_bindings.ExportersPulpExportsApi.read(export_href)
+def standalone_import_export_repositories(
+ file_bindings,
+ file_repository_factory,
+ file_remote_ssl_factory,
+ basic_manifest_path,
+ monitor_task,
+):
+ return _create_import_export_repositories(
+ file_bindings,
+ file_repository_factory,
+ file_remote_ssl_factory,
+ basic_manifest_path,
+ monitor_task,
+ )
- return export
- return _generate_export
+@pytest.fixture
+def standalone_importer_factory(
+ pulpcore_bindings, gen_object_with_cleanup, standalone_import_export_repositories
+):
+ return _make_importer_factory(
+ pulpcore_bindings, gen_object_with_cleanup, standalone_import_export_repositories
+ )
@pytest.fixture
def exported_version(
pulpcore_bindings,
file_bindings,
- importer_factory,
+ standalone_importer_factory,
gen_object_with_cleanup,
- import_export_repositories,
+ standalone_import_export_repositories,
generate_export,
perform_import,
file_repo,
monitor_task,
- tmpdir,
+ tmp_path_factory,
):
- import_repos, export_repos = import_export_repositories
+ import_repos, export_repos = standalone_import_export_repositories
file_list = pulpcore_bindings.ContentApi.list(
repository_version=export_repos[0].latest_version_href
@@ -503,7 +543,7 @@ def exported_version(
body = {
"name": str(uuid.uuid4()),
"repositories": [file_repo.pulp_href],
- "path": str(tmpdir),
+ "path": str(tmp_path_factory.mktemp("exported-version")),
}
exporter = gen_object_with_cleanup(pulpcore_bindings.ExportersPulpApi, body)
@@ -515,7 +555,7 @@ def exported_version(
}
export = generate_export(exporter, body)
- importer = importer_factory(exported_repos=[file_repo])
+ importer = standalone_importer_factory(exported_repos=[file_repo])
task_group = perform_import(importer, export, chunked=False)
return import_repos, task_group
diff --git a/pulpcore/tests/functional/api/using_plugin/test_repo_versions.py b/pulpcore/tests/functional/api/using_plugin/test_repo_versions.py
index 0ad7d065380..f7222d596b7 100644
--- a/pulpcore/tests/functional/api/using_plugin/test_repo_versions.py
+++ b/pulpcore/tests/functional/api/using_plugin/test_repo_versions.py
@@ -2,7 +2,6 @@
import uuid
from random import choice
-from tempfile import NamedTemporaryFile
from uuid import uuid4
import pytest
@@ -11,26 +10,15 @@
@pytest.fixture
-def file_9_contents(
- file_bindings,
- file_repository_factory,
- monitor_task,
-):
+def file_9_contents(file_bindings, tmp_path):
"""Create 9 content units with relative paths "A" through "I"."""
- bucket_repo = file_repository_factory()
+ names = ["A", "B", "C", "D", "E", "F", "G", "H", "I"]
content_units = {}
- for name in ["A", "B", "C", "D", "E", "F", "G", "H", "I"]:
- with NamedTemporaryFile() as tf:
- tf.write(name.encode())
- tf.flush()
- response = file_bindings.ContentFilesApi.create(
- relative_path=name, file=tf.name, repository=bucket_repo.pulp_href
- )
- result = monitor_task(response.task)
- content_href = next(
- (item for item in result.created_resources if "content/file/files/" in item)
- )
- content_units[name] = file_bindings.ContentFilesApi.read(content_href)
+ for name in names:
+ path = tmp_path / name
+ path.write_bytes(name.encode())
+ uploaded = file_bindings.ContentFilesApi.upload(relative_path=name, file=str(path))
+ content_units[name] = file_bindings.ContentFilesApi.read(uploaded.pulp_href)
return content_units
@@ -873,18 +861,17 @@ def test_repo_version_retention(
@pytest.mark.parallel
def test_repo_versions_protected_from_cleanup(
file_bindings,
- file_content_unit_with_name_factory,
file_repository_factory,
file_distribution_factory,
gen_object_with_cleanup,
monitor_task,
+ tmp_path,
):
"""Test that distributed repo versions are protected from retain_repo_versions."""
- def _modify_and_validate(repo, expected_version, expected_total):
- content = file_content_unit_with_name_factory(str(uuid.uuid4()))
+ def _modify_and_validate(repo, content_href, expected_version, expected_total):
task = file_bindings.RepositoriesFileApi.modify(
- repo.pulp_href, {"add_content_units": [content.pulp_href]}
+ repo.pulp_href, {"add_content_units": [content_href]}
).task
monitor_task(task)
@@ -896,6 +883,17 @@ def _modify_and_validate(repo, expected_version, expected_total):
return repo
+ content_hrefs = []
+ for i in range(6):
+ path = tmp_path / f"{i}.bin"
+ path.write_bytes(f"{i}".encode())
+ content_hrefs.append(
+ file_bindings.ContentFilesApi.upload(
+ file=str(path), relative_path=f"{uuid.uuid4()}.iso"
+ ).pulp_href
+ )
+ content_hrefs = iter(content_hrefs)
+
# Setup
repo = file_repository_factory(retain_repo_versions=1)
@@ -906,7 +904,7 @@ def _modify_and_validate(repo, expected_version, expected_total):
file_distribution_factory(publication=publication.pulp_href)
# Version 0 is protected since it's distributed
- repo = _modify_and_validate(repo, "1", 2)
+ repo = _modify_and_validate(repo, next(content_hrefs), "1", 2)
# Create a new publication and distribution which protects version 1 from deletion
file_distribution_factory(repository=repo.pulp_href)
@@ -916,10 +914,10 @@ def _modify_and_validate(repo, expected_version, expected_total):
file_distribution_factory(publication=publication.pulp_href)
# Create version 2 and there should be 3 versions now (2 protected)
- repo = _modify_and_validate(repo, "2", 3)
+ repo = _modify_and_validate(repo, next(content_hrefs), "2", 3)
# Version 2 will be removed since we're creating version 3 and it's not protected
- repo = _modify_and_validate(repo, "3", 3)
+ repo = _modify_and_validate(repo, next(content_hrefs), "3", 3)
# Publish version 3 as a checkpoint and distribute it
gen_object_with_cleanup(
@@ -929,7 +927,7 @@ def _modify_and_validate(repo, expected_version, expected_total):
file_distribution_factory(repository=repo.pulp_href, checkpoint=True)
# Version 3 is protected since it's distributed by the checkpoint distribution
- repo = _modify_and_validate(repo, "4", 4)
+ repo = _modify_and_validate(repo, next(content_hrefs), "4", 4)
# Publish version 4 as a checkpoint (it's already distributed)
gen_object_with_cleanup(
@@ -938,10 +936,10 @@ def _modify_and_validate(repo, expected_version, expected_total):
)
# Version 4 is protected since it's distributed by the checkpoint distribution
- repo = _modify_and_validate(repo, "5", 5)
+ repo = _modify_and_validate(repo, next(content_hrefs), "5", 5)
# Version 5 will be removed since it's not protected and we're creating version 6
- _modify_and_validate(repo, "6", 5)
+ _modify_and_validate(repo, next(content_hrefs), "6", 5)
@pytest.mark.parallel
diff --git a/pulpcore/tests/unit/tasking/test_missing_worker_cleanup.py b/pulpcore/tests/unit/tasking/test_missing_worker_cleanup.py
new file mode 100644
index 00000000000..db68e91d68c
--- /dev/null
+++ b/pulpcore/tests/unit/tasking/test_missing_worker_cleanup.py
@@ -0,0 +1,57 @@
+"""Unit tests for missing-worker cleanup."""
+
+from datetime import timedelta
+from uuid import uuid4
+
+import pytest
+from django.conf import settings
+from django.utils import timezone
+
+from pulpcore.app.models import AppStatus, Task
+from pulpcore.constants import TASK_STATES
+from pulpcore.tasking.worker import PulpcoreWorker
+
+
+@pytest.mark.django_db
+def test_missing_worker_cleanup_fails_abandoned_task(monkeypatch):
+ """
+ Surviving workers should fail tasks abandoned by a missing worker.
+
+ Mirrors the functional test_worker_cleanup_on_missing_worker path without
+ waiting on the heartbeat/cleanup interval.
+ """
+ monkeypatch.setattr(AppStatus.objects, "_current_app_status", None)
+ dead_worker = AppStatus.objects.create(app_type="worker", name=f"dead-worker-{uuid4()}")
+ AppStatus.objects.filter(pk=dead_worker.pk).update(
+ last_heartbeat=timezone.now() - timedelta(seconds=settings.WORKER_TTL + 60)
+ )
+ dead_worker.refresh_from_db()
+ assert dead_worker.missing
+
+ resource = f"exclusive:{uuid4()}"
+ task = Task.objects.create(
+ state=TASK_STATES.RUNNING,
+ name="pulpcore.app.tasks.test.sleep",
+ logging_cid=str(uuid4()),
+ app_lock=dead_worker,
+ unblocked_at=timezone.now(),
+ started_at=timezone.now(),
+ reserved_resources_record=[resource],
+ )
+
+ monkeypatch.setattr(AppStatus.objects, "_current_app_status", None)
+ survivor = PulpcoreWorker()
+
+ # Drop the missing worker record (nulls task.app_lock via SET_NULL).
+ survivor.app_worker_cleanup()
+ task.refresh_from_db()
+ assert task.app_lock_id is None
+ assert not AppStatus.objects.filter(pk=dead_worker.pk).exists()
+
+ # Pick up the orphaned RUNNING task and mark it failed.
+ survivor.handle_unblocked_tasks()
+ task.refresh_from_db()
+ assert task.state == TASK_STATES.FAILED
+ assert task.error is not None
+ reason = task.error.get("reason", "").lower()
+ assert "worker" in reason and "missing" in reason
diff --git a/pulpcore/tests/unit/tasking/test_redis_fetch_task.py b/pulpcore/tests/unit/tasking/test_redis_fetch_task.py
new file mode 100644
index 00000000000..d65976bbd7d
--- /dev/null
+++ b/pulpcore/tests/unit/tasking/test_redis_fetch_task.py
@@ -0,0 +1,59 @@
+"""Unit tests for RedisWorker.fetch_task batch expansion."""
+
+import time
+from uuid import uuid4
+
+import pytest
+
+from pulpcore.app.models import AppStatus, Task
+from pulpcore.constants import TASK_STATES
+from pulpcore.tasking.redis_worker import RedisWorker
+
+
+def _waiting_task(resource):
+ return Task.objects.create(
+ state=TASK_STATES.WAITING,
+ name="pulpcore.app.tasks.test.sleep",
+ logging_cid=str(uuid4()),
+ reserved_resources_record=[resource],
+ )
+
+
+@pytest.mark.django_db
+def test_fetch_task_beyond_initial_batch(monkeypatch):
+ """Blocked tasks filling the first fetch batch must not hide a later runnable task.
+
+ Replaces the functional test that slept 60s and queued 25 tasks so a live
+ RedisWorker would double FETCH_TASK_LIMIT.
+ """
+ monkeypatch.setattr(AppStatus.objects, "_current_app_status", None)
+ monkeypatch.setattr("pulpcore.tasking.redis_worker.FETCH_TASK_LIMIT", 3)
+
+ blocked_resource = f"exclusive:{uuid4()}"
+ other_resource = f"exclusive:{uuid4()}"
+
+ def acquire_locks(_conn, _name, _task_lock_key, exclusive_resources, _shared_resources):
+ if blocked_resource in exclusive_resources:
+ return [blocked_resource]
+ return []
+
+ monkeypatch.setattr("pulpcore.tasking.redis_worker.acquire_locks", acquire_locks)
+
+ worker = RedisWorker.__new__(RedisWorker)
+ worker.ignored_task_ids = []
+ worker.name = f"test-worker-{uuid4()}"
+ worker.app_status = AppStatus.objects.create(app_type="worker", name=worker.name)
+ worker.redis_conn = object()
+
+ for _ in range(4):
+ _waiting_task(blocked_resource)
+ # Task.pulp_created cannot be updated (DB trigger); sleep so this sorts after the batch.
+ time.sleep(0.05)
+ runnable = _waiting_task(other_resource)
+
+ fetched = worker.fetch_task()
+
+ assert fetched is not None
+ assert fetched.pk == runnable.pk
+ runnable.refresh_from_db()
+ assert runnable.app_lock_id == worker.app_status.pk