sources: add new sources_service fixture

Similar to the previous commit to include a `inputs_service` fixture
this does the same for `source.SourcesService` imports.

Note that we cannot easily share the helpers so we have to life with
a bit of very similar but duplicated code. To fix this we would have
to have a shared confftest.py that pytest can find. Which would mean
that we need to put the tests under a common dir that is reachable
via __init__.py files (which we currently not have because stages,
inputs etc do not have a __init__.py so python does not considers
them modules).

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

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)

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)