Try a context manager API to start systems. refs: #29

Author: spanezz

moncic/imagestorage.py

@@ -8,10 +8,10 @@
 import subprocess
 import tempfile
 from collections import defaultdict
-from typing import TYPE_CHECKING, Generator, List
+from typing import TYPE_CHECKING, ContextManager, Generator, List
 
 from .btrfs import do_dedupe, is_btrfs
-from .system import System, SystemConfig
+from .system import MaintenanceSystem, System, SystemConfig
 from .utils import is_on_rotational, pause_automounting
 
 if TYPE_CHECKING:
@@ -47,10 +47,36 @@ def list_images(self) -> List[str]:
 
     def create_system(self, name: str) -> System:
         """
-        Instantiate a System from its name or path
+        Instantiate a System by its name
         """
         raise NotImplementedError(f"{self.__class__.__name__}.create_system is not implemented")
 
+    def system(self, name: str) -> ContextManager[System]:
+        """
+        Instantiate a System that can only be used for the duration
+        of this context manager.
+        """
+        # raise NotImplementedError(f"{self.__class__.__name__}.maintenance_system is not implemented")
+        @contextlib.contextmanager
+        def res():
+            yield self.create_system(name)
+        return res()
+
+    def maintenance_system(self, name: str) -> ContextManager[MaintenanceSystem]:
+        """
+        Instantiate a MaintenanceSystem that can only be used for the duration
+        of this context manager.
+
+        This allows maintenance to be transactional, limited to backends that
+        support it, so that errors in the maintenance roll back to the previous
+        state and do not leave an inconsistent OS image
+        """
+        # raise NotImplementedError(f"{self.__class__.__name__}.maintenance_system is not implemented")
+        @contextlib.contextmanager
+        def res():
+            yield self.create_system(name)
+        return res()
+
     def add_dependencies(self, images: List[str]) -> List[str]:
         """
         Add dependencies to the given list of images, returning the extended

moncic/system.py

@@ -298,3 +298,9 @@ def create_container(
         # Import here to avoid an import loop
         from .container import NspawnContainer
         return NspawnContainer(self, config, instance_name)
+
+
+class MaintenanceSystem(System):
+    """
+    System used to do maintenance on an OS image
+    """

tests/test_machine.py

@@ -21,32 +21,35 @@ class RunTestCase:
     @classmethod
     def setUpClass(cls):
         super().setUpClass()
-        cls.exit_stack = contextlib.ExitStack()
-        # cls.workdir = cls.exit_stack.enter_context(tempfile.TemporaryDirectory())
+        cls.cls_exit_stack = contextlib.ExitStack()
+        # cls.workdir = cls.cls_exit_stack.enter_context(tempfile.TemporaryDirectory())
         # cls.moncic = make_moncic(imagedir=cls.workdir)
         cls.moncic = make_moncic()
-        cls.images = cls.exit_stack.enter_context(cls.moncic.images())
+        cls.images = cls.cls_exit_stack.enter_context(cls.moncic.images())
 
     @classmethod
     def tearDownClass(cls):
-        cls.exit_stack.close()
+        cls.cls_exit_stack.close()
         cls.images = None
         cls.moncic = None
         # cls.workdir = None
-        cls.exit_stack = None
+        cls.cls_exit_stack = None
         super().tearDownClass()
 
     def setUp(self):
         super().setUp()
+        self.exit_stack = contextlib.ExitStack()
         with privs.root():
             if self.distro_name not in self.images.list_images():
                 raise unittest.SkipTest(f"Image {self.distro_name} not available")
-            self.system = self.images.create_system(self.distro_name)
+            self.system = self.exit_stack.enter_context(self.images.system(self.distro_name))
             if not os.path.exists(self.system.path):
                 raise unittest.SkipTest(f"Image {self.distro_name} has not been bootstrapped")
 
     def tearDown(self):
         self.system = None
+        self.exit_stack.close()
+        self.exit_stack = None
         super().tearDown()
 
     def test_true(self):