wip

inputs/org.osbuild.containers-storage

@@ -99,7 +99,7 @@ class ContainersStorageInput(inputs.InputService):
         os.makedirs(dest, exist_ok=True)
         self.mg.mount(source, dest)
 
-    def map(self, store, origin, refs, target, _options):
+    def map(self, store, origin, refs, target, options):
         self.bind_mount_local_storage(target)
 
         images = {}

inputs/org.osbuild.files

@@ -196,7 +196,7 @@ class FilesInput(inputs.InputService):
         data = data.get("metadata", {})
         return ref, data
 
-    def map(self, store, origin, refs, target, _options):
+    def map(self, store, origin, refs, target, options):
 
         source = store.source("org.osbuild.files")
         files = {}

inputs/org.osbuild.ostree

@@ -117,7 +117,7 @@ def export(checksums, cache, output):
 
 class OSTreeInput(inputs.InputService):
 
-    def map(self, store, origin, refs, target, _options):
+    def map(self, store, origin, refs, target, options):
 
         if origin == "org.osbuild.pipeline":
             for ref, options in refs.items():

inputs/org.osbuild.ostree.checkout

@@ -97,7 +97,7 @@ def checkout(checksums, cache, output):
 
 class OSTreeCheckoutInput(inputs.InputService):
 
-    def map(self, store, origin, refs, target, _options):
+    def map(self, store, origin, refs, target, options):
 
         ids = []
 

inputs/org.osbuild.tree

@@ -72,7 +72,7 @@ SCHEMA = """
 
 class TreeInput(inputs.InputService):
 
-    def map(self, store, _origin, refs, target, _options):
+    def map(self, store, origin, refs, target, options):
 
         # input verification *must* have been done via schema
         # verification. It is expected that origin is a pipeline

osbuild/devices.py

@@ -96,7 +96,7 @@ def open(self, dev: Device) -> Dict:
         return res
 
 
-class DeviceService(host.Service):
+class DeviceService(host.DispatchMixin, host.Service):
     """Device host service"""
 
     @staticmethod
@@ -122,16 +122,3 @@ def close(self):
 
     def stop(self):
         self.close()
-
-    def dispatch(self, method: str, args, _fds):
-        if method == "open":
-            r = self.open(args["dev"],
-                          args["parent"],
-                          args["tree"],
-                          args["options"])
-            return r, None
-        if method == "close":
-            r = self.close()
-            return r, None
-
-        raise host.ProtocolError("Unknown method")

osbuild/host.py

@@ -41,6 +41,7 @@ class of this module can be used, which sets up and handles the
 import fcntl
 import importlib
 import io
+import json
 import os
 import signal
 import subprocess
@@ -550,3 +551,29 @@ def __exit__(self, *args):
         self.event_loop.call_soon_threadsafe(self.event_loop.stop)
         self.thread.join()
         self.event_loop.close()
+
+
+class DispatchMixin:
+
+    def dispatch(self, method, args, fds):
+        with os.fdopen(fds.steal(0)) as f:
+            args = json.load(f)
+
+        pre_fn = getattr(self, "pre_" + method, None)
+        if pre_fn:
+            pre_fn(args, fds)
+
+        fn_with_fds = getattr(self, method + "_with_fds", None)
+        if fn_with_fds:
+            r = fn_with_fds(**args, fds=FdSet.from_list(list(fds._fds[2:])))
+        else:
+            fn = getattr(self, method, None)
+            if fn is None:
+                host.ProtocolError(f"Unknown method {method}")
+            r = fn(**args)
+
+        with os.fdopen(fds.steal(1), "w") as f:
+            f.write(json.dumps(r))
+            f.seek(0)
+        # XXX: remove entirely?
+        return "{}", None

osbuild/inputs.py

@@ -116,27 +116,7 @@ def make_args_and_reply_files(tmp, args):
         yield f_args.fileno(), f_reply.fileno()
 
 
-class DispatchMixin:
-    def dispatch(self, method, args, fds):
-        with os.fdopen(fds.steal(0)) as f:
-            args = json.load(f)
-
-        pre_fn = getattr(self, "pre_" + method, None)
-        if pre_fn:
-            pre_fn(args, fds)
-        fn = getattr(self, method, None)
-        if fn is None:
-            host.ProtocolError(f"Unknown method {method}")
-        r = fn(**args)
-
-        with os.fdopen(fds.steal(1), "w") as f:
-            f.write(json.dumps(r))
-            f.seek(0)
-        # XXX: remove entirely?
-        return "{}", None
-
-
-class InputService(DispatchMixin, host.Service):
+class InputService(host.DispatchMixin, host.Service):
     """Input host service"""
 
     @abc.abstractmethod

osbuild/mounts.py

@@ -112,7 +112,7 @@ def mount(self, mount: Mount) -> Dict:
         return {"path": path}
 
 
-class MountService(host.Service):
+class MountService(host.DispatchMixin, host.Service):
     """Mount host service"""
 
     @abc.abstractmethod
@@ -126,13 +126,6 @@ def umount(self):
     def stop(self):
         self.umount()
 
-    def dispatch(self, method: str, args, _fds):
-        if method == "mount":
-            r = self.mount(args)
-            return r, None
-
-        raise host.ProtocolError("Unknown method")
-
 
 class FileSystemMountService(MountService):
     """Specialized mount host service for file system mounts"""

osbuild/sources.py

@@ -79,7 +79,7 @@ def stages(self):
         return []
 
 
-class SourceService(host.Service):
+class SourceService(host.DispatchMixin, host.Service):
     """Source host service"""
 
     max_workers = 1
@@ -105,22 +105,9 @@ def exists(self, checksum, _desc) -> bool:
         """Returns True if the item to download is in cache. """
         return os.path.isfile(f"{self.cache}/{checksum}")
 
-    @staticmethod
-    def load_items(fds):
-        with os.fdopen(fds.steal(0)) as f:
-            items = json.load(f)
-        return items
-
-    def setup(self, args):
-        self.cache = os.path.join(args["cache"], self.content_type)
+    def download_with_fds(self, cache, options, fds):
+        self.cache = os.path.join(cache, self.content_type)
         os.makedirs(self.cache, exist_ok=True)
-        self.options = args["options"]
-
-    def dispatch(self, method: str, args, fds):
-        if method == "download":
-            self.setup(args)
-            with tempfile.TemporaryDirectory(prefix=".unverified-", dir=self.cache) as self.tmpdir:
-                self.fetch_all(SourceService.load_items(fds))
-                return None, None
-
-        raise host.ProtocolError("Unknown method")
+        self.options = options
+        with tempfile.TemporaryDirectory(prefix=".unverified-", dir=self.cache) as self.tmpdir:
+            self.fetch_all(SourceService.load_items(fds))

test/mod/test_inputs.py

@@ -13,7 +13,7 @@ def __init__(self, *args, **kwargs):
         # do not call "super().__init__()" here to make it testable
         self._map_calls = []
 
-    def map(self, _store, origin, refs, target, options):
+    def map(self, store, origin, refs, target, options):
         self._map_calls.append([origin, refs, target, options])
         return "complex", 2, "reply"