add ability to clear/update the environment for commands

Author: mathiasertl

README.md

@@ -73,13 +73,13 @@ Configure the tutorial that is being displayed - this will not show any output:
 
 ## TODO
 
-* Test file existance or something like that
-* Pre/post text for each part
 * Running commands:
 
   * Clear env
   * Add env (value is template)
 
+* Pre/post text for each part
+* Test file existence or something like that
 * Platform independent "echo" step (useful for debugging/testing)
 * Run in vagrant
 

structured_tutorials/models/base.py

@@ -43,6 +43,10 @@ class CommandBaseModel(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
     status_code: Annotated[int, Field(ge=0, le=255)] = 0
+    clear_environment: bool = Field(default=False, description="Clear the environment.")
+    environment: dict[str, Any] = Field(
+        default_factory=dict, description="Additional environment variables for the process."
+    )
     show_output: bool = Field(
         default=True, description="Set to `False` to always hide the output of this command."
     )

structured_tutorials/runners/base.py

@@ -6,6 +6,7 @@
 import abc
 import io
 import logging
+import os
 import shlex
 import subprocess
 import sys
@@ -75,12 +76,12 @@ def test_output(self, proc: subprocess.CompletedProcess[bytes], test: TestOutput
             try:
                 check_count(value.splitlines(), test.line_count)
             except ValueError as ex:
-                raise CommandOutputTestError(f"Line count error: {ex}")
+                raise CommandOutputTestError(f"Line count error: {ex}") from ex
         if test.character_count:
             try:
                 check_count(value, test.character_count)
             except ValueError as ex:
-                raise CommandOutputTestError(f"Character count error: {ex}")
+                raise CommandOutputTestError(f"Character count error: {ex}") from ex
 
     def validate_alternatives(self) -> None:
         """Validate that for each alternative part, an alternative was selected."""
@@ -104,11 +105,14 @@ def run_shell_command(
         capture_output: bool = False,
         stdin: int | io.BufferedReader | None = None,
         input: bytes | None = None,
+        environment: dict[str, Any] | None = None,
+        clear_environment: bool = False,
     ) -> CompletedProcess[bytes]:
         # Only show output if runner itself is not configured to hide all output
         if show_output:
             show_output = self.show_command_output
 
+        # Configure stdout/stderr streams
         if capture_output:
             stdout: int | None = subprocess.PIPE
             stderr: int | None = subprocess.PIPE
@@ -117,6 +121,20 @@ def run_shell_command(
         else:
             stdout = stderr = subprocess.DEVNULL
 
+        # Configure environment
+        if environment:
+            # If environment is passed, we render variables as templates
+            environment = {k: self.render(v) for k, v in environment.items()}
+        elif environment is None:  # pragma: no cover  # dict is always passed directly
+            # just to simplify the next branch -> environment is always a dict
+            environment = {}
+        if clear_environment:
+            env = environment
+        elif environment:
+            env = {**os.environ, **environment}
+        else:
+            env = None
+
         # Render the command (args) as template
         command = self.render_command(command)
 
@@ -127,7 +145,9 @@ def run_shell_command(
             logged_command = shlex.join(logged_command)
 
         command_logger.info(logged_command)
-        proc = subprocess.run(command, shell=shell, stdin=stdin, input=input, stdout=stdout, stderr=stderr)
+        proc = subprocess.run(
+            command, shell=shell, stdin=stdin, input=input, stdout=stdout, stderr=stderr, env=env
+        )
 
         # If we want to show the output and capture it at the same time, we need can only show the streams
         # separately at the end.

structured_tutorials/runners/local.py

@@ -13,7 +13,7 @@
 import time
 from pathlib import Path
 
-from structured_tutorials.errors import CommandOutputTestError, CommandTestError, PromptNotConfirmedError
+from structured_tutorials.errors import CommandTestError, PromptNotConfirmedError
 from structured_tutorials.models.parts import AlternativeModel, CommandsPartModel, FilePartModel, PromptModel
 from structured_tutorials.models.tests import TestCommandModel, TestOutputModel, TestPortModel
 from structured_tutorials.runners.base import RunnerBase
@@ -45,7 +45,12 @@ def run_test(
             tries += 1
 
             if isinstance(test, TestCommandModel):
-                test_proc = self.run_shell_command(test.command, show_output=test.show_output)
+                test_proc = self.run_shell_command(
+                    test.command,
+                    show_output=test.show_output,
+                    environment=test.environment,
+                    clear_environment=test.clear_environment,
+                )
 
                 if test.status_code == test_proc.returncode:
                     return
@@ -96,13 +101,17 @@ def run_commands(self, part: CommandsPartModel) -> None:
                         show_output=command_config.run.show_output,
                         capture_output=capture_output,
                         stdin=stdin,
+                        environment=command_config.run.environment,
+                        clear_environment=command_config.run.clear_environment,
                     )
             else:
                 proc = self.run_shell_command(
                     command_config.command,
                     show_output=command_config.run.show_output,
                     capture_output=capture_output,
                     input=proc_input,
+                    environment=command_config.run.environment,
+                    clear_environment=command_config.run.clear_environment,
                 )
 
             # Update list of cleanup commands

structured_tutorials/utils.py

@@ -76,7 +76,12 @@ def cleanup(runner: "RunnerBase") -> Iterator[None]:
             log.info("Running cleanup commands.")
 
         for command_config in runner.cleanup:
-            runner.run_shell_command(command_config.command, command_config.show_output)
+            runner.run_shell_command(
+                command_config.command,
+                command_config.show_output,
+                environment=command_config.environment,
+                clear_environment=command_config.clear_environment,
+            )
 
 
 def git_export(destination: str | Path, ref: str = "HEAD") -> Path:

tests/data/tutorials/command-simple-env.yaml

@@ -0,0 +1,11 @@
+parts:
+  - commands:
+      - command: echo 1 $VARIABLE
+        run:
+          environment:
+            VARIABLE: VALUE 1
+      - command: echo 2 $VARIABLE
+        run:
+          clear_environment: true
+          environment:
+            VARIABLE: VALUE 2

tests/runners/local/test_command.py

@@ -64,7 +64,7 @@ def test_command_as_list(fp: FakeProcess, runner: LocalTutorialRunner) -> None:
     recorder_test = fp.register(["ls", "test with spaces"])
     recorder_cleanup = fp.register(["ls", "cleanup with spaces"])
     runner.run()
-    expected = {"shell": False, "stdin": None, "stderr": None, "stdout": None}
+    expected = {"shell": False, "stdin": None, "stderr": None, "stdout": None, "env": None}
     assert recorder_main.calls[0].kwargs == expected
     assert recorder_test.calls[0].kwargs == expected
     assert recorder_cleanup.calls[0].kwargs == expected
@@ -103,7 +103,13 @@ def test_command_hide_output(fp: FakeProcess, runner: LocalTutorialRunner) -> No
     recorder_test = fp.register("ls test")
     recorder_cleanup = fp.register("ls cleanup")
     runner.run()
-    expected = {"shell": True, "stdin": None, "stderr": subprocess.DEVNULL, "stdout": subprocess.DEVNULL}
+    expected = {
+        "shell": True,
+        "stdin": None,
+        "stderr": subprocess.DEVNULL,
+        "stdout": subprocess.DEVNULL,
+        "env": None,
+    }
     assert recorder_main.calls[0].kwargs == expected
     assert recorder_test.calls[0].kwargs == expected
     assert recorder_cleanup.calls[0].kwargs == expected
@@ -116,7 +122,13 @@ def test_command_capture_output(
     """Test running a commands with capturing the output."""
     recorder = fp.register("echo foo bar bla", stdout="foo bar bla", stderr="foo bla bla")
     runner.run()
-    expected = {"shell": True, "stdin": None, "stderr": subprocess.PIPE, "stdout": subprocess.PIPE}
+    expected = {
+        "shell": True,
+        "stdin": None,
+        "stderr": subprocess.PIPE,
+        "stdout": subprocess.PIPE,
+        "env": None,
+    }
     assert recorder.calls[0].kwargs == expected
     output = capsys.readouterr()
     assert output.out == "--- stdout ---\nfoo bar bla\n--- stderr ---\nfoo bla bla\n"
@@ -156,6 +168,22 @@ def test_command_with_invalid_output(fp: FakeProcess, runner: LocalTutorialRunne
         runner.run()
 
 
+@pytest.mark.tutorial("command-simple-env")
+def test_environment(fp: FakeProcess, runner: LocalTutorialRunner) -> None:
+    """Test running a commands with environment variables."""
+    recorder1 = fp.register("echo 1 $VARIABLE")
+    recorder2 = fp.register("echo 2 $VARIABLE")
+    runner.run()
+    assert recorder1.calls[0].kwargs["env"]["VARIABLE"] == "VALUE 1"
+    assert recorder2.calls[0].kwargs == {
+        "shell": True,
+        "stdin": None,
+        "stderr": None,
+        "stdout": None,
+        "env": {"VARIABLE": "VALUE 2"},
+    }
+
+
 @pytest.mark.doc_tutorial("test-command")
 def test_test_commands_with_command_error(fp: FakeProcess, doc_runner: LocalTutorialRunner) -> None:
     """Test the cleanup from docs."""