fiotest: add support for test execution context

Key-value dictionary can now be passed as environment to the test
execution routine. This allows to write test scripts which accept
external variables for making pass/fail decisions.

Signed-off-by: Milosz Wasilewski <milosz.wasilewski@foundries.io>

Author: mwasilew

docker-compose.yml

@@ -13,5 +13,6 @@ services:
       - ${SOTA_DIR-/var/sota}:/var/sota
       - ${TEST_SPEC-./test-spec.yml}:/test-spec.yml
       - ${FIOTEST_DIR-/var/lib/fiotest}:/var/lib/fiotest
+      - ${ETC_DIR-/etc}:/var/etc
       # Uncomment for devices registered with softhsm
       # - /var/lib/softhsm/:/var/lib/softhsm/

fiotest/api.py

@@ -1,11 +1,14 @@
 import json
+import logging
 import os
 import requests
 import sys
 from typing import Optional
 
 from fiotest.gateway_client import DeviceGatewayClient
 
+log = logging.getLogger()
+
 
 def status(msg: str, prefix: str = "== "):
     """Print a commonly formatted status message to stdout.
@@ -86,19 +89,38 @@ def complete_test(
 
     @staticmethod
     def target_name(sota_dir: str) -> str:
-        with open(os.path.join(sota_dir, "current-target")) as f:
-            for line in f:
-                if line.startswith("TARGET_NAME"):
-                    k, v = line.split("=")
-                    return v.replace('"', "").strip()  # remove spaces and quotes
-        sys.exit("Unable to find current target")
+        try:
+            with open(os.path.join(sota_dir, "current-target")) as f:
+                for line in f:
+                    if line.startswith("TARGET_NAME"):
+                        k, v = line.split("=")
+                        return v.replace('"', "").strip()  # remove spaces and quotes
+        except FileNotFoundError:
+            sys.exit("Unable to find current target")
 
     @staticmethod
     def test_url(sota_dir: str) -> str:
-        with open(os.path.join(sota_dir, "sota.toml")) as f:
-            for line in f:
-                if line.startswith("server ="):
+        try:
+            with open(os.path.join(sota_dir, "sota.toml")) as f:
+                for line in f:
+                    if line.startswith("server ="):
+                        k, v = line.split("=")
+                        v = v.replace('"', "").strip()  # remove spaces and quotes
+                        return v + "/tests"
+        except FileNotFoundError:
+            pass  # ignore the error and exit
+        sys.exit("Unable to find server url")
+
+    @staticmethod
+    def file_variables(sota_dir: str, file_name: str) -> dict:
+        ret_dict = {}
+        try:
+            with open(os.path.join(sota_dir, file_name)) as f:
+                for line in f:
                     k, v = line.split("=")
                     v = v.replace('"', "").strip()  # remove spaces and quotes
-                    return v + "/tests"
-        sys.exit("Unable to find server url")
+                    ret_dict.update({k: v})
+        except FileNotFoundError:
+            log.warning(f"File {file_name} not found in {sota_dir}")
+            pass
+        return ret_dict

fiotest/runner.py

@@ -1,6 +1,7 @@
 import json
 import logging
-from os import execv, unlink
+import requests
+import os
 import subprocess
 from threading import Thread
 from time import sleep
@@ -40,7 +41,7 @@ def run(self):
                     "Detectected rebooted sequence, continuing after sequence %d",
                     completed,
                 )
-            unlink(self.reboot_state)
+            os.unlink(self.reboot_state)
             self.api.complete_test(data["test_id"], {})
         except FileNotFoundError:
             pass  # This is the "normal" case - no reboot has occurred
@@ -84,9 +85,34 @@ def _reboot(self, seq_idx: int, reboot: Reboot):
         with open(self.reboot_state, "w") as f:
             state = {"seq_idx": seq_idx + 1, "test_id": test_id}
             json.dump(state, f)
-        execv(reboot.command[0], reboot.command)
+        os.execv(reboot.command[0], reboot.command)
+
+    def _prepare_context(self, context: dict):
+        return_dict = {}
+        if "url" in context.keys():
+            context_dict = {}
+            if os.path.exists("/var/sota/current-target"):
+                context_dict = API.file_variables("/var/sota/", "current-target")
+            if os.path.exists("/var/etc/os-release"):
+                context_dict.update(API.file_variables("/var/etc/", "os-release"))
+            target_url = context["url"]
+            try:
+                target_url = target_url.format(**context_dict)
+            except KeyError:
+                # ignore any missing keys
+                pass
+            log.info(f"Retrieving context from {target_url}")
+            env_response = requests.get(target_url)
+            if env_response.status_code == 200:
+                return_dict = env_response.json()
+        else:
+            return_dict = context
+        return return_dict
 
     def _run_test(self, test: Test):
+        environment = os.environ.copy()
+        if test.context:
+            environment.update(self._prepare_context(test.context))
         host_ip = netifaces.gateways()["default"][netifaces.AF_INET][0]
         args = ["/usr/local/bin/fio-test-wrap", test.name]
         if test.on_host:
@@ -100,9 +126,15 @@ def _run_test(self, test: Test):
                     "fio@" + host_ip,
                 ]
             )
+            # pass environment to host
+            # this only works if PermitUserEnvironment is enabled
+            # on host
+            with open("~/.ssh/environment", "w") as sshfile:
+                for env_name, env_value in environment.items():
+                    sshfile.write(f"{env_name}={env_value}\n")
         args.extend(test.command)
         with open("/tmp/tmp.log", "wb") as f:
-            p = subprocess.Popen(args, stderr=f, stdout=f)
+            p = subprocess.Popen(args, stderr=f, stdout=f, env=environment)
             while p.poll() is None:
                 if not self.running:
                     log.info("Killing test")

fiotest/spec.py

@@ -7,6 +7,7 @@ class Test(BaseModel):
     name: str
     command: List[str]
     on_host: bool = False
+    context: Optional[dict]
 
 
 class Reboot(BaseModel):

fiotest/tests.py

@@ -0,0 +1,117 @@
+import unittest
+import yaml
+
+from time import sleep
+from unittest.mock import patch, MagicMock, PropertyMock
+
+from fiotest.main import Coordinator
+from fiotest.runner import SpecRunner
+from fiotest.spec import TestSpec
+
+SIMPLE_TEST_SPEC = """
+sequence:
+  - tests:
+      - name: test1
+        context:
+            url: https://example.com/{LMP_FACTORY}/{CUSTOM_VERSION}/{LMP_MACHINE}
+        command:
+          - /bin/true
+        on_host: true
+      - name: test2
+        command:
+          - /bin/true
+"""
+
+class TestMain(unittest.TestCase):
+    def setUp(self):
+        data = yaml.safe_load(SIMPLE_TEST_SPEC)
+        self.testspec = TestSpec.parse_obj(data)
+
+    @patch("fiotest.host.sudo_execute", return_value=0)
+    def test_coordinator_check_for_updates_pre(self, mock_sudo_execute):
+        self.coordinator = Coordinator(self.testspec)
+        self.assertEqual(False, self.coordinator.callbacks_enabled)
+        self.assertEqual(True, self.coordinator.timer.is_alive())
+        self.coordinator.on_check_for_updates_pre("foo")
+        sleep(1)  # it takes a moment for thread to complete
+        self.assertEqual(True, self.coordinator.callbacks_enabled)
+        self.assertEqual(False, self.coordinator.timer.is_alive())
+
+    @patch("fiotest.main.SpecRunner")
+    @patch("fiotest.host.sudo_execute", return_value=0)
+    def test_coordinator_on_install_post_ok(self, mock_sudo_execute, mock_specrunner):
+        mock_runner = MagicMock()
+        mock_specrunner.return_value = mock_runner
+        type(mock_specrunner).reboot_state = PropertyMock(return_value="/foo/bar")
+
+        self.coordinator = Coordinator(self.testspec)
+        self.coordinator.on_check_for_updates_pre("foo")
+        self.coordinator.on_install_post("foo", "OK")
+        mock_specrunner.assert_called_once_with(self.testspec)
+        mock_runner.start.assert_called_once()
+
+    @patch("fiotest.main.SpecRunner")
+    @patch("fiotest.host.sudo_execute", return_value=0)
+    def test_coordinator_on_install_post_fail(self, mock_sudo_execute, mock_specrunner):
+        mock_runner = MagicMock()
+        mock_specrunner.return_value = mock_runner
+        type(mock_specrunner).reboot_state = PropertyMock(return_value="/foo/bar")
+
+        self.coordinator = Coordinator(self.testspec)
+        self.coordinator.on_check_for_updates_pre("foo")
+        self.coordinator.on_install_post("foo", "FAIL")
+        mock_specrunner.assert_not_called()
+        mock_runner.start.assert_not_called()
+
+    @patch("fiotest.main.SpecRunner")
+    @patch("fiotest.host.sudo_execute", return_value=0)
+    def test_coordinator_on_install_pre_no_runner(self, mock_sudo_execute, mock_specrunner):
+        mock_runner = MagicMock()
+        mock_specrunner.return_value = mock_runner
+        type(mock_specrunner).reboot_state = PropertyMock(return_value="/foo/bar")
+
+        self.coordinator = Coordinator(self.testspec)
+        self.coordinator.on_check_for_updates_pre("foo")
+        self.coordinator.on_install_pre("foo")
+        mock_runner.stop.assert_not_called()
+
+    @patch("fiotest.main.SpecRunner")
+    @patch("fiotest.host.sudo_execute", return_value=0)
+    def test_coordinator_on_install_pre(self, mock_sudo_execute, mock_specrunner):
+        mock_runner = MagicMock()
+        mock_specrunner.return_value = mock_runner
+        type(mock_specrunner).reboot_state = PropertyMock(return_value="/foo/bar")
+
+        self.coordinator = Coordinator(self.testspec)
+        self.coordinator.on_check_for_updates_pre("foo")
+        self.coordinator.on_install_post("foo", "OK")
+        mock_specrunner.assert_called_once_with(self.testspec)
+        mock_runner.start.assert_called_once()
+        self.coordinator.on_install_pre("foo")
+        mock_runner.stop.assert_called()
+
+
+class TestRunner(unittest.TestCase):
+    def setUp(self):
+        data = yaml.safe_load(SIMPLE_TEST_SPEC)
+        self.testspec = TestSpec.parse_obj(data)
+
+    @patch("fiotest.api.API.test_url", return_value="https://example.com/{FOO}")
+    @patch("fiotest.runner.API.file_variables", return_value={"LMP_FACTORY": "factory", "CUSTOM_VERSION": 123, "LMP_MACHINE": "foo"})
+    @patch("fiotest.api.DeviceGatewayClient")
+    @patch("requests.get")
+    @patch("subprocess.Popen")
+    @patch("os.path.exists", return_value=True)
+    def test_run(self, mock_path_exists, mock_popen, mock_requests_get, mock_gateway_client, mock_file_variables, mock_test_url):
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_requests_get.return_value = mock_response
+        specrunner = SpecRunner(self.testspec)
+        specrunner.running = True  # run synchronously for testing
+        specrunner.run()
+        mock_requests_get.assert_called_with("https://example.com/factory/123/foo")
+        mock_popen.assert_called()
+
+
+if __name__ == '__main__':
+    unittest.main()

test-spec.yml

@@ -1,6 +1,8 @@
 sequence:
   - tests:
       - name: block devices
+        context:
+            url: https://conductor.infra.foundries.io/api/context/{LMP_FACTORY}/{CUSTOM_VERSION}/{LMP_MACHINE}
         command:
           - /usr/bin/lsblk
         on_host: true