inputs: add inputs_service fixture

This new fixture avoids some repetitivness when building tests
for input services.

inputs/test/conftest.py

@@ -1,14 +1,16 @@
+import inspect
 import os
 import pathlib
 from types import ModuleType
 
 import pytest
 
+from osbuild import inputs, testutil
 from osbuild.testutil.imports import import_module_from_path
 
 
-@pytest.fixture
-def inputs_module(request: pytest.FixtureRequest) -> ModuleType:
+@pytest.fixture(name="inputs_module")
+def inputs_module_fixture(request: pytest.FixtureRequest) -> ModuleType:
     """inputs_module is a fixture that imports a stage module by its name
     defined in INPUTS_NAME in the test module.
     """
@@ -19,3 +21,21 @@ def inputs_module(request: pytest.FixtureRequest) -> ModuleType:
     caller_dir = pathlib.Path(request.node.fspath).parent
     module_path = caller_dir.parent / inputs_name
     return import_module_from_path("inputs", os.fspath(module_path))
+
+
+@pytest.fixture
+def inputs_service(inputs_module) -> ModuleType:
+    """inputs_service is a fixture that imports a inputs module by its name
+    defined in INPUTS_NAME in the test module and returns a InputService
+    """
+    service_cls = None
+    for memb in inspect.getmembers(
+            inputs_module,
+            predicate=lambda obj: inspect.isclass(obj) and issubclass(
+                obj, inputs.InputService)):
+        if service_cls:
+            raise ValueError(f"already have {service_cls}, also found {memb}")
+        service_cls = memb[1]
+    fd = testutil.make_fake_service_fd()
+    srv_obj = service_cls.from_args(["--service-fd", str(fd)])
+    return srv_obj

inputs/test/test_containers.py

@@ -7,7 +7,6 @@ import tempfile
 
 import pytest
 
-from osbuild import testutil
 from osbuild.testutil import has_executable, make_container
 
 INPUTS_NAME = "org.osbuild.containers-storage"
@@ -26,7 +25,7 @@ class FakeStoreClient:
 
 @pytest.mark.skipif(not has_executable("podman"), reason="no podman executable")
 @pytest.mark.skipif(os.getuid() != 0, reason="root only")
-def test_containers_local_inputs_integration(tmp_path, inputs_module):
+def test_containers_local_inputs_integration(tmp_path, inputs_service):
     with make_container(tmp_path, {"file1": "file1 content"}) as base_tag:
         image_id = subprocess.check_output(
             ["podman", "inspect", "-f", "{{ .Id }}", base_tag],
@@ -40,8 +39,6 @@ def test_containers_local_inputs_integration(tmp_path, inputs_module):
                 }
             }
         }
-        fd = testutil.make_fake_service_fd()
-        cnt_inputs = inputs_module.ContainersStorageInput.from_args(["--service-fd", str(fd)])
         store = FakeStoreClient(tmp_path / "fake-sources")
         # not using "tmp_path" here as it will "rm -rf" on cleanup and
         # that is dangerous as during the tests we bind mount the
@@ -49,12 +46,12 @@ def test_containers_local_inputs_integration(tmp_path, inputs_module):
         target = pathlib.Path(tempfile.TemporaryDirectory("cnt-target").name)
         options = None
         try:
-            reply = cnt_inputs.map(store, inputs["origin"], inputs["references"], target, options)
+            reply = inputs_service.map(store, inputs["origin"], inputs["references"], target, options)
             assert reply["path"] == target
             assert len(reply["data"]["archives"]) == 1
             assert (target / "storage").exists()
         finally:
-            cnt_inputs.unmap()
+            inputs_service.unmap()
             # cleanup manually, note that we only remove empty dirs here,
             # because we only expect a bind mount under "$target/storage"
             # Anything non-empty here means a umount() failed