Inject env variables inside SUT via option

Added --env option that permits to inject environment variables inside
SUT before any execution.

Signed-off-by: Andrea Cervesato <andrea.cervesato@suse.com>

Author: acerv

ltp/main.py

@@ -90,6 +90,20 @@ def _sut_config(value: str) -> dict:
     return config
 
 
+def _env_config(value: str) -> dict:
+    """
+    Return an environment configuration dictionary, parsing strings such as
+    "key=value:key=value:key=value".
+    """
+    if not value:
+        return None
+
+    params = value.split(':')
+    config = _from_params_to_config(params)
+
+    return config
+
+
 def _discover_sut(folder: str) -> list:
     """
     Discover new SUT implementations inside a specific folder.
@@ -185,7 +199,8 @@ def _ltp_run(parser: ArgumentParser, args: Namespace) -> None:
         args.run_cmd,
         args.ltp_dir,
         tmpdir,
-        skip_tests=skip_tests)
+        skip_tests=skip_tests,
+        env=args.env)
 
     ltp.events.stop_event_loop()
 
@@ -243,6 +258,11 @@ def run() -> None:
         type=int,
         default=3600,
         help="Timeout before stopping a single execution")
+    parser.add_argument(
+        "--env",
+        "-e",
+        type=_env_config,
+        help="List of key=value environment values separated by ':'")
     parser.add_argument(
         "--run-suite",
         "-r",

ltp/session.py

@@ -147,27 +147,35 @@ def _start_sut(
             self,
             ltpdir: str,
             tmpdir: TempDir,
-            sut_config: dict) -> SUT:
+            sut_config: dict,
+            env: dict) -> SUT:
         """
         Start a new SUT and return it initialized.
         """
         sut_name = sut_config.get("name", None)
         testcases = os.path.join(ltpdir, "testcases", "bin")
 
-        env = {}
-        env["PATH"] = "/sbin:/usr/sbin:/usr/local/sbin:" + \
+        sut_env = {}
+        sut_env["PATH"] = "/sbin:/usr/sbin:/usr/local/sbin:" + \
             f"/root/bin:/usr/local/bin:/usr/bin:/bin:{testcases}"
-        env["LTPROOT"] = ltpdir
-        env["TMPDIR"] = tmpdir.root if tmpdir.root else "/tmp"
-        env["LTP_TIMEOUT_MUL"] = str((self._exec_timeout * 0.9) / 300.0)
+        sut_env["LTPROOT"] = ltpdir
+        sut_env["TMPDIR"] = tmpdir.root if tmpdir.root else "/tmp"
+        sut_env["LTP_TIMEOUT_MUL"] = str((self._exec_timeout * 0.9) / 300.0)
 
         if self._no_colors:
-            env["LTP_COLORIZE_OUTPUT"] = "0"
+            sut_env["LTP_COLORIZE_OUTPUT"] = "0"
         else:
-            env["LTP_COLORIZE_OUTPUT"] = "1"
+            sut_env["LTP_COLORIZE_OUTPUT"] = "1"
+
+        if env:
+            for key, value in env.items():
+                if key not in sut_env:
+                    self._logger.info(
+                        "Add %s=%s environment variable into SUT")
+                    sut_env[key] = value
 
         config = {}
-        config['env'] = env
+        config['env'] = sut_env
         config['cwd'] = testcases
         config['tmpdir'] = tmpdir.abspath
         config.update(sut_config)
@@ -257,7 +265,8 @@ def run_single(
             command: str,
             ltpdir: str,
             tmpdir: TempDir,
-            skip_tests: str = None) -> int:
+            skip_tests: str = None,
+            env: dict = None) -> int:
         """
         Run some testing suites with a specific SUT configurations.
         :param sut_config: system under test configuration.
@@ -274,6 +283,8 @@ def run_single(
         :type tmpdir: TempDir
         :param skip_tests: tests to ignore in a regex form
         :type skip_tests: str
+        :param env: environment variables to inject inside SUT
+        :type env: dict
         :returns: exit code for the session
         """
         if not sut_config:
@@ -296,7 +307,8 @@ def run_single(
                 self._sut = self._start_sut(
                     ltpdir,
                     tmpdir,
-                    sut_config)
+                    sut_config,
+                    env)
 
                 self._logger.info("Created SUT: %s", self._sut.name)
 

ltp/tests/test_session.py

@@ -1,9 +1,9 @@
 """
 Unittests for session module.
 """
-import json
 import os
 import stat
+import json
 import queue
 import pytest
 import ltp
@@ -353,3 +353,43 @@ def test_suite_timeout_report(self, suts, tmpdir, sut_config, ltpdir):
         assert tracer.next_event() == "suite_timeout"
         assert tracer.next_event() == "sut_stop"
         assert tracer.next_event() == "session_completed"
+
+    def test_env(self, suts, tmpdir, sut_config, ltpdir):
+        """
+        Run a session without suites but only one command run.
+        """
+        report_path = tmpdir / "report.json"
+
+        ltpdir = tmpdir.mkdir("ltp")
+        script_sh = ltpdir.mkdir("testcases").mkdir("bin") / "script.sh"
+        script_sh.write("#!/bin/sh\necho -n $VAR0:$VAR1")
+
+        st = os.stat(str(script_sh))
+        os.chmod(str(script_sh), st.st_mode | stat.S_IEXEC)
+
+        suite = ltpdir.mkdir("runtest") / "suite"
+        suite.write("test script.sh")
+
+        tmpdir_obj = TempDir(root=tmpdir)
+        try:
+            session = Session(suts)
+            retcode = session.run_single(
+                sut_config,
+                report_path,
+                ["suite"],
+                None,
+                ltpdir,
+                tmpdir_obj,
+                env=dict(VAR0="0", VAR1="1"))
+
+            assert retcode == Session.RC_OK
+            assert os.path.isfile(report_path)
+
+            report_d = None
+            with open(report_path, 'r') as report_f:
+                report_d = json.loads(report_f.read())
+
+            assert len(report_d["results"]) > 0
+            assert report_d["results"][0]["test"]["log"] == "0:1"
+        finally:
+            session.stop()