Tidy up typing, use DI rather than protected attribute access.

Author: tobyclemson

src/infrablocks/invoke_terraform/invoke_executor.py

@@ -1,4 +1,4 @@
-from typing import Iterable, Optional
+from typing import Iterable
 
 from invoke.context import Context
 
@@ -11,6 +11,6 @@ def __init__(self, context: Context):
         self._context = context
 
     def execute(
-        self, command: Iterable[str], env: Optional[Environment] = None
+        self, command: Iterable[str], env: Environment | None = None
     ) -> None:
         self._context.run(" ".join(command), env=(env or {}))

src/infrablocks/invoke_terraform/task_factory.py

@@ -1,5 +1,5 @@
+from collections.abc import Callable
 from dataclasses import dataclass
-from typing import Callable, Optional
 
 from invoke.collection import Collection
 from invoke.context import Context
@@ -20,9 +20,9 @@ class InitConfiguration:
 class Configuration:
     source_directory: str
     variables: tf.Variables
-    workspace: Optional[str]
+    workspace: str | None
     init_configuration: InitConfiguration
-    environment: Optional[Environment] = None
+    environment: Environment | None = None
     auto_approve: bool = True
 
     @staticmethod
@@ -44,8 +44,10 @@ def create_empty():
 
 
 class TaskFactory:
-    def __init__(self):
-        self._terraformFactory = TerraformFactory()
+    def __init__(
+        self, terraform_factory: TerraformFactory = TerraformFactory()
+    ):
+        self._terraform_factory = terraform_factory
 
     def create(
         self,
@@ -111,7 +113,7 @@ def _pre_command_setup(
             arguments,
             configuration,
         )
-        terraform = self._terraformFactory.build(context)
+        terraform = self._terraform_factory.build(context)
         terraform.init(
             chdir=configuration.source_directory,
             backend_config=configuration.init_configuration.backend_config,

src/infrablocks/invoke_terraform/terraform/__init__.py

@@ -2,4 +2,10 @@
 from .terraform import Executor as Executor
 from .terraform import Terraform as Terraform
 
-__all__ = ["ConfigurationValue", "Variables", "BackendConfig"]
+__all__ = [
+    "BackendConfig",
+    "ConfigurationValue",
+    "Executor",
+    "Terraform",
+    "Variables",
+]

src/infrablocks/invoke_terraform/terraform/terraform.py

@@ -1,15 +1,13 @@
-from typing import Dict, Iterable, List, Optional, Union
+from collections.abc import Mapping, Sequence
 
-type ConfigurationValue = Union[bool, int, float, str, None]
-type Variables = Dict[str, ConfigurationValue]
-type BackendConfig = Union[str, Dict[str, ConfigurationValue]]
-type Environment = Dict[str, str]
+type ConfigurationValue = bool | int | float | str | None
+type Variables = Mapping[str, ConfigurationValue]
+type BackendConfig = str | Mapping[str, ConfigurationValue]
+type Environment = Mapping[str, str]
 
 
 class Executor:
-    def execute(
-        self, command: Iterable[str], env: Optional[Environment]
-    ) -> None:
+    def execute(self, command: Sequence[str], env: Environment | None) -> None:
         raise Exception("NotImplementedException")
 
 
@@ -19,10 +17,10 @@ def __init__(self, executor: Executor):
 
     def init(
         self,
-        chdir: Optional[str] = None,
-        backend_config: Optional[BackendConfig] = {},
-        reconfigure: Optional[bool] = False,
-        environment: Optional[Environment] = None,
+        chdir: str | None = None,
+        backend_config: BackendConfig | None = None,
+        reconfigure: bool = False,
+        environment: Environment | None = None,
     ):
         base_command = self._build_base_command(chdir)
         command = (
@@ -38,9 +36,9 @@ def init(
 
     def plan(
         self,
-        chdir: Optional[str] = None,
-        vars: Optional[Variables] = {},
-        environment: Optional[Environment] = None,
+        chdir: str | None = None,
+        vars: Variables | None = None,
+        environment: Environment | None = None,
     ):
         base_command = self._build_base_command(chdir)
         command = base_command + ["plan"] + self._build_vars(vars)
@@ -49,10 +47,10 @@ def plan(
 
     def apply(
         self,
-        chdir: Optional[str] = None,
-        vars: Optional[Variables] = {},
+        chdir: str | None = None,
+        vars: Variables | None = None,
         autoapprove: bool = False,
-        environment: Optional[Environment] = None,
+        environment: Environment | None = None,
     ):
         base_command = self._build_base_command(chdir)
         autoapprove_flag = ["-auto-approve"] if autoapprove else []
@@ -68,9 +66,9 @@ def apply(
     def select_workspace(
         self,
         workspace: str,
-        chdir: Optional[str] = None,
+        chdir: str | None = None,
         or_create: bool = False,
-        environment: Optional[Environment] = None,
+        environment: Environment | None = None,
     ):
         base_command = self._build_base_command(chdir)
         command = base_command + ["workspace", "select"]
@@ -83,16 +81,16 @@ def select_workspace(
         self._executor.execute(command, env=environment)
 
     @staticmethod
-    def _build_base_command(chdir: Optional[str]) -> List[str]:
+    def _build_base_command(chdir: str | None) -> list[str]:
         command = ["terraform"]
 
         if chdir is not None:
             return command + [f"-chdir={chdir}"]
 
         return command
 
-    def _build_vars(self, variables: Optional[Variables]) -> List[str]:
-        if not variables:
+    def _build_vars(self, variables: Variables | None) -> list[str]:
+        if variables is None:
             return []
 
         return [
@@ -118,9 +116,9 @@ def _format_configuration_value(
         )
 
     def _build_backend_config(
-        self, backend_config: Optional[BackendConfig]
-    ) -> List[str]:
-        if not backend_config:
+        self, backend_config: BackendConfig | None
+    ) -> list[str]:
+        if backend_config is None:
             return []
 
         if isinstance(backend_config, str):

tests/unit/infrablocks/invoke_terraform/test_task_factory.py

@@ -35,8 +35,9 @@ def test_creates_plan_task(self):
 
     def test_plan_does_not_use_workspace_when_not_set(self):
         terraform = Mock(spec=tf.Terraform)
-        task_factory = TaskFactory()
-        task_factory._terraformFactory = TerraformFactory(terraform)
+        task_factory = TaskFactory(
+            terraform_factory=TerraformFactory(terraform)
+        )
 
         def pre_task_function(_context, _, configuration: Configuration):
             configuration.source_directory = "/some/path"
@@ -51,8 +52,9 @@ def pre_task_function(_context, _, configuration: Configuration):
 
     def test_plan_uses_workspace_when_set(self):
         terraform = Mock(spec=tf.Terraform)
-        task_factory = TaskFactory()
-        task_factory._terraformFactory = TerraformFactory(terraform)
+        task_factory = TaskFactory(
+            terraform_factory=TerraformFactory(terraform)
+        )
         workspace = "workspace"
         source_directory = "/some/path"
 
@@ -71,8 +73,9 @@ def pre_task_function(_context, _, configuration: Configuration):
 
     def test_plan_initialises_with_reconfigure(self):
         terraform = Mock(spec=tf.Terraform)
-        task_factory = TaskFactory()
-        task_factory._terraformFactory = TerraformFactory(terraform)
+        task_factory = TaskFactory(
+            terraform_factory=TerraformFactory(terraform)
+        )
         source_directory = "/some/path"
 
         def pre_task_function(_context, _, configuration: Configuration):
@@ -93,8 +96,9 @@ def pre_task_function(_context, _, configuration: Configuration):
 
     def test_plan_invokes_init_and_plan(self):
         terraform = Mock(spec=tf.Terraform)
-        task_factory = TaskFactory()
-        task_factory._terraformFactory = TerraformFactory(terraform)
+        task_factory = TaskFactory(
+            terraform_factory=TerraformFactory(terraform)
+        )
         source_directory = "/some/path"
         variables: tf.Variables = {"foo": 1}
         backend_config: tf.BackendConfig = {"path": "state_file.tfstate"}
@@ -121,8 +125,9 @@ def pre_task_function(_context, _, configuration: Configuration):
 
     def test_plan_uses_environment_in_all_commands_when_set(self):
         terraform = Mock(spec=tf.Terraform)
-        task_factory = TaskFactory()
-        task_factory._terraformFactory = TerraformFactory(terraform)
+        task_factory = TaskFactory(
+            terraform_factory=TerraformFactory(terraform)
+        )
         source_directory = "/some/path"
         environment = {"ENV_VAR": "value"}
         workspace = "workspace"
@@ -166,8 +171,9 @@ def test_creates_apply_task(self):
 
     def test_apply_invokes_init_and_apply(self):
         terraform = Mock(spec=tf.Terraform)
-        task_factory = TaskFactory()
-        task_factory._terraformFactory = TerraformFactory(terraform)
+        task_factory = TaskFactory(
+            terraform_factory=TerraformFactory(terraform)
+        )
         source_directory = "/some/path"
         variables: tf.Variables = {"foo": 1}
         backend_config: tf.BackendConfig = {"path": "state_file.tfstate"}
@@ -197,8 +203,9 @@ def pre_task_function(_context, _, configuration: Configuration):
 
     def test_apply_uses_workspace(self):
         terraform = Mock(spec=tf.Terraform)
-        task_factory = TaskFactory()
-        task_factory._terraformFactory = TerraformFactory(terraform)
+        task_factory = TaskFactory(
+            terraform_factory=TerraformFactory(terraform)
+        )
         workspace = "workspace"
         source_directory = "/some/path"
 
@@ -217,8 +224,9 @@ def pre_task_function(_context, _, configuration: Configuration):
 
     def test_apply_uses_environment_in_all_commands_when_set(self):
         terraform = Mock(spec=tf.Terraform)
-        task_factory = TaskFactory()
-        task_factory._terraformFactory = TerraformFactory(terraform)
+        task_factory = TaskFactory(
+            terraform_factory=TerraformFactory(terraform)
+        )
         source_directory = "/some/path"
         environment = {"ENV_VAR": "value"}
         workspace = "workspace"