diff --git a/sources/test/conftest.py b/sources/test/conftest.py index 9b9ec2fb..8ff0a308 100644 --- a/sources/test/conftest.py +++ b/sources/test/conftest.py @@ -1,14 +1,16 @@ +import inspect import os import pathlib from types import ModuleType import pytest +from osbuild import sources, testutil from osbuild.testutil.imports import import_module_from_path -@pytest.fixture -def sources_module(request: pytest.FixtureRequest) -> ModuleType: +@pytest.fixture(name="sources_module") +def sources_module_fixture(request: pytest.FixtureRequest) -> ModuleType: """sources_module is a fixture that imports a stage module by its name defined in SOURCES_NAME in the test module. """ @@ -19,3 +21,21 @@ def sources_module(request: pytest.FixtureRequest) -> ModuleType: caller_dir = pathlib.Path(request.node.fspath).parent module_path = caller_dir.parent / sources_name return import_module_from_path("sources", os.fspath(module_path)) + + +@pytest.fixture +def sources_service(sources_module) -> ModuleType: + """sources_service is a fixture that imports a sources module by its name + defined in SOURCES_NAME in the test module and returns a SourcesService + """ + service_cls = None + for memb in inspect.getmembers( + sources_module, + predicate=lambda obj: inspect.isclass(obj) and issubclass( + obj, sources.SourceService)): + if service_cls: + raise ValueError(f"already have {service_cls}, also found {memb}") + service_cls = memb[1] + fd = testutil.make_fake_service_fd() + services_obj = service_cls.from_args(["--service-fd", str(fd)]) + return services_obj diff --git a/sources/test/test_container_storage_source.py b/sources/test/test_container_storage_source.py index 140ab985..090c94d9 100644 --- a/sources/test/test_container_storage_source.py +++ b/sources/test/test_container_storage_source.py @@ -5,7 +5,6 @@ import subprocess import pytest -from osbuild import testutil from osbuild.testutil import has_executable, make_container SOURCES_NAME = "org.osbuild.containers-storage" @@ -13,35 +12,29 @@ SOURCES_NAME = "org.osbuild.containers-storage" @pytest.mark.skipif(not has_executable("podman"), reason="no podman executable") @pytest.mark.skipif(os.getuid() != 0, reason="root only") -def test_containers_storage_integration(tmp_path, sources_module): +def test_containers_storage_integration(tmp_path, sources_service): with make_container(tmp_path, { "file1": "file1 content", }) as base_tag: image_id = subprocess.check_output(["podman", "inspect", "-f", "{{ .Id }}", base_tag], universal_newlines=True).strip() checksum = f"sha256:{image_id}" - fd = testutil.make_fake_service_fd() - cnt_storage = sources_module.ContainersStorageSource.from_args(["--service-fd", str(fd)]) - assert cnt_storage.exists(checksum, None) + assert sources_service.exists(checksum, None) @pytest.mark.skipif(not has_executable("podman"), reason="no podman executable") @pytest.mark.skipif(os.getuid() != 0, reason="root only") -def test_containers_storage_integration_missing(sources_module): +def test_containers_storage_integration_missing(sources_service): checksum = "sha256:1234567890123456789012345678901234567890909b14ffb032aa20fa23d9ad6" - fd = testutil.make_fake_service_fd() - cnt_storage = sources_module.ContainersStorageSource.from_args(["--service-fd", str(fd)]) - assert not cnt_storage.exists(checksum, None) + assert not sources_service.exists(checksum, None) @pytest.mark.skipif(not has_executable("podman"), reason="no podman executable") @pytest.mark.skipif(os.getuid() != 0, reason="root only") -def test_containers_storage_integration_invalid(sources_module): +def test_containers_storage_integration_invalid(sources_service): # put an invalid reference into the source to ensure skopeo errors with # a different error than image not found checksum = "sha256:[" - fd = testutil.make_fake_service_fd() - cnt_storage = sources_module.ContainersStorageSource.from_args(["--service-fd", str(fd)]) with pytest.raises(RuntimeError) as exc: - cnt_storage.exists(checksum, None) + sources_service.exists(checksum, None) assert "unknown skopeo error:" in str(exc) diff --git a/sources/test/test_curl_source.py b/sources/test/test_curl_source.py index 267b668e..4d2dd7d1 100644 --- a/sources/test/test_curl_source.py +++ b/sources/test/test_curl_source.py @@ -7,41 +7,33 @@ import tempfile import pytest -from osbuild import testutil - SOURCES_NAME = "org.osbuild.curl" -def test_curl_source_not_exists(sources_module): - fd = testutil.make_fake_service_fd() - curl_source = sources_module.CurlSource.from_args(["--service-fd", str(fd)]) +def test_curl_source_not_exists(sources_service): tmpdir = tempfile.TemporaryDirectory() - curl_source.cache = tmpdir.name + sources_service.cache = tmpdir.name desc = { "url": "http://localhost:80/a", } checksum = "sha256:1234567890123456789012345678901234567890909b14ffb032aa20fa23d9ad6" - assert not curl_source.exists(checksum, desc) + assert not sources_service.exists(checksum, desc) -def test_curl_source_exists(sources_module): - fd = testutil.make_fake_service_fd() - curl_source = sources_module.CurlSource.from_args(["--service-fd", str(fd)]) +def test_curl_source_exists(sources_service): tmpdir = tempfile.TemporaryDirectory() - curl_source.cache = tmpdir.name + sources_service.cache = tmpdir.name desc = { "url": "http://localhost:80/a", } checksum = "sha256:1234567890123456789012345678901234567890909b14ffb032aa20fa23d9ad6" pathlib.Path(os.path.join(tmpdir.name, checksum)).touch() - assert curl_source.exists(checksum, desc) + assert sources_service.exists(checksum, desc) -def test_curl_source_amend_secrets(sources_module): - fd = testutil.make_fake_service_fd() - curl_source = sources_module.CurlSource.from_args(["--service-fd", str(fd)]) +def test_curl_source_amend_secrets(sources_service): tmpdir = tempfile.TemporaryDirectory() - curl_source.cache = tmpdir.name + sources_service.cache = tmpdir.name desc = { "url": "http://localhost:80/a", "secrets": { @@ -59,17 +51,15 @@ def test_curl_source_amend_secrets(sources_module): cm.callback(cb) checksum = "sha256:1234567890123456789012345678901234567890909b14ffb032aa20fa23d9ad6" pathlib.Path(os.path.join(tmpdir.name, checksum)).touch() - new_desc = curl_source.amend_secrets(checksum, desc) + new_desc = sources_service.amend_secrets(checksum, desc) assert new_desc[1]["secrets"]["ssl_client_key"] == "key" assert new_desc[1]["secrets"]["ssl_client_cert"] == "cert" assert new_desc[1]["secrets"]["ssl_ca_cert"] is None -def test_curl_source_amend_secrets_fail(sources_module): - fd = testutil.make_fake_service_fd() - curl_source = sources_module.CurlSource.from_args(["--service-fd", str(fd)]) +def test_curl_source_amend_secrets_fail(sources_service): tmpdir = tempfile.TemporaryDirectory() - curl_source.cache = tmpdir.name + sources_service.cache = tmpdir.name desc = { "url": "http://localhost:80/a", "secrets": { @@ -79,5 +69,5 @@ def test_curl_source_amend_secrets_fail(sources_module): checksum = "sha256:1234567890123456789012345678901234567890909b14ffb032aa20fa23d9ad6" pathlib.Path(os.path.join(tmpdir.name, checksum)).touch() with pytest.raises(RuntimeError) as exc: - curl_source.amend_secrets(checksum, desc) + sources_service.amend_secrets(checksum, desc) assert "mtls secrets required" in str(exc)