diff --git a/pulp_file/tests/functional/api/test_filesystem_export.py b/pulp_file/tests/functional/api/test_filesystem_export.py
index da92c1b471..1caf547269 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 d44ba9f5b7..e39d94713f 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 792c15812c..7640a870a5 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/tests/functional/api/test_replication.py b/pulpcore/tests/functional/api/test_replication.py
index 17ab460363..f208a4be08 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 d9e95a56a0..4b8135882d 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 b3844d7e69..305b650071 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 a3fd6bab5c..3dcf6957b0 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_pagination.py b/pulpcore/tests/functional/api/using_plugin/test_pagination.py
index 333a7d3bcc..9b12b05f13 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_pulpimport.py b/pulpcore/tests/functional/api/using_plugin/test_pulpimport.py
index 349d678137..7865a2aef7 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 0ad7d06538..f7222d596b 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 0000000000..db68e91d68
--- /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 0000000000..d65976bbd7
--- /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