auto/terraform: typing hints for base terraform module

app/terraform/terraform.py

@@ -1,5 +1,6 @@
 import json
 import subprocess
+from abc import abstractmethod
 from typing import Any, Dict, List, Optional, Tuple
 
 import jinja2
@@ -18,7 +19,23 @@ class TerraformAutomation(BaseAutomation):
     Default parallelism for remote API calls.
     """
 
-    def automate(self, full: bool = False):
+    def automate(self, full: bool = False) -> Tuple[bool, str]:
+        """
+        Runs the Terraform automation module. The run will follow these steps:
+
+        1. The :func:`tf_prehook` hook is run.
+        2. Generate a Terraform configuration and write it to a single ``main.tf`` file in the working directory
+           (see :func:`working_directory <app.terraform.BaseAutomation.working_directory>`).
+        3. Run ``terraform init``.
+        4. Run ``terraform apply``. This will only include a refresh if *full* is **True**. The apply will wait up
+           to *lock_timeout* minutes for a lock to be released before failing. Up to *parallelism* requests will be
+           sent to remote APIs concurrently.
+        5. The :func:`tf_posthook` hook is run.
+        6. The logs from the apply step are returned as a string.
+
+        :param full: include a Terraform refresh in the automation module run
+        :return: success status and Terraform apply logs
+        """
         prehook_result = self.tf_prehook()
         self.tf_generate()
         self.tf_init()
@@ -45,11 +62,12 @@ class TerraformAutomation(BaseAutomation):
             stdout=subprocess.PIPE)
         return tf.returncode, tf.stdout.decode('utf-8')
 
-    def tf_generate(self):
+    @abstractmethod
+    def tf_generate(self) -> None:
         raise NotImplementedError()
 
     def tf_init(self, *,
-                lock_timeout: int = 15):
+                lock_timeout: int = 15) -> None:
         # The init command does not support JSON output
         subprocess.run(
             ['terraform',
@@ -58,7 +76,7 @@ class TerraformAutomation(BaseAutomation):
              ],
             cwd=self.working_directory())
 
-    def tf_output(self) -> Dict[str, Any]:
+    def tf_output(self) -> Any:
         tf = subprocess.run(
             ['terraform', 'output', '-json'],
             cwd=self.working_directory(),
@@ -68,7 +86,7 @@ class TerraformAutomation(BaseAutomation):
     def tf_plan(self, *,
                 refresh: bool = True,
                 parallelism: Optional[int] = None,
-                lock_timeout: int = 15):
+                lock_timeout: int = 15) -> Tuple[int, str]:
         tf = subprocess.run(
             ['terraform',
              'plan',
@@ -78,11 +96,7 @@ class TerraformAutomation(BaseAutomation):
              f'-lock-timeout={str(lock_timeout)}m',
              ],
             cwd=self.working_directory())
-        logs = []
-        for line in tf.stdout.decode('utf-8').split('\n'):
-            if line.strip():
-                logs.append(json.loads(line))
-        return tf.returncode, str(logs)
+        return tf.returncode, tf.stdout.decode('utf-8')
 
     def tf_posthook(self, *, prehook_result: Any = None) -> None:
         """
@@ -108,14 +122,14 @@ class TerraformAutomation(BaseAutomation):
         """
         pass
 
-    def tf_show(self) -> Dict[str, Any]:
+    def tf_show(self) -> Any:
         terraform = subprocess.run(
             ['terraform', 'show', '-json'],
             cwd=self.working_directory(),
             stdout=subprocess.PIPE)
         return json.loads(terraform.stdout)
 
-    def tf_write(self, template: str, **kwargs):
+    def tf_write(self, template: str, **kwargs: Any) -> None:
         tmpl = jinja2.Template(template)
         with open(self.working_directory("main.tf"), 'w') as tf:
             tf.write(tmpl.render(**kwargs))