kovagoadi/hunfabric
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

New generate_plan_summary.py tool

Browse Source
This commit is contained in:
Julio Castillo
2026-04-27 12:00:54 +02:00
parent 8450edbf2b
commit 99f53d37f0
5 changed files with 359 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
CONTRIBUTING.md
View File

@@ -914,7 +914,15 @@ pytest -k 'modules and gke-cluster-autopilot: and monitoring and :2' tests/examp
#### Generating the inventory automatically
Building an inventory file by hand is difficult. To simplify this task, the default test runner for examples prints the inventory for the full plan if it succeeds. Therefore, you can start without an inventory and then run a test to get the full plan and extract the pieces you want to build the inventory file.
Building an inventory file by hand is difficult. To simplify this task, you can use the unified tool `tools/generate_plan_summary.py` to generate the inventory. This script parses the README, extracts any embedded files or fixtures, and runs the plan. It can also automatically save the inventory to the correct location if you pass the `--save` flag and have specified `inventory=filename.yaml` in the `# tftest` directive in the README.
```bash
uv run tools/generate_plan_summary.py modules/dns/README.md "Private Zone" --save
```
#### Alternative: Generating the inventory via `pytest` (Legacy)
The default test runner for examples also prints the inventory for the full plan if it succeeds. Therefore, you can start without an inventory and then run a test to get the full plan and extract the pieces you want to build the inventory file.
Suppose you want to generate the inventory for the last DNS example above (the one creating the recordsets from a YAML file). Assuming that example is the first code block under the "Private Zone" section in the README for the `dns` module, you can run the following command to build the inventory:
@@ -1103,59 +1111,25 @@ Run the specific `pytest` plan test. The test will fail, and the captured output
#### Generating the inventory for `tftest`-based tests
Just as you can generate an initial inventory for example-based tests, you can do the same for `tftest`-based tests. Currently the process relies on an additional tool (`tools/plan_summary.py`) but but we have plans to unify both cases in the future.
Just as you can generate an initial inventory for example-based tests, you can do the same for `tftest`-based tests using the unified tool `tools/generate_plan_summary.py`.
As an example, if you want to generate the inventory for the `organization` module using the `common.tfvars` and `audit_config.tfvars` found in `tests/modules/organization/`, simply run `plan_summary.py` as follows:
As an example, if you want to generate the inventory for the `organization` module for the test case `audit_config` defined in its `tftest.yaml`, simply run:
```bash
$ python tools/plan_summary.py modules/organization \
tests/modules/organization/common.tfvars \
tests/modules/organization/audit_config.tfvars
values:
google_organization_iam_audit_config.config["allServices"]:
audit_log_config:
- exempted_members:
- user:me@example.org
log_type: DATA_WRITE
- exempted_members: []
log_type: DATA_READ
org_id: '1234567890'
service: allServices
counts:
google_organization_iam_audit_config: 1
modules: 0
resources: 1
outputs:
custom_role_id: {}
custom_roles: {}
firewall_policies: {}
firewall_policy_id: {}
network_tag_keys: {}
network_tag_values: {}
organization_id: organizations/1234567890
sink_writer_identities: {}
tag_keys: {}
tag_values: {}
uv run tools/generate_plan_summary.py tests/modules/organization/tftest.yaml audit_config
```
You can now use this output to create the inventory file for your test. As mentioned before, please only use those values relevant to your test scenario.
You can optionally pass to the command additional files that your plan might need to properly execute.
In this example we pass in two extra files from the organization folder.
If you want to automatically save the generated inventory to the correct location (e.g., `tests/modules/organization/audit_config.yaml`), add the `--save` flag:
```bash
$ python tools/plan_summary.py modules/organization \
tests/modules/organization/common.tfvars \
tests/modules/organization/audit_config.tfvars \
--extra-files ../my-file-1.tf \
--extra-files ../my-file-2.yaml
uv run tools/generate_plan_summary.py tests/modules/organization/tftest.yaml audit_config --save
```
This will generate the inventory file with the correct structure and a valid license header.
If your test requires extra files or directories, you should specify them in the `tftest.yaml` file under the specific test case (using `extra_files` or `extra_dirs`), rather than passing them via command line flags.
### Running end-to-end tests
You can use end-to-end tests to verify your code against GCP API. These tests verify that `terraform apply` succeeds, `terraform plan` is empty afterwards and that `terraform destroy` raises no error.

78
tests/examples/conftest.py
View File

@@ -21,7 +21,7 @@ from pathlib import Path
import marko
import pytest
from .utils import File, TerraformExample, YamlExample, get_tftest_directive
from .utils import File, TerraformExample, YamlExample, get_tftest_directive, get_readme_examples
FABRIC_ROOT = Path(__file__).parents[2]
@@ -35,70 +35,22 @@ def pytest_generate_tests(metafunc, test_group='example',
ids = []
for readme in readmes:
module = readme.parent
doc = marko.parse(readme.read_text())
index = 0
files = collections.defaultdict(dict)
fixtures = {}
# first pass: collect all examples tagged with tftest-file
last_header = None
for child in doc.children:
if isinstance(child, marko.block.FencedCode):
code = child.children[0].children
directive = get_tftest_directive(code)
if directive is None:
continue
if directive.name == 'tftest-file':
name, path = directive.kwargs['id'], directive.kwargs['path']
files[last_header][name] = File(path, code)
if directive.name == 'tftest-fixture':
name = directive.kwargs['id']
fixtures[name] = code
elif isinstance(child, marko.block.Heading):
last_header = child.children[0].children
# second pass: collect all examples tagged with tftest
last_header = None
index = 0
for child in doc.children:
if isinstance(child, marko.block.FencedCode):
index += 1
code = child.children[0].children
directive = get_tftest_directive(code)
if directive is None:
continue
if directive and not filter_tests(directive.args):
readme_examples = get_readme_examples(readme, FABRIC_ROOT)
for example, example_id, marks, header, index in readme_examples:
if isinstance(example, TerraformExample):
if not filter_tests(example.directive.args):
continue
if os.environ.get(
'TERRAFORM') == 'tofu' and 'skip-tofu' in directive.args:
'TERRAFORM') == 'tofu' and 'skip-tofu' in example.directive.args:
continue
if child.lang in ('hcl', 'tfvars'):
path = module.relative_to(FABRIC_ROOT)
name = f'{path}:{last_header}'
if index > 1:
name += f' {index}'
ids.append(f'terraform:{path}:{last_header}:{index}')
# if test is marked with 'serial' in tftest line then add them to this xdist group
# this, together with `--dist loadgroup` will ensure that those tests will be run one after another
# even if multiple workers are used
# see: https://pytest-xdist.readthedocs.io/en/latest/distribution.html
marks = [pytest.mark.xdist_group('serial')
] if 'serial' in directive.args else []
example = TerraformExample(name, code, path, files[last_header],
fixtures, child.lang, directive)
examples.append(pytest.param(example, marks=marks))
elif child.lang == "yaml":
schema = directive.kwargs.get('schema')
name = directive.kwargs.get('id')
if directive.name == "tftest-file" and schema:
schema = module / 'schemas' / schema
example = YamlExample(code, module, schema)
yaml_path = directive.kwargs['path']
ids.append(f'yaml:{path}:{last_header}:{yaml_path}:{index}')
examples.append(pytest.param(example))
elif isinstance(child, marko.block.Heading):
last_header = child.children[0].children
index = 0
pytest_marks = [
pytest.mark.xdist_group('serial') for m in marks if m == 'serial'
]
examples.append(pytest.param(example, marks=pytest_marks))
ids.append(example_id)
elif isinstance(example, YamlExample):
examples.append(pytest.param(example))
ids.append(example_id)
metafunc.parametrize(test_group, examples, ids=ids)

71
tests/examples/utils.py
View File

@@ -13,7 +13,10 @@
# limitations under the License.
import collections
import os
import re
from pathlib import Path
import marko
Directive = collections.namedtuple('Directive', 'name args kwargs')
TerraformExample = collections.namedtuple(
@@ -38,3 +41,71 @@ def get_tftest_directive(s):
args.append(arg)
return Directive(name, args, kwargs)
return None
def get_readme_examples(readme_path, fabric_root):
"""Find all code examples tagged for testing in a README file."""
readme_path = readme_path.resolve()
fabric_root = fabric_root.resolve()
doc = marko.parse(readme_path.read_text())
module = readme_path.parent
path = module.relative_to(fabric_root)
files = collections.defaultdict(dict)
fixtures = {}
examples = []
# first pass: collect all examples tagged with tftest-file or tftest-fixture
last_header = None
for child in doc.children:
if isinstance(child, marko.block.FencedCode):
code = child.children[0].children
directive = get_tftest_directive(code)
if directive is None:
continue
if directive.name == 'tftest-file':
name, filepath = directive.kwargs['id'], directive.kwargs['path']
files[last_header][name] = File(filepath, code)
if directive.name == 'tftest-fixture':
name = directive.kwargs['id']
fixtures[name] = code
elif isinstance(child, marko.block.Heading):
last_header = child.children[0].children
# second pass: collect all examples tagged with tftest
last_header = None
index = 0
for child in doc.children:
if isinstance(child, marko.block.FencedCode):
index += 1
code = child.children[0].children
directive = get_tftest_directive(code)
if directive is None:
continue
if child.lang in ('hcl', 'tfvars'):
name = f'{path}:{last_header}'
if index > 1:
name += f' {index}'
example_id = f'terraform:{path}:{last_header}:{index}'
marks = []
if 'serial' in directive.args:
marks.append('serial')
example = TerraformExample(name, code, path, files[last_header],
fixtures, child.lang, directive)
examples.append((example, example_id, marks, last_header, index))
elif child.lang == "yaml":
schema = directive.kwargs.get('schema')
if directive.name == "tftest-file" and schema:
schema = module / 'schemas' / schema
example = YamlExample(code, module, schema)
yaml_path = directive.kwargs['path']
example_id = f'yaml:{path}:{last_header}:{yaml_path}:{index}'
examples.append((example, example_id, [], last_header, index))
elif isinstance(child, marko.block.Heading):
last_header = child.children[0].children
index = 0
return examples

254
tools/generate_plan_summary.py Executable file
View File

@@ -0,0 +1,254 @@
#!/usr/bin/env python3
# Copyright 2026 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
#
# https://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.
# /// script
# requires-python = ">=3.11"
# dependencies = [
# "click",
# "marko",
# "pytest>=7.2.1",
# "PyYAML>=6.0",
# "tftest>=1.8.1",
# ]
# ///
"""Generate plan summary for README examples or tftest.yaml tests.
This script unifies the functionality of generating inventory files from
either README code blocks or tftest.yaml test specifications.
"""
import collections
import datetime
import glob
import os
import re
import shutil
import sys
import tempfile
import click
import marko
import yaml
from pathlib import Path
BASEDIR = Path(__file__).parents[1]
sys.path.append(str(BASEDIR / 'tests'))
try:
import fixtures
from examples.utils import get_readme_examples, get_tftest_directive
except ImportError as e:
print(f"Error importing fixtures or utils: {e}")
sys.exit(1)
FILTERED_ATTRIBUTES = [
'source_md5hash',
]
HEADER = "".join(open(__file__).readlines()[2:15])
current_year = datetime.date.today().year
HEADER = re.sub(r"Copyright \d{4}", f"Copyright {current_year}", HEADER)
def output_summary(summary, inventory_path, save):
values = fixtures.filter_plan_values(summary.values, FILTERED_ATTRIBUTES)
outputs = {
k: v.get('value', '__missing__') for k, v in summary.outputs.items()
}
if save:
if not inventory_path:
print("Error: Cannot determine inventory path for saving.")
sys.exit(1)
inventory_path.parent.mkdir(parents=True, exist_ok=True)
with open(inventory_path, 'w') as f:
f.write(HEADER)
f.write('\n')
yaml.dump({'values': values}, f)
f.write('\n')
yaml.dump({'counts': summary.counts}, f)
f.write('\n')
yaml.dump({'outputs': outputs}, f)
print(f"Inventory saved to {inventory_path}")
else:
print(yaml.dump({'values': values}))
print()
print(yaml.dump({'counts': summary.counts}))
print()
print(yaml.dump({'outputs': outputs}))
def prepare_files(test_path, files, fixtures_dict, requested_files,
requested_fixtures):
if requested_files:
for f in requested_files.split(','):
if f in files:
destination = test_path / files[f].path
destination.parent.mkdir(parents=True, exist_ok=True)
destination.write_text(files[f].content)
if requested_fixtures:
for f in requested_fixtures.split(','):
if f.startswith('fixtures/'):
source = BASEDIR / 'tests' / f
destination = test_path / source.name
if not destination.exists():
destination.symlink_to(source)
elif f in fixtures_dict:
destination = test_path / f'{f}.tf'
destination.write_text(fixtures_dict[f])
def handle_readme(readme_path, target, index, save):
examples = get_readme_examples(readme_path, BASEDIR)
header = target
if not header:
headers = sorted(
set(exp_header
for exp, example_id, marks, exp_header, exp_index in examples
if exp_header))
if not headers:
print(f"No tests found in {readme_path}")
sys.exit(0)
print("Available headers with tests:")
for i, h in enumerate(headers, 1):
print(f" {i}. {h}")
choice = click.prompt("Select a header by number", type=int)
if 1 <= choice <= len(headers):
header = headers[choice - 1]
else:
print("Invalid selection")
sys.exit(1)
target_example = None
for exp, example_id, marks, exp_header, exp_index in examples:
if exp_header == header and exp_index == index:
target_example = exp
break
if not target_example:
print(f"Test not found for header '{header}' and index {index}")
sys.exit(1)
directive = target_example.directive
module_path = readme_path.parent
inventory_path = None
if save:
inventory_name = directive.kwargs.get('inventory')
if not inventory_name:
print("Error: No inventory file specified in the # tftest directive.")
print("Please add `inventory=filename.yaml` to the directive first.")
sys.exit(1)
module_str = str(target_example.module).replace('-', '_')
inventory_path = (BASEDIR / 'tests' / module_str / 'examples' /
inventory_name)
with tempfile.TemporaryDirectory(prefix='tftest-') as tmp_path:
tmp_path = Path(tmp_path)
if target_example.type == 'hcl':
(tmp_path / 'fabric').symlink_to(BASEDIR)
(tmp_path / 'variables.tf').symlink_to(BASEDIR / 'tests' / 'examples' /
'variables.tf')
(tmp_path / 'main.tf').write_text(target_example.code)
assets_path = module_path / 'assets'
if assets_path.exists():
(tmp_path / 'assets').symlink_to(assets_path)
prepare_files(tmp_path, target_example.files, target_example.fixtures,
directive.kwargs.get('files'),
directive.kwargs.get('fixtures'))
summary = fixtures.plan_summary(tmp_path, Path(), [])
elif target_example.type == 'tfvars':
(tmp_path / 'terraform.auto.tfvars').write_text(target_example.code)
shutil.copytree(module_path, tmp_path, dirs_exist_ok=True)
summary = fixtures.plan_summary(tmp_path, Path(),
[tmp_path / 'terraform.auto.tfvars'])
output_summary(summary, inventory_path, save)
def handle_tftest(test_file, target, save):
test_base_dir = Path(test_file).parent
with open(test_file) as f:
raw = yaml.safe_load(f)
module = raw.pop('module')
test_name = target
if not test_name:
tests = sorted(raw.get('tests', {}).keys())
if not tests:
print(f"No tests found in {test_file}")
sys.exit(0)
print("Available tests:")
for i, t in enumerate(tests, 1):
print(f" {i}. {t}")
choice = click.prompt("Select a test by number", type=int)
if 1 <= choice <= len(tests):
test_name = tests[choice - 1]
else:
print("Invalid selection")
sys.exit(1)
common = raw.pop('common_tfvars', [])
spec = raw.get('tests', {})[test_name] or {}
extra_dirs = spec.get('extra_dirs', [])
extra_files = spec.get('extra_files', [])
tf_var_files = common + [f'{test_name}.tfvars'] + spec.get('tfvars', [])
module_path = BASEDIR / module
summary = fixtures.plan_summary(module_path, test_base_dir, tf_var_files,
extra_files=extra_files,
extra_dirs=extra_dirs)
inventory_path = test_base_dir / f'{test_name}.yaml' if save else None
output_summary(summary, inventory_path, save)
@click.command()
@click.argument('file_path', type=click.Path(exists=True), nargs=1)
@click.argument('target', required=False)
@click.option('--index', default=1,
help='Index of the test under the header (README only)')
@click.option('--save', is_flag=True,
help='Automatically save inventory to the right location')
def main(file_path, target, index, save):
"""Generate plan summary for a README example or a tftest.yaml test.
FILE_PATH: Path to README.md or tftest.yaml.
TARGET: Header name (for README) or test name (for tftest.yaml).
"""
file_path = Path(file_path)
if file_path.suffix == '.md' or file_path.name == 'README.md':
handle_readme(file_path, target, index, save)
elif file_path.suffix in ('.yaml', '.yml') or file_path.name == 'tftest.yaml':
handle_tftest(file_path, target, save)
else:
print(f"Unsupported file type: {file_path.suffix}")
print("Please provide a README.md or a tftest.yaml file.")
sys.exit(1)
if __name__ == '__main__':
main()

69
tools/tftest_plan_summary.py
View File

@@ -1,69 +0,0 @@
#!/usr/bin/env python3
# Copyright 2025 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
#
# https://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 click
import sys
import yaml
from pathlib import Path
try:
import fixtures
except ImportError:
BASEDIR = Path(__file__).parents[1]
sys.path.append(str(BASEDIR / 'tests'))
import fixtures
FILTERED_ATTRIBUTES = [
'source_md5hash',
]
@click.command()
@click.argument('test_file', type=click.Path(), nargs=1)
@click.argument('test_name', nargs=1)
def main(test_file, test_name):
test_base_dir = Path(test_file).parent
try:
with open(test_file) as f:
raw = yaml.safe_load(f)
module = raw.pop('module')
except (IOError, OSError, yaml.YAMLError) as e:
raise Exception(f'cannot read test spec {test_file}: {e}')
except KeyError as e:
raise Exception(f'`module` key not found in {test_file}: {e}')
common = raw.pop('common_tfvars', [])
spec = raw.get('tests', {})[test_name] or {}
extra_dirs = spec.get('extra_dirs', [])
extra_files = spec.get('extra_files', [])
tf_var_files = common + [f'{test_name}.tfvars'] + spec.get('tfvars', [])
module_path = BASEDIR / module
summary = fixtures.plan_summary(module_path, test_base_dir, tf_var_files,
extra_files=extra_files,
extra_dirs=extra_dirs)
values = fixtures.filter_plan_values(summary.values, FILTERED_ATTRIBUTES)
print(yaml.dump({'values': values}))
print(yaml.dump({'counts': summary.counts}))
outputs = {
k: v.get('value', '__missing__') for k, v in summary.outputs.items()
}
print(yaml.dump({'outputs': outputs}))
if __name__ == '__main__':
main()