diff --git a/tests/conftest.py b/tests/conftest.py index 0b23ed084..fc2ea4aa3 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -158,179 +158,221 @@ def basedir(): return BASEDIR +def _generic_plan_summary(module_path, tf_var_files=None, basedir=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 + + ''' + + module_path = Path(BASEDIR) / module_path + + # FIXME: find a way to prevent the temp dir if TFTEST_COPY is not + # in the environment + with tempfile.TemporaryDirectory(dir=module_path.parent) as tmp_path: + # if TFTEST_COPY is set, copy the fixture to a temporary + # directory before running the plan. This is needed if you want + # to run multiple tests for the same module in parallel + if os.environ.get('TFTEST_COPY'): + test_path = Path(tmp_path) + shutil.copytree(module_path, test_path, dirs_exist_ok=True) + + # 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 + # surprises if, for example, 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, 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) + + # compute resource type counts and address->values map + values = {} + counts = collections.defaultdict(int) + q = collections.deque([plan.root_module]) + while q: + e = q.popleft() + + if 'type' in e: + counts[e['type']] += 1 + if 'values' in e: + values[e['address']] = e['values'] + + for x in e.get('resources', []): + q.append(x) + for x in e.get('child_modules', []): + q.append(x) + + # extract planned outputs + outputs = plan.get('planned_values', {}).get('outputs', {}) + + return PlanSummary(values, dict(counts), outputs) + + @pytest.fixture def generic_plan_summary(request): 'Returns a function to generate a PlanSummary' def inner(module_path, tf_var_files=None, basedir=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 - with tempfile.TemporaryDirectory(dir=module_path.parent) as tmp_path: - # if TFTEST_COPY is set, copy the fixture to a temporary - # directory before running the plan. This is needed if you want - # to run multiple tests for the same module in parallel - if os.environ.get('TFTEST_COPY'): - test_path = Path(tmp_path) - shutil.copytree(module_path, test_path, dirs_exist_ok=True) - - # 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 - # surprises if, for example, 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, 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) - - # compute resource type counts and address->values map - values = {} - counts = collections.defaultdict(int) - q = collections.deque([plan.root_module]) - while q: - e = q.popleft() - - if 'type' in e: - counts[e['type']] += 1 - if 'values' in e: - values[e['address']] = e['values'] - - for x in e.get('resources', []): - q.append(x) - for x in e.get('child_modules', []): - q.append(x) - - # extract planned outputs - outputs = plan.get('planned_values', {}).get('outputs', {}) - - return PlanSummary(values, dict(counts), outputs) + return _generic_plan_summary(module_path, tf_var_files, basedir, **tf_vars) return inner +def _generic_plan_validator(module_path, inventory_paths, tf_var_files=None, + basedir=None, **tf_vars): + summary = _generic_plan_summary(module_path=module_path, + tf_var_files=tf_var_files, basedir=basedir, + **tf_vars) + + # allow single single string for inventory_paths + if not isinstance(inventory_paths, list): + inventory_paths = [inventory_paths] + + for path in inventory_paths: + # allow tfvars and inventory to be relative to the caller + path = basedir / path + inventory = yaml.safe_load(path.read_text()) + assert inventory is not None, f'Inventory {path} is empty' + + # If you add additional asserts to this function: + # - put the values coming from the plan on the left side of + # any comparison operators + # - put the values coming from user's inventory the right + # side of any comparison operators. + # - include a descriptive error message to the assert + + # for values: + # - verify each address in the user's inventory exists in the plan + # - for those address that exist on both the user's inventory and + # the plan output, ensure the set of keys on the inventory are a + # subset of the keys in the plan, and compare their values by + # equality + if 'values' in inventory: + expected_values = inventory['values'] + for address, expected_value in expected_values.items(): + assert address in summary.values, \ + f'{address} is not a valid address in the plan' + for k, v in expected_value.items(): + assert k in summary.values[address], \ + f'{k} not found at {address}' + plan_value = summary.values[address][k] + assert plan_value == v, \ + f'{k} at {address} failed. Got `{plan_value}`, expected `{v}`' + + if 'counts' in inventory: + expected_counts = inventory['counts'] + for type_, expected_count in expected_counts.items(): + assert type_ in summary.counts, \ + f'module does not create any resources of type `{type_}`' + plan_count = summary.counts[type_] + assert plan_count == expected_count, \ + f'count of {type_} resources failed. Got {plan_count}, expected {expected_count}' + + if 'outputs' in inventory: + expected_outputs = inventory['outputs'] + for output_name, expected_output in expected_outputs.items(): + assert output_name in summary.outputs, \ + f'module does not output `{output_name}`' + output = summary.outputs[output_name] + # assert 'value' in output, \ + # f'output `{output_name}` does not have a value (is it sensitive or dynamic?)' + plan_output = output.get('value', '__missing__') + assert plan_output == expected_output, \ + f'output {output_name} failed. Got `{plan_output}`, expected `{expected_output}`' + + return summary + + @pytest.fixture def generic_plan_validator(generic_plan_summary, request): 'Return a function that builds a PlanSummary and compares it to an yaml inventory' def inner(module_path, inventory_paths, tf_var_files=None, basedir=None, **tf_vars): - if basedir is None: basedir = Path(request.fspath).parent - - summary = generic_plan_summary(module_path=module_path, - tf_var_files=tf_var_files, basedir=basedir, - **tf_vars) - - # allow single single string for inventory_paths - if not isinstance(inventory_paths, list): - inventory_paths = [inventory_paths] - - for path in inventory_paths: - # allow tfvars and inventory to be relative to the caller - path = basedir / path - inventory = yaml.safe_load(path.read_text()) - assert inventory is not None, f'Inventory {path} is empty' - - # If you add additional asserts to this function: - # - put the values coming from the plan on the left side of - # any comparison operators - # - put the values coming from user's inventory the right - # side of any comparison operators. - # - include a descriptive error message to the assert - - # for values: - # - verify each address in the user's inventory exists in the plan - # - for those address that exist on both the user's inventory and - # the plan output, ensure the set of keys on the inventory are a - # subset of the keys in the plan, and compare their values by - # equality - if 'values' in inventory: - expected_values = inventory['values'] - for address, expected_value in expected_values.items(): - assert address in summary.values, \ - f'{address} is not a valid address in the plan' - for k, v in expected_value.items(): - assert k in summary.values[address], \ - f'{k} not found at {address}' - plan_value = summary.values[address][k] - assert plan_value == v, \ - f'{k} at {address} failed. Got `{plan_value}`, expected `{v}`' - - if 'counts' in inventory: - expected_counts = inventory['counts'] - for type_, expected_count in expected_counts.items(): - assert type_ in summary.counts, \ - f'module does not create any resources of type `{type_}`' - plan_count = summary.counts[type_] - assert plan_count == expected_count, \ - f'count of {type_} resources failed. Got {plan_count}, expected {expected_count}' - - if 'outputs' in inventory: - expected_outputs = inventory['outputs'] - for output_name, expected_output in expected_outputs.items(): - assert output_name in summary.outputs, \ - f'module does not output `{output_name}`' - output = summary.outputs[output_name] - # assert 'value' in output, \ - # f'output `{output_name}` does not have a value (is it sensitive or dynamic?)' - plan_output = output.get('value', '__missing__') - assert plan_output == expected_output, \ - f'output {output_name} failed. Got `{plan_output}`, expected `{expected_output}`' - - return summary + return _generic_plan_validator(module_path, inventory_paths, tf_var_files, + basedir, **tf_vars) return inner + + +def pytest_collect_file(parent, file_path): + if file_path.suffix == ".yaml" and file_path.name.startswith("tftest"): + return YamlFile.from_parent(parent, path=file_path) + + +class YamlFile(pytest.File): + + def collect(self): + raw = yaml.safe_load(self.path.open()) + module = raw['module'] + for test_name, spec in raw['tests'].items(): + inventory = spec.get('inventory', f'{test_name}.yaml') + tfvars = spec['tfvars'] + yield YamlItem.from_parent(self, name=test_name, module=module, + inventory=inventory, tfvars=tfvars) + + +class YamlItem(pytest.Item): + + def __init__(self, name, parent, module, inventory, tfvars): + super().__init__(name, parent) + self.module = module + self.inventory = inventory + self.tfvars = tfvars + + def runtest(self): + _generic_plan_validator(self.module, self.inventory, self.tfvars, + self.parent.path.parent) + + def reportinfo(self): + return self.path, None, self.name diff --git a/tests/modules/net_vpc/test_plan.py b/tests/modules/net_vpc/test_plan.py deleted file mode 100644 index 5deec3004..000000000 --- a/tests/modules/net_vpc/test_plan.py +++ /dev/null @@ -1,61 +0,0 @@ -# Copyright 2022 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -_VAR_ROUTES_TEMPLATE = '''{ - next-hop = { - 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" - } -}''' -_VAR_ROUTES_NEXT_HOPS = { - 'gateway': 'global/gateways/default-internet-gateway', - 'instance': 'zones/europe-west1-b/test', - 'ip': '192.168.0.128', - 'ilb': 'regions/europe-west1/forwardingRules/test', - 'vpn_tunnel': 'regions/europe-west1/vpnTunnels/foo' -} - - -def test_simple(generic_plan_validator): - generic_plan_validator('modules/net-vpc', 'simple.yaml', ['common.tfvars']) - - -def test_vpc_shared(generic_plan_validator): - generic_plan_validator('modules/net-vpc', 'shared_vpc.yaml', - ['common.tfvars', 'shared_vpc.tfvars']) - - -def test_vpc_peering(generic_plan_validator): - generic_plan_validator('modules/net-vpc', 'peering.yaml', - ['common.tfvars', 'peering.tfvars']) - - -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) - summary = generic_plan_summary('modules/net-vpc', ['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 diff --git a/tests/modules/net_vpc/test_plan_psa.py b/tests/modules/net_vpc/test_plan_psa.py deleted file mode 100644 index ad26e559e..000000000 --- a/tests/modules/net_vpc/test_plan_psa.py +++ /dev/null @@ -1,35 +0,0 @@ -# Copyright 2022 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -import yaml - - -def test_simple(generic_plan_validator): - generic_plan_validator("modules/net-vpc", 'psa_simple.yaml', - ['common.tfvars', 'psa_simple.tfvars']) - - -def test_routes_export(generic_plan_validator): - generic_plan_validator("modules/net-vpc", 'psa_routes_export.yaml', - ['common.tfvars', 'psa_routes_export.tfvars']) - - -def test_routes_import(generic_plan_validator): - generic_plan_validator("modules/net-vpc", 'psa_routes_import.yaml', - ['common.tfvars', 'psa_routes_import.tfvars']) - - -def test_routes_import_export(generic_plan_validator): - generic_plan_validator("modules/net-vpc", 'psa_routes_import_export.yaml', - ['common.tfvars', 'psa_routes_import_export.tfvars']) diff --git a/tests/modules/net_vpc/test_routes.py b/tests/modules/net_vpc/test_routes.py new file mode 100644 index 000000000..8ee81ad4e --- /dev/null +++ b/tests/modules/net_vpc/test_routes.py @@ -0,0 +1,47 @@ +# Copyright 2022 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import pytest + +_route_parameters = [('gateway', 'global/gateways/default-internet-gateway'), + ('instance', 'zones/europe-west1-b/test'), + ('ip', '192.168.0.128'), + ('ilb', 'regions/europe-west1/forwardingRules/test'), + ('vpn_tunnel', 'regions/europe-west1/vpnTunnels/foo')] + + +@pytest.mark.parametrize('next_hop_type,next_hop', _route_parameters) +def test_vpc_routes(generic_plan_summary, next_hop_type, next_hop): + 'Test vpc routes.' + + var_routes = '''{ + next-hop = { + 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" + } + }''' % (next_hop_type, next_hop) + summary = generic_plan_summary('modules/net-vpc', ['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 diff --git a/tests/modules/net_vpc/tftest.psa.yaml b/tests/modules/net_vpc/tftest.psa.yaml new file mode 100644 index 000000000..1d59c861f --- /dev/null +++ b/tests/modules/net_vpc/tftest.psa.yaml @@ -0,0 +1,22 @@ +module: modules/net-vpc + +tests: + psa_simple: + tfvars: + - common.tfvars + - psa_simple.tfvars + + psa_routes_export: + tfvars: + - common.tfvars + - psa_routes_export.tfvars + + psa_routes_import: + tfvars: + - common.tfvars + - psa_routes_import.tfvars + + psa_routes_import_export: + tfvars: + - common.tfvars + - psa_routes_import_export.tfvars diff --git a/tests/modules/net_vpc/tftest.yaml b/tests/modules/net_vpc/tftest.yaml new file mode 100644 index 000000000..387ca9ee0 --- /dev/null +++ b/tests/modules/net_vpc/tftest.yaml @@ -0,0 +1,27 @@ +module: modules/net-vpc + +tests: + simple: + tfvars: + - common.tfvars + inventory: + - simple.yaml + + subnets: + tfvars: + - common.tfvars + - subnets.tfvars + + peering: + tfvars: + - common.tfvars + - peering.tfvars + inventory: + - peering.yaml + + shared_vpc: + tfvars: + - common.tfvars + - shared_vpc.tfvars + inventory: + - shared_vpc.yaml