Bring back `tests` key in test yaml spec

This commit is contained in:
Julio Castillo 2022-12-06 00:02:16 +01:00
parent fded49cc67
commit be0e807435
7 changed files with 276 additions and 240 deletions

View File

@ -11,6 +11,13 @@
# 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.
"""Pytest plugin to discover tests specified in YAML files.
This plugin uses the pytest_collect_file hook to collect all files
matching tftest*.yaml and runs plan_validate for each test found.
See FabricTestFile for details on the file structure.
"""
import pytest
import yaml
@ -23,11 +30,11 @@ class FabricTestFile(pytest.File):
def collect(self):
"""Read yaml test spec and yield test items for each test definition.
Test spec should contain a `module` key with the path of the
The test spec should contain a `module` key with the path of the
terraform module to test, relative to the root of the repository
All other top-level keys in the yaml are taken as test names, and
should have the following structure:
Tests are defined within the top-level `tests` key, and should
have the following structure:
test-name:
tfvars:
@ -42,6 +49,7 @@ class FabricTestFile(pytest.File):
will be taken from the file test-name.yaml
"""
try:
raw = yaml.safe_load(self.path.open())
module = raw.pop('module')
@ -49,13 +57,13 @@ class FabricTestFile(pytest.File):
raise Exception(f'cannot read test spec {self.path}: {e}')
except KeyError as e:
raise Exception(f'`module` key not found in {self.path}: {e}')
for test_name, spec in raw.items():
for test_name, spec in raw.get('tests', {}).items():
inventories = spec.get('inventory', [f'{test_name}.yaml'])
try:
tfvars = spec['tfvars']
except KeyError:
raise Exception(
f'test definition `{test_name}` in {self.path} does not contain a `tfvars` key'
f'test `{test_name}` in {self.path} does not contain a `tfvars` key'
)
for i in inventories:
name = test_name

View File

@ -11,147 +11,12 @@
# 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.
'Shared fixtures.'
import inspect
import os
import shutil
import tempfile
'Pytest configuration.'
import pytest
import tftest
import yaml
BASEDIR = os.path.dirname(os.path.dirname(__file__))
pytest_plugins = ('tests.fixtures', 'tests.collectors')
@pytest.fixture(scope='session')
def _plan_runner():
'Return a function to run Terraform plan on a fixture.'
def run_plan(fixture_path=None, extra_files=None, tf_var_file=None,
targets=None, refresh=True, tmpdir=True, **tf_vars):
'Run Terraform plan and returns parsed output.'
if fixture_path is None:
# find out the fixture directory from the caller's directory
caller = inspect.stack()[2]
fixture_path = os.path.join(os.path.dirname(caller.filename), 'fixture')
fixture_parent = os.path.dirname(fixture_path)
fixture_prefix = os.path.basename(fixture_path) + '_'
with tempfile.TemporaryDirectory(prefix=fixture_prefix,
dir=fixture_parent) as tmp_path:
# copy fixture to a temporary directory so we can execute
# multiple tests in parallel
if tmpdir:
shutil.copytree(fixture_path, tmp_path, dirs_exist_ok=True)
tf = tftest.TerraformTest(tmp_path if tmpdir else fixture_path, BASEDIR,
os.environ.get('TERRAFORM', 'terraform'))
tf.setup(extra_files=extra_files, upgrade=True)
plan = tf.plan(output=True, refresh=refresh, tf_var_file=tf_var_file,
tf_vars=tf_vars, targets=targets)
return plan
return run_plan
@pytest.fixture(scope='session')
def plan_runner(_plan_runner):
'Return a function to run Terraform plan on a module fixture.'
def run_plan(fixture_path=None, extra_files=None, tf_var_file=None,
targets=None, **tf_vars):
'Run Terraform plan and returns plan and module resources.'
plan = _plan_runner(fixture_path, extra_files=extra_files,
tf_var_file=tf_var_file, targets=targets, **tf_vars)
# skip the fixture
root_module = plan.root_module['child_modules'][0]
return plan, root_module['resources']
return run_plan
@pytest.fixture(scope='session')
def e2e_plan_runner(_plan_runner):
'Return a function to run Terraform plan on an end-to-end fixture.'
def run_plan(fixture_path=None, tf_var_file=None, targets=None, refresh=True,
include_bare_resources=False, **tf_vars):
'Run Terraform plan on an end-to-end module using defaults, returns data.'
plan = _plan_runner(fixture_path, tf_var_file=tf_var_file, targets=targets,
refresh=refresh, **tf_vars)
# skip the fixture
root_module = plan.root_module['child_modules'][0]
modules = dict((mod['address'], mod['resources'])
for mod in root_module['child_modules'])
resources = [r for m in modules.values() for r in m]
if include_bare_resources:
bare_resources = root_module['resources']
resources.extend(bare_resources)
return modules, resources
return run_plan
@pytest.fixture(scope='session')
def recursive_e2e_plan_runner(_plan_runner):
"""
Plan runner for end-to-end root module, returns total number of
(nested) modules and resources
"""
def walk_plan(node, modules, resources):
new_modules = node.get('child_modules', [])
resources += node.get('resources', [])
modules += new_modules
for module in new_modules:
walk_plan(module, modules, resources)
def run_plan(fixture_path=None, tf_var_file=None, targets=None, refresh=True,
include_bare_resources=False, compute_sums=True, tmpdir=True,
**tf_vars):
'Run Terraform plan on a root module using defaults, returns data.'
plan = _plan_runner(fixture_path, tf_var_file=tf_var_file, targets=targets,
refresh=refresh, tmpdir=tmpdir, **tf_vars)
modules = []
resources = []
walk_plan(plan.root_module, modules, resources)
return len(modules), len(resources)
return run_plan
@pytest.fixture(scope='session')
def apply_runner():
'Return a function to run Terraform apply on a fixture.'
def run_apply(fixture_path=None, **tf_vars):
'Run Terraform plan and returns parsed output.'
if fixture_path is None:
# find out the fixture directory from the caller's directory
caller = inspect.stack()[1]
fixture_path = os.path.join(os.path.dirname(caller.filename), 'fixture')
fixture_parent = os.path.dirname(fixture_path)
fixture_prefix = os.path.basename(fixture_path) + '_'
with tempfile.TemporaryDirectory(prefix=fixture_prefix,
dir=fixture_parent) as tmp_path:
# copy fixture to a temporary directory so we can execute
# multiple tests in parallel
shutil.copytree(fixture_path, tmp_path, dirs_exist_ok=True)
tf = tftest.TerraformTest(tmp_path, BASEDIR,
os.environ.get('TERRAFORM', 'terraform'))
tf.setup(upgrade=True)
apply = tf.apply(tf_vars=tf_vars)
output = tf.output(json_format=True)
return apply, output
return run_apply
@pytest.fixture
def basedir():
return BASEDIR
pytest_plugins = (
'tests.fixtures',
'tests.legacy_fixtures',
'tests.collectors',
)

View File

@ -2,10 +2,11 @@
module: fast/stages/00-bootstrap
simple:
tfvars:
- simple.tfvars
inventory:
- simple.yaml
- simple_projects.yaml
- simple_sas.yaml
tests:
simple:
tfvars:
- simple.tfvars
inventory:
- simple.yaml
- simple_projects.yaml
- simple_sas.yaml

View File

@ -11,6 +11,8 @@
# 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.
"""Common fixtures."""
import collections
import itertools
import os
@ -205,8 +207,7 @@ def plan_validator(module_path, inventory_paths, basedir, tf_var_files=None,
@pytest.fixture(name='plan_validator')
def plan_validator_fixture(request):
"""Return a function to builds a PlanSummary and compare it to YAML
inventory.
"""Return a function to build a PlanSummary and compare it to a YAML inventory.
In the returned function `basedir` becomes optional and it defaults
to the directory of the calling test'
@ -222,3 +223,9 @@ def plan_validator_fixture(request):
tf_var_files=tf_var_paths, **tf_vars)
return inner
# @pytest.fixture
# def repo_root():
# 'Return a pathlib.Path to the root of the repository'
# return Path(__file__).parents[1]

153
tests/legacy_fixtures.py Normal file
View File

@ -0,0 +1,153 @@
# 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.
"""Legacy pytest fixtures.
The fixtures contained in this file will eventually go away. Consider
using one of the fixtures in fixtures.py
"""
import inspect
import os
import shutil
import tempfile
import pytest
import tftest
BASEDIR = os.path.dirname(os.path.dirname(__file__))
@pytest.fixture(scope='session')
def _plan_runner():
'Return a function to run Terraform plan on a fixture.'
def run_plan(fixture_path=None, extra_files=None, tf_var_file=None,
targets=None, refresh=True, tmpdir=True, **tf_vars):
'Run Terraform plan and returns parsed output.'
if fixture_path is None:
# find out the fixture directory from the caller's directory
caller = inspect.stack()[2]
fixture_path = os.path.join(os.path.dirname(caller.filename), 'fixture')
fixture_parent = os.path.dirname(fixture_path)
fixture_prefix = os.path.basename(fixture_path) + '_'
with tempfile.TemporaryDirectory(prefix=fixture_prefix,
dir=fixture_parent) as tmp_path:
# copy fixture to a temporary directory so we can execute
# multiple tests in parallel
if tmpdir:
shutil.copytree(fixture_path, tmp_path, dirs_exist_ok=True)
tf = tftest.TerraformTest(tmp_path if tmpdir else fixture_path, BASEDIR,
os.environ.get('TERRAFORM', 'terraform'))
tf.setup(extra_files=extra_files, upgrade=True)
plan = tf.plan(output=True, refresh=refresh, tf_var_file=tf_var_file,
tf_vars=tf_vars, targets=targets)
return plan
return run_plan
@pytest.fixture(scope='session')
def plan_runner(_plan_runner):
'Return a function to run Terraform plan on a module fixture.'
def run_plan(fixture_path=None, extra_files=None, tf_var_file=None,
targets=None, **tf_vars):
'Run Terraform plan and returns plan and module resources.'
plan = _plan_runner(fixture_path, extra_files=extra_files,
tf_var_file=tf_var_file, targets=targets, **tf_vars)
# skip the fixture
root_module = plan.root_module['child_modules'][0]
return plan, root_module['resources']
return run_plan
@pytest.fixture(scope='session')
def e2e_plan_runner(_plan_runner):
'Return a function to run Terraform plan on an end-to-end fixture.'
def run_plan(fixture_path=None, tf_var_file=None, targets=None, refresh=True,
include_bare_resources=False, **tf_vars):
'Run Terraform plan on an end-to-end module using defaults, returns data.'
plan = _plan_runner(fixture_path, tf_var_file=tf_var_file, targets=targets,
refresh=refresh, **tf_vars)
# skip the fixture
root_module = plan.root_module['child_modules'][0]
modules = dict((mod['address'], mod['resources'])
for mod in root_module['child_modules'])
resources = [r for m in modules.values() for r in m]
if include_bare_resources:
bare_resources = root_module['resources']
resources.extend(bare_resources)
return modules, resources
return run_plan
@pytest.fixture(scope='session')
def recursive_e2e_plan_runner(_plan_runner):
"""
Plan runner for end-to-end root module, returns total number of
(nested) modules and resources
"""
def walk_plan(node, modules, resources):
new_modules = node.get('child_modules', [])
resources += node.get('resources', [])
modules += new_modules
for module in new_modules:
walk_plan(module, modules, resources)
def run_plan(fixture_path=None, tf_var_file=None, targets=None, refresh=True,
include_bare_resources=False, compute_sums=True, tmpdir=True,
**tf_vars):
'Run Terraform plan on a root module using defaults, returns data.'
plan = _plan_runner(fixture_path, tf_var_file=tf_var_file, targets=targets,
refresh=refresh, tmpdir=tmpdir, **tf_vars)
modules = []
resources = []
walk_plan(plan.root_module, modules, resources)
return len(modules), len(resources)
return run_plan
@pytest.fixture(scope='session')
def apply_runner():
'Return a function to run Terraform apply on a fixture.'
def run_apply(fixture_path=None, **tf_vars):
'Run Terraform plan and returns parsed output.'
if fixture_path is None:
# find out the fixture directory from the caller's directory
caller = inspect.stack()[1]
fixture_path = os.path.join(os.path.dirname(caller.filename), 'fixture')
fixture_parent = os.path.dirname(fixture_path)
fixture_prefix = os.path.basename(fixture_path) + '_'
with tempfile.TemporaryDirectory(prefix=fixture_prefix,
dir=fixture_parent) as tmp_path:
# copy fixture to a temporary directory so we can execute
# multiple tests in parallel
shutil.copytree(fixture_path, tmp_path, dirs_exist_ok=True)
tf = tftest.TerraformTest(tmp_path, BASEDIR,
os.environ.get('TERRAFORM', 'terraform'))
tf.setup(upgrade=True)
apply = tf.apply(tf_vars=tf_vars)
output = tf.output(json_format=True)
return apply, output
return run_apply

View File

@ -14,46 +14,47 @@
module: modules/net-vpc
simple:
tfvars:
- common.tfvars
tests:
simple:
tfvars:
- common.tfvars
subnets:
tfvars:
- common.tfvars
- subnets.tfvars
subnets:
tfvars:
- common.tfvars
- subnets.tfvars
peering:
tfvars:
- common.tfvars
- peering.tfvars
peering:
tfvars:
- common.tfvars
- peering.tfvars
shared_vpc:
tfvars:
- common.tfvars
- shared_vpc.tfvars
shared_vpc:
tfvars:
- common.tfvars
- shared_vpc.tfvars
factory:
tfvars:
- common.tfvars
- factory.tfvars
factory:
tfvars:
- common.tfvars
- factory.tfvars
psa_simple:
tfvars:
- common.tfvars
- psa_simple.tfvars
psa_simple:
tfvars:
- common.tfvars
- psa_simple.tfvars
psa_routes_export:
tfvars:
- common.tfvars
- psa_routes_export.tfvars
psa_routes_export:
tfvars:
- common.tfvars
- psa_routes_export.tfvars
psa_routes_import:
tfvars:
- common.tfvars
- psa_routes_import.tfvars
psa_routes_import:
tfvars:
- common.tfvars
- psa_routes_import.tfvars
psa_routes_import_export:
tfvars:
- common.tfvars
- psa_routes_import_export.tfvars
psa_routes_import_export:
tfvars:
- common.tfvars
- psa_routes_import_export.tfvars

View File

@ -14,64 +14,65 @@
module: modules/organization
audit_config:
tfvars:
- common.tfvars
- audit_config.tfvars
tests:
audit_config:
tfvars:
- common.tfvars
- audit_config.tfvars
iam:
tfvars:
- common.tfvars
- iam.tfvars
iam:
tfvars:
- common.tfvars
- iam.tfvars
iam_additive:
tfvars:
- common.tfvars
- iam_additive.tfvars
iam_additive:
tfvars:
- common.tfvars
- iam_additive.tfvars
logging:
tfvars:
- common.tfvars
- logging.tfvars
logging:
tfvars:
- common.tfvars
- logging.tfvars
logging_exclusions:
tfvars:
- common.tfvars
- logging_exclusions.tfvars
logging_exclusions:
tfvars:
- common.tfvars
- logging_exclusions.tfvars
org_policies_list:
tfvars:
- common.tfvars
- org_policies_list.tfvars
org_policies_list:
tfvars:
- common.tfvars
- org_policies_list.tfvars
org_policies_boolean:
tfvars:
- common.tfvars
- org_policies_boolean.tfvars
org_policies_boolean:
tfvars:
- common.tfvars
- org_policies_boolean.tfvars
org_policies_custom_constraints:
tfvars:
- common.tfvars
- org_policies_custom_constraints.tfvars
org_policies_custom_constraints:
tfvars:
- common.tfvars
- org_policies_custom_constraints.tfvars
tags:
tfvars:
- common.tfvars
- network_tags.tfvars
- resource_tags.tfvars
tags:
tfvars:
- common.tfvars
- network_tags.tfvars
- resource_tags.tfvars
firewall_policies:
tfvars:
- common.tfvars
- firewall_policies.tfvars
firewall_policies:
tfvars:
- common.tfvars
- firewall_policies.tfvars
firewall_policies_factory:
tfvars:
- common.tfvars
- firewall_policies_factory.tfvars
firewall_policies_factory:
tfvars:
- common.tfvars
- firewall_policies_factory.tfvars
firewall_policies_factory_combined:
tfvars:
- common.tfvars
- firewall_policies.tfvars
- firewall_policies_factory.tfvars
firewall_policies_factory_combined:
tfvars:
- common.tfvars
- firewall_policies.tfvars
- firewall_policies_factory.tfvars