Add output task with stdout capture.

Author: tobyclemson

src/infrablocks/invoke_terraform/task_factory.py

@@ -16,24 +16,35 @@ class InitConfiguration:
     reconfigure: bool
 
 
+@dataclass
+class OutputConfiguration:
+    json: bool
+
+
 @dataclass
 class Configuration:
+    init_configuration: InitConfiguration
+    output_configuration: OutputConfiguration
+
     source_directory: str
     variables: tf.Variables
     workspace: str | None
-    init_configuration: InitConfiguration
-    environment: Environment | None = None
     auto_approve: bool = True
 
+    capture_stdout: bool = False
+    environment: Environment | None = None
+
     @staticmethod
     def create_empty():
         return Configuration(
-            source_directory="",
-            variables={},
-            workspace=None,
             init_configuration=InitConfiguration(
                 backend_config={}, reconfigure=False
             ),
+            output_configuration=OutputConfiguration(json=False),
+            source_directory="",
+            variables={},
+            workspace=None,
+            capture_stdout=False,
             environment={},
         )
 
@@ -62,10 +73,15 @@ def create(
         apply_task = invoke_factory.create_task(
             self._create_apply(pre_task_function), task_parameters
         )
+        output_task = invoke_factory.create_task(
+            self._create_output(pre_task_function), task_parameters
+        )
 
         # TODO: investigate type issue
         collection.add_task(plan_task)  # pyright: ignore[reportUnknownMemberType]
         collection.add_task(apply_task)  # pyright: ignore[reportUnknownMemberType]
+        collection.add_task(output_task)  # pyright: ignore[reportUnknownMemberType]
+
         return collection
 
     def _create_plan(
@@ -101,6 +117,35 @@ def apply(context: Context, arguments: invoke_factory.Arguments):
 
         return apply
 
+    def _create_output(
+        self, pre_task_function: PreTaskFunction
+    ) -> invoke_factory.BodyCallable[str | None]:
+        def output(
+            context: Context, arguments: invoke_factory.Arguments
+        ) -> str | None:
+            (terraform, configuration) = self._pre_command_setup(
+                pre_task_function, context, arguments
+            )
+
+            capture: tf.StreamNames | None = None
+            if configuration.capture_stdout:
+                capture = {"stdout"}
+
+            result = terraform.output(
+                chdir=configuration.source_directory,
+                capture=capture,
+                json=configuration.output_configuration.json,
+                environment=configuration.environment,
+            )
+
+            if configuration.capture_stdout and result.stdout is not None:
+                output = result.stdout.read()
+                return output.strip()
+
+            return None
+
+        return output
+
     def _pre_command_setup(
         self,
         pre_task_function: PreTaskFunction,

src/infrablocks/invoke_terraform/terraform/__init__.py

@@ -1,13 +1,17 @@
 from .terraform import BackendConfig, ConfigurationValue, Variables
 from .terraform import Environment as Environment
 from .terraform import Executor as Executor
+from .terraform import Result as Result
+from .terraform import StreamName as StreamName
+from .terraform import StreamNames as StreamNames
 from .terraform import Terraform as Terraform
 
 __all__ = [
     "BackendConfig",
     "ConfigurationValue",
     "Environment",
     "Executor",
+    "Result",
     "Terraform",
     "Variables",
 ]

src/infrablocks/invoke_terraform/terraform/terraform.py

@@ -20,7 +20,9 @@
 
 
 class Result:
-    def __init__(self, stdout: IO[str] | None, stderr: IO[str] | None):
+    def __init__(
+        self, stdout: IO[str] | None = None, stderr: IO[str] | None = None
+    ):
         self.stdout = stdout
         self.stderr = stderr
 

tests/unit/infrablocks/invoke_terraform/test_task_factory.py

@@ -1,3 +1,4 @@
+from io import StringIO
 from typing import cast
 from unittest.mock import Mock
 
@@ -261,3 +262,150 @@ def pre_task_function(_context, _, configuration: Configuration):
             autoapprove=True,
             environment=environment,
         )
+
+    def test_creates_output_task(self):
+        pre_task_function_mock = Mock()
+
+        collection = TaskFactory().create(
+            "collection", [], pre_task_function_mock
+        )
+
+        assert collection.tasks["output"] is not None
+
+    def test_output_invokes_init_and_output(self):
+        terraform = Mock(spec=tf.Terraform)
+        task_factory = TaskFactory(
+            terraform_factory=TerraformFactory(terraform)
+        )
+        source_directory = "/some/path"
+        backend_config: tf.BackendConfig = {"path": "state_file.tfstate"}
+
+        def pre_task_function(_context, _, configuration: Configuration):
+            configuration.source_directory = source_directory
+            configuration.init_configuration.backend_config = backend_config
+
+        collection = task_factory.create("collection", [], pre_task_function)
+        output: Task = cast(Task, collection.tasks["output"])
+
+        output(Context())
+
+        terraform.init.assert_called_once_with(
+            chdir=source_directory,
+            backend_config=backend_config,
+            reconfigure=False,
+            environment={},
+        )
+        terraform.output.assert_called_once_with(
+            chdir=source_directory,
+            capture=None,
+            json=False,
+            environment={},
+        )
+
+    def test_output_uses_workspace(self):
+        terraform = Mock(spec=tf.Terraform)
+        task_factory = TaskFactory(
+            terraform_factory=TerraformFactory(terraform)
+        )
+        workspace = "workspace"
+        source_directory = "/some/path"
+
+        def pre_task_function(_context, _, configuration: Configuration):
+            configuration.source_directory = source_directory
+            configuration.workspace = workspace
+
+        collection = task_factory.create("collection", [], pre_task_function)
+        output: Task = cast(Task, collection.tasks["output"])
+
+        output(Context())
+
+        terraform.select_workspace.assert_called_once_with(
+            workspace, chdir=source_directory, or_create=True, environment={}
+        )
+
+    def test_output_uses_json(self):
+        terraform = Mock(spec=tf.Terraform)
+        task_factory = TaskFactory(
+            terraform_factory=TerraformFactory(terraform)
+        )
+        workspace = "workspace"
+        source_directory = "/some/path"
+
+        def pre_task_function(_context, _, configuration: Configuration):
+            configuration.source_directory = source_directory
+            configuration.workspace = workspace
+            configuration.output_configuration.json = True
+
+        collection = task_factory.create("collection", [], pre_task_function)
+        output: Task = cast(Task, collection.tasks["output"])
+
+        output(Context())
+
+        terraform.output.assert_called_once_with(
+            chdir=source_directory,
+            capture=None,
+            json=True,
+            environment={},
+        )
+
+    def test_output_uses_environment_in_all_commands_when_set(self):
+        terraform = Mock(spec=tf.Terraform)
+        task_factory = TaskFactory(
+            terraform_factory=TerraformFactory(terraform)
+        )
+        source_directory = "/some/path"
+        environment = {"ENV_VAR": "value"}
+        workspace = "workspace"
+
+        def pre_task_function(_context, _, configuration: Configuration):
+            configuration.source_directory = source_directory
+            configuration.environment = environment
+            configuration.workspace = workspace
+
+        collection = task_factory.create("collection", [], pre_task_function)
+        output: Task = cast(Task, collection.tasks["output"])
+
+        output(Context())
+
+        terraform.init.assert_called_once_with(
+            chdir=source_directory,
+            backend_config={},
+            reconfigure=False,
+            environment=environment,
+        )
+
+        terraform.select_workspace.assert_called_once_with(
+            "workspace",
+            chdir=source_directory,
+            or_create=True,
+            environment=environment,
+        )
+
+        terraform.output.assert_called_once_with(
+            chdir=source_directory,
+            capture=None,
+            json=False,
+            environment=environment,
+        )
+
+    def test_output_returns_standard_output_when_capture_stdout_true(self):
+        terraform = Mock(spec=tf.Terraform)
+        task_factory = TaskFactory(
+            terraform_factory=TerraformFactory(terraform)
+        )
+        source_directory = "/some/path"
+
+        terraform.output.return_value = tf.Result(
+            stdout=StringIO("output_value\n"), stderr=None
+        )
+
+        def pre_task_function(_context, _, configuration: Configuration):
+            configuration.source_directory = source_directory
+            configuration.capture_stdout = True
+
+        collection = task_factory.create("collection", [], pre_task_function)
+        output: Task = cast(Task, collection.tasks["output"])
+
+        output_value = output(Context())
+
+        assert output_value == "output_value"