Add support for Terraform plan JSON output (#3)

Add support for Terraform plan JSON output

tests/test_tftest.py

@@ -53,6 +53,7 @@
 
 
 def test_parse_args():
+  "Test parsing Terraform command arguments."
   assert tftest.parse_args() == []
   for kwargs, expected in _ARGS_TESTS:
     assert tftest.parse_args(**kwargs) == expected
@@ -63,13 +64,15 @@ def test_parse_args():
 
 
 def test_json_output_class():
-  out = tftest.TerraformOutputs(
+  "Test the output and variables wrapper class."
+  out = tftest.TerraformValueDict(
       {'a': {'value': 1}, 'b': {'value': 2, 'sensitive': True}})
   assert out.sensitive == ('b',)
   assert (out['a'], out['b']) == (1, 2)
 
 
 def test_json_state_class():
+  "Test the state wrapper class."
   s = tftest.TerraformState({
       'version': 'foo',
       'modules': [
@@ -95,7 +98,71 @@ def test_json_state_class():
   })
   assert sorted(list(s.modules.keys())) == ['a', 'b']
   assert type(s.modules['a']) == tftest.TerraformStateModule
-  assert type(s.modules['a'].outputs) == tftest.TerraformOutputs
+  assert type(s.modules['a'].outputs) == tftest.TerraformValueDict
+
+
+def test_plan_out_class():
+  "Test the plan JSON output wrapper class."
+  s = {
+      "format_version": "0.1",
+      "terraform_version": "0.12.6",
+      "variables": {
+          "foo": {
+              "value": "bar"
+          }
+      },
+      "planned_values": {
+          "outputs": {
+              "spam": {
+                  "sensitive": False,
+                  "value": "baz"
+              },
+          },
+          "root_module": {
+              "child_modules": [
+                  {
+                      "resources": [],
+                      "address": "module.eggs"
+                  }
+              ]
+          }
+      },
+      "resource_changes": [
+          {
+              "address": "module.spam.resource_type.resource.name",
+              "module_address": "module.spam",
+              "mode": "managed"
+          }
+      ],
+      "output_changes": {
+          "spam": {
+              "actions": [
+                  "create"
+              ],
+              "before": None,
+              "after": "bar",
+              "after_unknown": False
+          },
+      },
+      "prior_state": {
+          "format_version": "0.1",
+          "terraform_version": "0.12.6"
+      },
+      "configuration": {
+          "provider_config": {
+              "google": {
+                  "name": "google"
+              }
+          },
+      }
+  }
+  plan_out = tftest.TerraformPlanOutput(s)
+  assert plan_out.terraform_version == "0.12.6", plan_out.terraform_version
+  assert plan_out.variables['foo'] == 'bar'
+  assert plan_out.outputs['spam'] == 'baz'
+  assert plan_out.modules['module.eggs'] == []
+  assert plan_out.resource_changes['module.spam.resource_type.resource.name'] == s['resource_changes'][0]
+  assert plan_out.configuration == s['configuration']
 
 
 def test_setup_files():

tftest.py

@@ -30,9 +30,10 @@
 import os
 import shutil
 import subprocess
+import tempfile
 import weakref
 
-__version__ = '0.5.0'
+__version__ = '0.6.0'
 
 _LOGGER = logging.getLogger('tftest')
 
@@ -107,26 +108,54 @@ def __str__(self):
     return str(self.raw)
 
 
-class TerraformOutputs(TerraformJSONBase):
-  """Minimal wrapper to directly expose output values."""
+class TerraformValueDict(TerraformJSONBase):
+  """Minimal wrapper to directly expose outputs or variables."""
 
   def __init__(self, raw):
-    super(TerraformOutputs, self).__init__(raw)
+    super(TerraformValueDict, self).__init__(raw)
+    # only matters for outputs
     self.sensitive = tuple(k for k, v in raw.items() if v.get('sensitive'))
 
   def __getitem__(self, name):
     return self.raw[name]['value']
 
 
+class TerraformPlanOutput(object):
+  """Minimal wrapper for Terraform plan JSON output."""
+
+  def __init__(self, raw):
+    self._raw = raw
+    self.variables = TerraformValueDict(raw['variables'])
+    self.outputs = TerraformValueDict(raw['planned_values']['outputs'])
+    self._modules = self._resources = None
+
+  @property
+  def modules(self):
+    if self._modules is None:
+      modules = self._raw['planned_values']['root_module']['child_modules']
+      self._modules = dict((mod['address'], mod['resources'])
+                           for mod in modules)
+    return self._modules
+
+  @property
+  def resource_changes(self):
+    if self._resources is None:
+      self._resources = dict((v['address'], v)
+                             for v in self._raw['resource_changes'])
+    return self._resources
+
+  def __getattr__(self, name):
+    return self._raw[name]
+
+
 class TerraformStateModule(object):
   """Minimal wrapper for Terraform state modules."""
 
   def __init__(self, path, raw):
     self._raw = raw
     self.path = path
-    self.outputs = TerraformOutputs(raw['outputs'])
+    self.outputs = TerraformValueDict(raw['outputs'])
     self.depends_on = raw['depends_on']
-    # key type provider attributes depends_on
     self.resources = {}
     for k, v in raw['resources'].items():
       self.resources[k] = TerraformStateResource(
@@ -300,6 +329,20 @@ def plan(self, input=False, color=False, refresh=True, tf_vars=None):
                           refresh=refresh, tf_vars=tf_vars)
     return self.execute_command('plan', *cmd_args).out
 
+  def plan_out(self, input=False, color=False, refresh=True, tf_vars=None, parsed=True):
+    """Run Terraform plan command and return saved output as JSON."""
+    cmd_args = parse_args(input=input, color=color,
+                          refresh=refresh, tf_vars=tf_vars)
+    with tempfile.NamedTemporaryFile() as fp:
+      cmd_args.append('-out={}'.format(fp.name))
+      self.execute_command('plan', *cmd_args)
+      result = self.execute_command('show', '-no-color', '-json', fp.name)
+    try:
+      plan_out = json.loads(result.out)
+    except json.JSONDecodeError as e:
+      _LOGGER.warning('error decoding plan output: {}'.format(e))
+    return plan_out if not parsed else TerraformPlanOutput(plan_out)
+
   def apply(self, input=False, color=False, auto_approve=True, tf_vars=None):
     """Run Terraform apply command."""
     cmd_args = parse_args(input=input, color=color,
@@ -316,7 +359,7 @@ def output(self, name=None, color=False, json_format=True):
     _LOGGER.debug('output %s', output)
     if json_format:
       try:
-        output = TerraformOutputs(json.loads(output))
+        output = TerraformValueDict(json.loads(output))
       except json.JSONDecodeError as e:
         _LOGGER.warning('error decoding output: {}'.format(e))
     return output