Improving code coverage

Dockerfile

@@ -2,7 +2,7 @@ ARG OUTPUT=run
 
 FROM python:3.9 as builder
 
-RUN pip install flake8
+RUN pip install flake8 coverage
 
 COPY requirements.txt /src/requirements.txt
 WORKDIR /src
@@ -15,6 +15,7 @@ COPY tests /src/tests/
 
 RUN flake8 . --count  --max-complexity=10 --max-line-length=127 --show-source --statistics
 RUN python -m unittest
+RUN coverage run --source ./runci --timid -m unittest && coverage report -m
 
 RUN pip install .
 
@@ -26,12 +27,12 @@ FROM builder as artifact
 CMD ["pyinstaller", "-F", "--distpath", "/out", "-n", "runci", "runci.spec"]
 
 
-FROM python:3.9 as run
+FROM python:3.9-alpine as run
 ARG EXTENSION=
 ENV LC_ALL=C.UTF-8
 ENV LANG=C.UTF-8
 COPY --from=builder /src/dist/runci${EXTENSION} /
 COPY docker-entrypoint.sh /docker-entrypoint.sh
 ENTRYPOINT ["/docker-entrypoint.sh"]
 
-FROM ${OUTPUT}
+FROM ${OUTPUT}

runci.yml

@@ -1,4 +1,4 @@
-version: "2.7"
+version: "2.4"
 
 x-variables: ## "x-" is facultative. Allows to be used directly by docker-compose
 

runci/cli/main.py

@@ -1,10 +1,13 @@
+import asyncio
 import click
 import logging
+import os
+import sys
+
 from runci.dal.yaml import load_config
 from runci.entities.parameters import Parameters
 from runci.engine import core
 from runci.engine.job import JobStatus
-import asyncio
 
 DEFAULT_CONFIG_FILE = "runci.yml"
 
@@ -13,16 +16,35 @@
 @click.option('-f', '--file', 'file', type=click.File('r', lazy=True), default=DEFAULT_CONFIG_FILE)
 @click.argument('targets', nargs=-1)
 def main(targets, file):
-    parameters = Parameters(file.name, targets, 1)
+    if hasattr(file, 'name') \
+       and isinstance(file.name, str) \
+       and os.path.isfile(file.name):
+        # If filename available and file exists, load file to allow docker-compose integration
+        parameters = Parameters(file.name, targets, 1)
+    else:
+        parameters = Parameters(file, targets, 1)
     project = load_config(parameters)
 
     logging.debug("Building the following targets: %s" % str.join(" ", targets))
     unknown_targets = [t for t in targets if t not in [t.name for t in project.targets]]
     if any(unknown_targets):
-        logging.error("Unkown targets: %s" % str.join(" ", unknown_targets))
-        return 1
+        print("Unkown targets: %s" % str.join(" ", unknown_targets), file=sys.stderr)
+        exit(1)
+
+    result = asyncio.run(run_project(project))
 
-    asyncio.run(run_project(project))
+    if result == JobStatus.SUCCEEDED:
+        print("Pipeline has run succesfully.")
+        exit(0)
+    elif result == JobStatus.FAILED:
+        print("Pipeline has failed.", file=sys.stderr)
+        exit(1)
+    elif result == JobStatus.CANCELED:
+        print("Pipeline has been canceled.", file=sys.stderr)
+        exit(2)
+    else:
+        print("Pipeline has been run but outcome is undetermined. Please report this as a bug.", file=sys.stderr)
+        exit(3)
 
 
 async def run_project(project):
@@ -42,4 +64,5 @@ async def run_project(project):
         else:
             await asyncio.sleep(0.1)
 
-    return await task
+    await task
+    return tree.status

runci/dal/yaml.py

@@ -52,7 +52,8 @@ def load_config(parameters):
     data = safe_load(datastream)
     datastream.close()
 
-    services = config.create_entities(__create_service, data['services'].items())
+    services_data = data.get('services', {})
+    services = config.create_entities(__create_service, services_data.items())
 
     if "targets" in data:
         targets = data['targets']

runci/engine/core.py

@@ -2,7 +2,7 @@
 from collections import namedtuple
 
 from runci.entities import config
-from runci.engine.job import Job
+from runci.engine.job import Job, JobStatus
 
 
 class RunCIEngineException(Exception):
@@ -21,35 +21,36 @@ class DependencyNode(namedtuple("dependency_node", "job dependencies")):
     "runci depencency tree node"
     async def _run_dependencies(self, noparallel):
         if noparallel:
-            ret = [job
-                   for node in self.dependencies
-                   for job in await node._run(noparallel)]
+            jobs = [job
+                    for node in self.dependencies
+                    for job in await node.start(noparallel)]
         else:
-            tasks = [node._run(noparallel) for node in self.dependencies]
+            tasks = [node.start(noparallel) for node in self.dependencies]
             jobs = [job
                     for task in tasks
                     for job in await task]
-            return jobs
 
-        return ret
+        return jobs
 
     async def _run(self, noparallel=False):
-        ret = [job for job in list(await self._run_dependencies(noparallel))]
-        if self.job is not None:
-            ret.append(self.job)
+        jobs = list(await self._run_dependencies(noparallel))
+        if any([job for job in jobs if job.status in [JobStatus.FAILED, JobStatus.CANCELED]]):
+            self.job.cancel()
+        else:
+            jobs.append(self.job)
             await self.job.start()
 
-        return ret
+        return jobs
 
     def start(self, noparallel=False):
         return asyncio.create_task(self._run(noparallel))
 
     def run(self, noparallel=False):
         return asyncio.run(self._run(noparallel))
 
-    def wait(self):
-        if self.job is not None:
-            return self.job.wait()
+    @property
+    def status(self):
+        return self.job.status
 
 
 class DependencyTree():
@@ -115,3 +116,7 @@ def start(self, noparallel=False):
 
     def run(self, noparallel=False):
         return self._root_node.run(noparallel)
+
+    @property
+    def status(self):
+        return self.root_node.status

runci/engine/job.py

@@ -3,10 +3,9 @@
 from collections import namedtuple
 from datetime import datetime
 import sys
-import traceback
 
 from runci.entities import config
-from runci.engine.runner import ComposeBuildRunner
+from runci.engine import runner
 
 
 class JobStatus(Enum):
@@ -35,7 +34,7 @@ def release(self):
         if self.stream in self._valid_streams:
             print(message, file=self.stream)
         else:
-            print("Unknown stream: " + self.stream, sys.stderr)
+            print("Unknown stream: " + self.stream, file=sys.stderr)
             print("Message: " + message, file=sys.stderr)
 
 
@@ -59,30 +58,25 @@ def _log_message(self, output_stream, message):
 
     async def _start(self):
         if self._status == JobStatus.CREATED:
+            self._status = JobStatus.STARTED
             self._messages = asyncio.Queue()
-            try:
-                self._status = JobStatus.STARTED
-
-                for step in self._target.steps:
-                    if step.type == 'compose-build':
-                        step_runner = ComposeBuildRunner(self._log_message, step.spec)
-                        await step_runner.run(self._project)
-                        if step_runner.is_succeeded:
-                            self._log_message(sys.stderr, 'Step %s failed.' % step.name)
-                            self._status = JobStatus.FAILED
-                            break
-                    else:
-                        self._log_message(sys.stderr, 'Unknown step type: %s' % step.type)
+
+            for step in self._target.steps:
+                step_runner_cls = runner.selector.get(step.type, None)
+                if step_runner_cls is not None:
+                    step_runner = step_runner_cls(self._log_message, step.spec)
+                    await step_runner.run(self._project)
+                    if not step_runner.is_succeeded:
+                        self._log_message(sys.stderr, 'Step %s failed.' % step.name)
                         self._status = JobStatus.FAILED
                         break
-            except Exception:
-                self._log_message(sys.stderr, traceback.format_exc())
-                self._status = JobStatus.FAILED
-            else:
-                self._status = JobStatus.SUCCEEDED
+                else:
+                    self._log_message(sys.stderr, 'Unknown step type: %s' % step.type)
+                    self._status = JobStatus.FAILED
+                    break
 
-    def wait(self):
-        return asyncio.ensure_future(self._task)
+            if self._status == JobStatus.STARTED:
+                self._status = JobStatus.SUCCEEDED
 
     def start(self):
         self._task = asyncio.create_task(self._start())
@@ -92,7 +86,9 @@ def run(self):
         return asyncio.run(self._start())
 
     def cancel(self):
-        if self._status == JobStatus.CREATED or self._status == JobStatus.STARTED:
+        if self._status == JobStatus.CREATED:
+            self._status = JobStatus.CANCELED
+        elif self._status == JobStatus.STARTED:
             self._task.cancel()
             self._status = JobStatus.CANCELED
 

runci/engine/runner/__init__.py

@@ -1,4 +1,8 @@
-from .base import RunnerBase
+from .base import RunnerBase, RunnerStatus
 from .compose_build_runner import ComposeBuildRunner
 
-__all__ = ["RunnerBase", "ComposeBuildRunner"]
+selector = {
+    'compose-build': ComposeBuildRunner
+}
+
+__all__ = ["selector", "RunnerStatus", "RunnerBase", "ComposeBuildRunner"]

runci/engine/runner/base.py

@@ -24,12 +24,12 @@ class RunnerStatus(Enum):
 
 class RunnerBase():
     spec: dict
-    status: RunnerStatus
+    _status: RunnerStatus
     _message_logger: asyncio.Queue
 
     def __init__(self, message_logger: Callable, spec: dict):
         self.spec = spec
-        self.status = RunnerStatus.CREATED
+        self._status = RunnerStatus.CREATED
         self._message_logger = message_logger
 
     def _log_message(self, output_stream, message):
@@ -47,31 +47,32 @@ async def run_internal(self, project: Project):
         pass
 
     async def run(self, project: Project):
-        self.status = RunnerStatus.STARTED
+        self._status = RunnerStatus.STARTED
         self._log_message(sys.stdout, "Starting %s runner\n" % type(self).__name__)
         try:
             await self.run_internal(project)
         except Exception:
-            self._log_message(sys.stderr, "Runner %s failed:" % type(self))
+            self._log_message(sys.stderr, "Runner %s failed:" % type(self).__name__)
             self._log_message(sys.stderr, traceback.format_exc())
-            self.status = RunnerStatus.FAILED
+            self._status = RunnerStatus.FAILED
 
-        if self.status == RunnerStatus.STARTED:
-            self.status = RunnerStatus.SUCCEEDED
+        if self._status == RunnerStatus.STARTED:
+            self._status = RunnerStatus.SUCCEEDED
 
     async def _run_process(self, args):
         self._log_message(sys.stdout, "Running command: %s\n" % str.join(" ", args))
         process = await asyncio.create_subprocess_exec(args[0], *args[1:], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 
-        await asyncio.wait([self._log_stream(sys.stdout, process.stdout),
-                            self._log_stream(sys.stderr, process.stderr),
-                            process.wait()])
+        await asyncio.wait([asyncio.create_task(coro)
+                            for coro in [self._log_stream(sys.stdout, process.stdout),
+                                         self._log_stream(sys.stderr, process.stderr),
+                                         process.wait()]])
 
         if process.returncode == 0:
-            self.status = RunnerStatus.SUCCEEDED
+            self._status = RunnerStatus.SUCCEEDED
         else:
-            self.status = RunnerStatus.FAILED
+            self._status = RunnerStatus.FAILED
 
     @property
     def is_succeeded(self):
-        return self.status == RunnerStatus.SUCCEEDED
+        return self._status == RunnerStatus.SUCCEEDED