Implement test runner

podctl/console_script.py

@@ -9,11 +9,75 @@ import os
 import sys
 
 from .container import Container
-from .pod import Pod
 from .exceptions import Mistake, WrongResult
+from .pod import Pod
+from .podfile import Podfile
+from .proc import output
 from .service import Service
 
 
+@cli2.option('debug', alias='d', help='Display debug output.')
+async def test(*args, **kwargs):
+    """Run podctl test over a bunch of paths."""
+    report = []
+
+    for arg in args:
+        candidates = [
+            os.path.join(os.getcwd(), arg, 'pod.py'),
+            os.path.join(os.getcwd(), arg, 'pod_test.py'),
+        ]
+        for candidate in candidates:
+            if not os.path.exists(candidate):
+                continue
+            podfile = Podfile.factory(candidate)
+
+            for name, test in podfile.tests.items():
+                name = '::'.join([podfile.path, name])
+                output.print(
+                    '\n\x1b[1;38;5;160;48;5;118m   TEST START \x1b[0m'
+                    + ' ' + name + '\n'
+                )
+
+                try:
+                    await test(podfile.pod)
+                except Exception as e:
+                    report.append((name, False))
+                    output.print('\x1b[1;38;5;15;48;5;196m    TEST FAIL \x1b[0m' + name)
+                else:
+                    report.append((name, True))
+                    output.print('\x1b[1;38;5;200;48;5;44m TEST SUCCESS \x1b[0m' + name)
+                output.print('\n')
+
+    print('\n')
+
+    for name, success in report:
+        if success:
+            output.print('\n\x1b[1;38;5;200;48;5;44m TEST SUCCESS \x1b[0m' + name)
+        else:
+            output.print('\n\x1b[1;38;5;15;48;5;196m    TEST FAIL \x1b[0m' + name)
+
+    print('\n')
+
+    success = [*filter(lambda i: i[1], report)]
+    failures = [*filter(lambda i: not i[1], report)]
+
+    output.print(
+        '\n\x1b[1;38;5;200;48;5;44m TEST TOTAL: \x1b[0m'
+        + str(len(report))
+    )
+    output.print(
+        '\n\x1b[1;38;5;15;48;5;196m    TEST FAIL: \x1b[0m'
+        + str(len(failures))
+    )
+    output.print(
+        '\n\x1b[1;38;5;200;48;5;44m TEST SUCCESS: \x1b[0m'
+        + str(len(success))
+    )
+
+    if failures:
+        console_script.exit_code = 1
+
+
 class ConsoleScript(cli2.ConsoleScript):
     class Parser(cli2.Parser):
         def parse(self):
@@ -38,9 +102,9 @@ class ConsoleScript(cli2.ConsoleScript):
         self.options = dict()
 
     def __call__(self, *args, **kwargs):
-        import inspect
+        podfile = os.getenv('PODFILE', 'pod.py')
-        from podctl.podfile import Podfile
+        if os.path.exists(podfile):
-        self.podfile = Podfile.factory(os.getenv('PODFILE', 'pod.py'))
+            self.podfile = Podfile.factory(podfile)
             for name, script in self.podfile.pod.scripts.items():
                 cb = self.podfile.pod.script(name)
                 cb.__doc__ = inspect.getdoc(script) or script.doc
@@ -59,10 +123,10 @@ class ConsoleScript(cli2.ConsoleScript):
             return super().call(command)
         except Mistake as e:
             print(e)
-            sys.exit(1)
+            self.exit_code = 1
         except WrongResult as e:
             print(e)
-            sys.exit(e.proc.rc)
+            self.exit_code = e.proc.rc
 
 
 console_script = ConsoleScript(__doc__).add_module('podctl.console_script')

podctl/pod.py

@@ -18,6 +18,13 @@ class Pod(Visitable):
         ),
     )
 
+    def script(self, name):
+        async def cb(*args, **kwargs):
+            asyncio.events.get_event_loop()
+            kwargs['pod'] = self
+            return await self.scripts[name].run(*args, **kwargs)
+        return cb
+
     async def down(self, script):
         try:
             await script.exec('podman', 'pod', 'inspect', self.name)
@@ -47,12 +54,5 @@ class Pod(Visitable):
     def containers(self):
         return [i for i in self.visitors if type(i) == Container]
 
-    def script(self, name):
-        async def cb(*args, **kwargs):
-            asyncio.events.get_event_loop()
-            kwargs['pod'] = self
-            return await self.scripts[name].run(*args, **kwargs)
-        return cb
-
     def __repr__(self):
         return self.name

podctl/podfile.py

@@ -6,9 +6,11 @@ from .pod import Pod
 
 
 class Podfile:
-    def __init__(self, pods, containers):
+    def __init__(self, pods, containers, path, tests):
         self.pods = pods
         self.containers = containers
+        self.path = path
+        self.tests = tests
 
         if not self.pods:
             self.pods['pod'] = Pod(*containers.values())
@@ -25,6 +27,7 @@ class Podfile:
     def factory(cls, path):
         containers = dict()
         pods = dict()
+        tests = dict()
         spec = importlib.util.spec_from_file_location('pod', path)
         pod = importlib.util.module_from_spec(spec)
         spec.loader.exec_module(pod)
@@ -35,5 +38,7 @@ class Podfile:
             elif isinstance(value, Pod):
                 pods[name] = value
                 value.name = name
+            elif callable(value) and value.__name__.startswith('test_'):
+                tests[value.__name__] = value
 
-        return cls(pods, containers)
+        return cls(pods, containers, path, tests)

podctl/proc.py

@@ -29,21 +29,18 @@ class Output:
         self.prefix_length = 0
 
     def __call__(self, line, prefix, highlight=True):
-        if prefix not in self.prefixes:
+        if prefix and prefix not in self.prefixes:
             self.prefixes[prefix] = (
                 self.colors[len([*self.prefixes.keys()]) - 1]
             )
             if len(prefix) > self.prefix_length:
                 self.prefix_length = len(prefix)
 
-        prefix_color = self.prefixes[prefix]
+        prefix_color = self.prefixes[prefix] if prefix else ''
-        prefix_padding = '.' * (self.prefix_length - len(prefix) - 2)
+        prefix_padding = '.' * (self.prefix_length - len(prefix) - 2) if prefix else ''
         if prefix_padding:
             prefix_padding = ' ' + prefix_padding + ' '
 
-        if not prefix:
-            breakpoint()
-        #breakpoint()
         sys.stdout.buffer.write((
             (
                 prefix_color
@@ -71,6 +68,13 @@ class Output:
             highlight=False
         )
 
+    def print(self, content):
+        self(
+            content,
+            prefix=None,
+            highlight=False
+        )
+
     def highlight(self, line, highlight=True):
         if not highlight:
             return line

podctl/script.py

@@ -3,6 +3,7 @@ import copy
 import cli2
 import os
 import textwrap
+import subprocess
 import sys
 
 from .proc import output, Proc
@@ -94,9 +95,20 @@ class Script:
 
     async def run(self, *args, **kwargs):
         if self.unshare and os.getuid() != 0:
-            import sys
             # restart under buildah unshare environment !
-            os.execvp('buildah', ['buildah', 'unshare'] + sys.argv)
+            argv = [
+                'buildah', 'unshare',
+                sys.argv[0],  # current podctl location
+                type(self).__name__.lower()  # script name ?
+            ] + list(args)
+            pp = subprocess.Popen(
+                argv,
+                stderr=sys.stderr,
+                stdin=sys.stdin,
+                stdout=sys.stdout,
+            )
+            pp.communicate()
+            return pp.returncode
 
         for key, value in kwargs.items():
             setattr(self, key, value)
@@ -122,4 +134,4 @@ class Script:
             )
             for container in containers
         ]
-        return await asyncio.gather(*procs)
+        await asyncio.gather(*procs)

podctl/scripts.py

@@ -6,7 +6,7 @@ import sys
 
 from .build import Build
 from .exceptions import WrongResult
-from .proc import Proc
+from .proc import output, Proc
 from .script import Script
 
 

podctl/visitors/packages.py

@@ -136,7 +136,8 @@ class Packages:
         return cachedir
 
     async def dnf_setup(self, script):
-        await script.mount(self.cache, f'/var/cache/{self.mgr}')
+        cachedir = os.path.join(self.cache_root, self.mgr)
+        await script.mount(cachedir, f'/var/cache/{self.mgr}')
         await script.run('echo keepcache=True >> /etc/dnf/dnf.conf')
 
     async def apt_setup(self, script):