Simplify path handling

tests/conftest.py

@@ -159,11 +159,43 @@ def basedir():
 
 
 @pytest.fixture
-def generic_plan_summary():
+def generic_plan_summary(request):
+  'Returns a function to generate a PlanSummary'
 
-  def inner(module_path, tf_var_files=None, basedir=None, **tf_vars):
-    basedir = Path(basedir or BASEDIR)
-    module_path = basedir / module_path
+  def inner(module_path, basedir=None, tf_var_files=None, **tf_vars):
+    '''Run a Terraform plan on the module located at `module_path`.\
+
+    - module_path: terraform root module to run. Can be an absolute
+      path or relative to the root of the repository
+
+    - basedir: directory root to use for relative paths in
+      tf_var_files. If None, then paths are relative to the calling
+      test function
+    
+    - tf_var_files: set of terraform variable files (tfvars) to pass
+      in to terraform
+
+    Returns a PlanSummary object containing 3 attributes:
+    - values: dictionary where the keys are terraform plan addresses
+      and values are the JSON representation (converted to python
+      types) of the attribute values of the resource. 
+
+    - counts: dictionary where the keys are the terraform resource
+      types and the values are the number of times that type appears
+      in the plan
+    
+    - outputs: dictionary of the modules outputs that can be
+      determined at plan type. 
+
+    Consult [1] for mode details on the structure of values and outputs
+
+    [1] https://developer.hashicorp.com/terraform/internals/json-format
+
+    '''
+
+    if basedir is None:
+      basedir = Path(request.fspath).parent
+    module_path = Path(BASEDIR) / module_path
 
     # FIXME: find a way to prevent the temp dir if TFTEST_COPY is not
     # in the environment
@@ -175,21 +207,32 @@ def generic_plan_summary():
         test_path = Path(tmp_path)
         shutil.copytree(module_path, test_path, dirs_exist_ok=True)
 
-        # if we're copying, we might as well remove any auto.tfvars
-        # files from the destintion directory to avoid surprises (e.g.
-        # if you have an active fast deployment with links to configs)
-        autofiles = itertools.chain(test_path.glob("*.auto.tfvars"),
-                                    test_path.glob("*.auto.tfvars.json"),
-                                    test_path.glob("terraform.tfvars"))
-        for f in autofiles:
-          f.unlink()
+        # if we're copying the module, we might as well remove any
+        # files and directories from the test directory that are
+        # automatically read by terraform. Useful to avoid surprises
+        # with (e.g. if you have an active fast deployment with links
+        # to configs)
+        autopaths = itertools.chain(
+            test_path.glob("*.auto.tfvars"),
+            test_path.glob("*.auto.tfvars.json"),
+            test_path.glob("terraform.tfstate*"),
+            test_path.glob("terraform.tfvars"),
+            test_path.glob(".terraform"),
+            # any symlinks?
+        )
+        for p in autopaths:
+          if p.is_dir():
+            shutil.rmtree(p)
+          else:
+            p.unlink()
       else:
         test_path = module_path
 
       # prepare tftest and run plan
       binary = os.environ.get('TERRAFORM', 'terraform')
-      tf = tftest.TerraformTest(test_path, basedir=basedir, binary=binary)
+      tf = tftest.TerraformTest(test_path, binary=binary)
       tf.setup(upgrade=True)
+      tf_var_files = [basedir / x for x in tf_var_files or []]
       plan = tf.plan(output=True, refresh=True, tf_var_file=tf_var_files,
                      tf_vars=tf_vars)
 
@@ -219,15 +262,17 @@ def generic_plan_summary():
 
 
 @pytest.fixture
-def generic_plan_validator(generic_plan_summary):
+def generic_plan_validator(generic_plan_summary, request):
+  'Return a function that builds a PlanSummary and compares it to an yaml inventory'
 
-  def inner(inventory_path, module_path, tf_var_files=None, basedir=None,
+  def inner(inventory_path, module_path, basedir=None, tf_var_files=None,
             **tf_vars):
 
+    if basedir is None:
+      basedir = Path(request.fspath).parent
+
     # allow tfvars and inventory to be relative to the caller
-    caller_path = Path(inspect.stack()[1].filename).parent
-    tf_var_files = [caller_path / x for x in tf_var_files]
-    inventory_path = caller_path / inventory_path
+    inventory_path = basedir / inventory_path
     inventory = yaml.safe_load(inventory_path.read_text())
     assert inventory is not None, f'Inventory {inventory_path} is empty'
 

tests/fast/stages/s00_bootstrap/test_plan.py

@@ -14,6 +14,8 @@
 
 
 def test_simple(generic_plan_validator):
-  s = generic_plan_validator(inventory_path='simple.yaml',
-                             module_path="fast/stages/00-bootstrap",
-                             tf_var_files=['simple.tfvars'])
+  generic_plan_validator(
+      inventory_path='simple.yaml',
+      module_path="fast/stages/00-bootstrap",
+      tf_var_files=['simple.tfvars'],
+  )

tests/modules/net_vpc/test_plan.py

@@ -14,12 +14,18 @@
 
 _VAR_ROUTES_TEMPLATE = '''{
   next-hop = {
-  dest_range="192.168.128.0/24", tags=null,
-  next_hop_type="%s", next_hop="%s"},
+    dest_range    = "192.168.128.0/24"
+    tags          = null
+    next_hop_type = "%s"
+    next_hop      = "%s"
+  }
   gateway = {
-  dest_range="0.0.0.0/0", priority=100, tags=["tag-a"],
-  next_hop_type="gateway",
-  next_hop="global/gateways/default-internet-gateway"}
+    dest_range    = "0.0.0.0/0",
+    priority      = 100
+    tags          = ["tag-a"]
+    next_hop_type = "gateway",
+    next_hop      = "global/gateways/default-internet-gateway"
+  }
 }'''
 _VAR_ROUTES_NEXT_HOPS = {
     'gateway': 'global/gateways/default-internet-gateway',
@@ -29,33 +35,40 @@ _VAR_ROUTES_NEXT_HOPS = {
     'vpn_tunnel': 'regions/europe-west1/vpnTunnels/foo'
 }
 
-import yaml
-
 
 def test_simple(generic_plan_validator):
-  generic_plan_validator(inventory_path='simple.yaml',
-                         module_path="modules/net-vpc",
-                         tf_var_files=['common.tfvars'])
+  generic_plan_validator(
+      inventory_path='simple.yaml',
+      module_path='modules/net-vpc',
+      tf_var_files=['common.tfvars'],
+  )
 
 
 def test_vpc_shared(generic_plan_validator):
-  generic_plan_validator(inventory_path='shared_vpc.yaml',
-                         module_path="modules/net-vpc",
-                         tf_var_files=['common.tfvars', 'shared_vpc.tfvars'])
+  generic_plan_validator(
+      inventory_path='shared_vpc.yaml',
+      module_path='modules/net-vpc',
+      tf_var_files=['common.tfvars', 'shared_vpc.tfvars'],
+  )
 
 
 def test_vpc_peering(generic_plan_validator):
-  generic_plan_validator(inventory_path='peering.yaml',
-                         module_path="modules/net-vpc",
-                         tf_var_files=['common.tfvars', 'peering.tfvars'])
+  generic_plan_validator(
+      inventory_path='peering.yaml',
+      module_path='modules/net-vpc',
+      tf_var_files=['common.tfvars', 'peering.tfvars'],
+  )
 
 
-def test_vpc_routes(plan_runner):
-  "Test vpc routes."
+def test_vpc_routes(generic_plan_summary):
+  'Test vpc routes.'
   for next_hop_type, next_hop in _VAR_ROUTES_NEXT_HOPS.items():
-    _var_routes = _VAR_ROUTES_TEMPLATE % (next_hop_type, next_hop)
-    _, resources = plan_runner(routes=_var_routes)
-    assert len(resources) == 3
-    resource = [r for r in resources if r['values']['name'] == 'test-next-hop'
-               ][0]
-    assert resource['values']['next_hop_%s' % next_hop_type]
+    var_routes = _VAR_ROUTES_TEMPLATE % (next_hop_type, next_hop)
+    summary = generic_plan_summary(
+        module_path='modules/net-vpc',
+        tf_var_files=['common.tfvars'],
+        routes=var_routes,
+    )
+    assert len(summary.values) == 3
+    route = summary.values[f'google_compute_route.{next_hop_type}["next-hop"]']
+    assert route[f'next_hop_{next_hop_type}'] == next_hop

tests/modules/net_vpc/test_plan_subnets.py

@@ -15,6 +15,7 @@
 DATA_FOLDER = "data"
 
 import yaml
+from pathlib import Path
 
 
 def test_subnet_factory(plan_runner):
@@ -30,6 +31,8 @@ def test_subnet_factory(plan_runner):
 
 
 def test_subnets(generic_plan_validator):
-  generic_plan_validator(inventory_path='subnets.yaml',
-                         module_path="modules/net-vpc",
-                         tf_var_files=['common.tfvars', 'subnets.tfvars'])
+  generic_plan_validator(
+      inventory_path='subnets.yaml',
+      module_path="modules/net-vpc",
+      tf_var_files=['common.tfvars', 'subnets.tfvars'],
+  )