From 2b3b3bbbc23ef4c43b59b4f29a5f7b2abe150f18 Mon Sep 17 00:00:00 2001 From: Rafid Aslam Date: Fri, 28 Aug 2026 15:52:12 +0700 Subject: [PATCH 1/2] feat(terraform_plan): name the resource and attribute in result messages A result message only stated the comparison it made, so a check run across a plan with 30 buckets printed 30 identical lines with no way to tell which resource had failed. The detail was already in the result document - the provider attaches the whole resource_change as `meta` on every output - but nothing between the provider and the pretty printer surfaced it, so reaching it meant piping --json into another tool. Providers may now return an optional `context` alongside value/meta/err. The core renders it into the front of the message and keeps it as structured fields in the result document, so a failure reads: [aws_s3_bucket.example (create)] acl: `"public-read"` is not equal to `"private"` Only terraform_plan populates it, so every other provider's messages are unchanged. Two related improvements come with it: a wildcard attribute reports the index it resolved to (ebs_block_device.0.tags.application_acronym), which tells apart results that used to read identically, and an "attribute is not found" error names the resource it is about instead of repeating the same text once per resource. --- CHANGELOG.md | 18 ++ README.md | 18 +- src/tirith/core/core.py | 55 ++++- src/tirith/providers/common.py | 66 +++++- .../providers/terraform_plan/handler.py | 197 ++++++++++++++++-- tests/core/test_core.py | 2 +- tests/example-output.jsonc | 9 + tests/providers/json/test_get_value.py | 6 + tests/providers/kubernetes/test_attribute.py | 5 + .../terraform_plan/test_message_context.py | 196 +++++++++++++++++ tests/providers/test_common.py | 41 +++- 11 files changed, 588 insertions(+), 25 deletions(-) create mode 100644 tests/providers/terraform_plan/test_message_context.py diff --git a/CHANGELOG.md b/CHANGELOG.md index b2ed0dfd..c4f2cf00 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,24 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Added +- `core`: Provider results can now carry a `context` object saying where the evaluated value came + from. It is rendered into the front of the result `message` and kept as structured fields in the + result document +- `terraform_plan`: Result messages now name the resource address, its planned action and the + attribute being evaluated, e.g. ``[aws_s3_bucket.example (create)] acl: `"public-read"` is not + equal to `"private"` `` instead of just ``` `"public-read"` is not equal to `"private"` ``` + +### Changed +- `terraform_plan`: A wildcard attribute now reports the index it resolved to + (`ebs_block_device.0.tags.application_acronym`), so results from the same resource can be told + apart +- `terraform_plan`: An "attribute is not found" error now names the resource it is about, instead + of repeating the same text once per resource +- `core`: "Could not find input value" now names the provider arguments that produced no value + ## [1.0.5] - 2025-11-19 ### Fixed diff --git a/README.md b/README.md index 786a16e5..a116fe0b 100644 --- a/README.md +++ b/README.md @@ -369,7 +369,14 @@ JSON Output: "result": [ { "passed": true, - "message": "default is equal to default", + "message": "[aws_vpc.this[0] (create)] instance_tenancy: `\"default\"` is equal to `\"default\"`", + "context": { + "operation_type": "attribute", + "resource_type": "aws_vpc", + "resource_address": "aws_vpc.this[0]", + "action": "create", + "attribute": "instance_tenancy" + }, "meta": { "address": "aws_vpc.this[0]", "mode": "managed", @@ -416,7 +423,14 @@ JSON Output: }, { "passed": true, - "message": "default is equal to default", + "message": "[aws_vpc.this[0] (create)] instance_tenancy: `\"default\"` is equal to `\"default\"`", + "context": { + "operation_type": "attribute", + "resource_type": "aws_vpc", + "resource_address": "aws_vpc.this[0]", + "action": "create", + "attribute": "instance_tenancy" + }, "meta": { "address": "aws_vpc.this[0]", "mode": "managed", diff --git a/src/tirith/core/core.py b/src/tirith/core/core.py index 27c60646..fe147126 100644 --- a/src/tirith/core/core.py +++ b/src/tirith/core/core.py @@ -7,7 +7,7 @@ from typing import Any, Dict, List, Tuple, Optional -from tirith.providers.common import ProviderError +from tirith.providers.common import ProviderError, format_context_prefix from ..providers import PROVIDERS_DICT from .evaluators import EVALUATORS_DICT from .policy_parameterization import get_policy_with_vars_replaced @@ -15,6 +15,18 @@ logger = logging.getLogger(__name__) +# Provider arguments named in the "could not find input value" message, in the order they +# are rendered. Providers use different argument names, so only the present ones are used. +_NO_INPUT_VALUE_DESCRIBED_ARGS = ( + "operation_type", + "terraform_resource_type", + "terraform_resource_attribute", + "terraform_provider_full_name", + "kubernetes_kind", + "attribute_path", + "attribute", +) + def get_evaluator_inputs_from_provider_inputs(provider_inputs, provider_module, input_data): # TODO: Get the inputs from given providers @@ -26,6 +38,34 @@ def get_evaluator_inputs_from_provider_inputs(provider_inputs, provider_module, return provider_func(provider_inputs, input_data) +def _no_input_value_message(provider_inputs: Optional[Dict]) -> str: + """ + Build the message used when a provider returns no inputs at all. + + The bare "Could not find input value" says nothing about what was looked for, so name + the provider arguments that produced no value whenever they are available. + + :param provider_inputs: The `provider_args` of the evaluator + :type provider_inputs: Optional[Dict] + + :returns: The message to report against the failed evaluation + :rtype: str + """ + BASE_MESSAGE = "Could not find input value" + + if not provider_inputs: + return BASE_MESSAGE + + described_args = ", ".join( + f"{key}: '{provider_inputs[key]}'" for key in _NO_INPUT_VALUE_DESCRIBED_ARGS if provider_inputs.get(key) + ) + + if not described_args: + return BASE_MESSAGE + + return f"{BASE_MESSAGE} for {described_args}" + + def generate_evaluator_result(evaluator_obj, input_data, provider_module): DEFAULT_ERROR_TOLERANCE = 0 @@ -60,7 +100,7 @@ def generate_evaluator_result(evaluator_obj, input_data, provider_module): # In this case, the evaluation should fail if not evaluator_inputs: has_evaluation_passed = False - evaluation_results = [{"passed": False, "message": "Could not find input value"}] + evaluation_results = [{"passed": False, "message": _no_input_value_message(provider_inputs)}] else: # Track if we've had at least one valid evaluation (not skipped) has_valid_evaluation = False @@ -68,7 +108,10 @@ def generate_evaluator_result(evaluator_obj, input_data, provider_module): for evaluator_input in evaluator_inputs: if isinstance(evaluator_input["value"], ProviderError) and evaluator_input.get("err", None): severity_value = evaluator_input["value"].severity_value - err_result = dict(message=evaluator_input["err"]) + context = evaluator_input.get("context") + err_result = dict(message=format_context_prefix(context) + evaluator_input["err"]) + if context: + err_result["context"] = context if severity_value > evaluator_error_tolerance: err_result.update(dict(passed=False)) @@ -82,6 +125,12 @@ def generate_evaluator_result(evaluator_obj, input_data, provider_module): continue evaluation_result = evaluator_instance.evaluate(evaluator_input["value"], evaluator_data) + context = evaluator_input.get("context") + if context: + # Say which resource and attribute the evaluated value came from, both in the + # message and as structured fields for whoever reads the result document + evaluation_result["message"] = format_context_prefix(context) + evaluation_result["message"] + evaluation_result["context"] = context evaluation_result["meta"] = evaluator_input.get("meta") evaluation_results.append(evaluation_result) has_valid_evaluation = True diff --git a/src/tirith/providers/common.py b/src/tirith/providers/common.py index 5c0c54f3..2c71e3fb 100644 --- a/src/tirith/providers/common.py +++ b/src/tirith/providers/common.py @@ -1,6 +1,6 @@ import pydash -from typing import Dict, Any +from typing import Dict, Any, Optional def create_result_dict(value=None, meta=None, err=None) -> Dict: @@ -167,3 +167,67 @@ class ProviderError: def __init__(self, severity_value: int) -> None: self.severity_value = severity_value + + +def format_context_prefix(context: Optional[Dict]) -> str: + """ + Render a provider result ``context`` dictionary as a message prefix. + + Providers may attach a ``context`` dictionary to each of their outputs to describe where + the evaluated value came from. The core prepends the rendered prefix to the evaluator + message, so that a result reads ``[aws_s3_bucket.example (create)] acl: `"public-read"` is + not equal to `"private"``` instead of just ```"public-read"` is not equal to `"private"```. + + Recognised keys: + + ``resource_address`` + Address of the resource the value belongs to. Rendered as the bracketed subject. + ``label`` + Fallback subject when there is no single resource address (for example a resource + count). Only used when ``resource_address`` is absent. + ``action`` + Planned action(s) for the resource. Rendered next to the subject. + ``attribute`` + Name of the attribute being evaluated. + + Any other key is ignored here but is still carried into the result document, so providers + can supply structured detail without it showing up in the message. + + :param context: The context dictionary attached by the provider, or None + :type context: Optional[Dict] + + :returns: The prefix to prepend to a message, or an empty string when there is no context + :rtype: str + + **Examples:** + + >>> format_context_prefix({"resource_address": "aws_vpc.main", "action": "create", "attribute": "cidr_block"}) + '[aws_vpc.main (create)] cidr_block: ' + >>> format_context_prefix({"resource_address": "aws_vpc.main", "attribute": "action"}) + '[aws_vpc.main] action: ' + >>> format_context_prefix({"label": "aws_vpc", "attribute": "count"}) + '[aws_vpc] count: ' + >>> format_context_prefix({"attribute": "terraform_version"}) + 'terraform_version: ' + >>> format_context_prefix({"resource_address": "aws_vpc.main", "action": "create"}) + '[aws_vpc.main (create)] ' + >>> format_context_prefix(None) + '' + """ + if not context: + return "" + + subject = context.get("resource_address") or context.get("label") + action = context.get("action") + attribute = context.get("attribute") + + subject_prefix = "" + if subject: + subject_prefix = "[{} ({})] ".format(subject, action) if action else "[{}] ".format(subject) + + if not attribute: + # Without an attribute to name, the message that follows reads as a sentence of its + # own, so the subject is left as a bare lead-in rather than being followed by a colon + return subject_prefix + + return "{}{}: ".format(subject_prefix, attribute) diff --git a/src/tirith/providers/terraform_plan/handler.py b/src/tirith/providers/terraform_plan/handler.py index bbcec745..9d556e5f 100644 --- a/src/tirith/providers/terraform_plan/handler.py +++ b/src/tirith/providers/terraform_plan/handler.py @@ -8,7 +8,7 @@ # input->(list ["a.b","c", "d"],value of resource) # returns->[any, any, any] -from typing import Iterable, Tuple +from typing import Any, Dict, Iterable, List, Optional, Tuple import pydash from ..common import ProviderError @@ -18,12 +18,45 @@ class PydashPathNotFound: pass +def _join_attribute_path(*parts) -> str: + """ + Join the parts of an attribute path, skipping the empty ones. + + :param parts: The path segments, which may be strings or list indices + :return: The dotted attribute path + """ + return ".".join(str(part) for part in parts if part != "" and part is not None) + + def _wrapper_get_exp_attribute(attribute, input_resource_change_attrs): + return [value for _, value in _wrapper_get_exp_attribute_with_paths(attribute, input_resource_change_attrs)] + + +def _wrapper_get_exp_attribute_with_paths(attribute, input_resource_change_attrs): splitted_attribute = attribute.split(".*.") - return _get_exp_attribute(splitted_attribute, input_resource_change_attrs) + return _get_exp_attribute_with_paths(splitted_attribute, input_resource_change_attrs) def _get_exp_attribute(split_expressions, input_data): + return [value for _, value in _get_exp_attribute_with_paths(split_expressions, input_data)] + + +def _get_exp_attribute_with_paths(split_expressions, input_data, path_prefix: str = "") -> List[Tuple[str, Any]]: + """ + Resolve an attribute expression, reporting the concrete path each value was found at. + + A wildcard expression such as ``ingress.*.cidr_blocks`` yields one value per list item, so + the expression alone cannot tell the reader which item a result belongs to. Alongside every + value this returns the path that produced it, with the wildcards replaced by the list index + they matched (``ingress.0.cidr_blocks``, ``ingress.1.cidr_blocks``, ...). When a list item + does not have the attribute at all, the unresolved tail of the expression is kept in the + path so that the reader can still see what was looked for. + + :param split_expressions: The attribute expression split on ``.*.`` + :param input_data: The data to resolve the expression against + :param path_prefix: The already resolved part of the path, used when recursing + :return: A list of ``(resolved_path, value)`` tuples + """ # split_expressions=expression.split('*') final_data = [] for i, expression in enumerate(split_expressions): @@ -31,31 +64,73 @@ def _get_exp_attribute(split_expressions, input_data): if isinstance(intermediate_val, list) and i < len(split_expressions) - 1: # For each item in the list, recursively get attributes # Track if at least one item had the attribute - for val in intermediate_val: - final_attributes = _get_exp_attribute(split_expressions[i + 1 :], val) + remaining_expressions = split_expressions[i + 1 :] + for index, val in enumerate(intermediate_val): + item_path = _join_attribute_path(path_prefix, expression, index) + final_attributes = _get_exp_attribute_with_paths(remaining_expressions, val, item_path) if final_attributes: for final_attribute in final_attributes: final_data.append(final_attribute) else: # If no attributes found for this list item, append None # This ensures list items without the attribute are still evaluated - final_data.append(None) + unresolved_path = _join_attribute_path(item_path, ".*.".join(remaining_expressions)) + final_data.append((unresolved_path, None)) # We've already processed all remaining expressions for this list # so we can return early return final_data elif i == len(split_expressions) - 1 and intermediate_val is not PydashPathNotFound: - final_data.append(intermediate_val) + final_data.append((_join_attribute_path(path_prefix, expression), intermediate_val)) elif expression.endswith(".*"): intermediate_exp = expression.split(".*") intermediate_data = pydash.get(input_data, intermediate_exp[0], default=PydashPathNotFound) if intermediate_data is not PydashPathNotFound and isinstance(intermediate_data, list): # For each item in the list, check if it has attributes or append None - for val in intermediate_data: - final_data.append(val) + for index, val in enumerate(intermediate_data): + final_data.append((_join_attribute_path(path_prefix, intermediate_exp[0], index), val)) return final_data +def _build_context(**fields) -> Dict: + """ + Build a provider result context, dropping the fields that carry no value. + + See :func:`tirith.providers.common.format_context_prefix` for how the context is rendered + into the result message. + + :param fields: The context fields + :return: The context dictionary + """ + return {key: value for key, value in fields.items() if value is not None and value != ""} + + +def _resource_context( + operation_type: str, resource_change: Dict, attribute: Optional[str] = None, include_action: bool = True +) -> Dict: + """ + Build the context for a value read off a single resource change. + + :param operation_type: The `operation_type` of the provider args + :param resource_change: The resource change the value was read from + :param attribute: The attribute being evaluated, if any + :param include_action: Whether to report the planned action of the resource. Set this to + False when the evaluated value already is the action. + :return: The context dictionary + """ + action = None + if include_action: + action = "/".join(str(planned) for planned in resource_change.get("change", {}).get("actions") or []) + + return _build_context( + operation_type=operation_type, + resource_type=resource_change.get("type"), + resource_address=resource_change.get("address"), + action=action, + attribute=attribute, + ) + + def provide(provider_inputs, input_data): # """Provides the value of the attribute from the input_data""" outputs = [] @@ -102,15 +177,25 @@ def provide(provider_inputs, input_data): "value": attribute_value, "meta": resource_change, "err": None, + "context": _resource_context(input_type, resource_change, attribute), } ) elif "." in attribute or "*" in attribute: - evaluated_outputs = _wrapper_get_exp_attribute(attribute, input_resource_change_attrs) + evaluated_outputs = _wrapper_get_exp_attribute_with_paths( + attribute, input_resource_change_attrs + ) if evaluated_outputs: is_attribute_found = True local_is_found_attribute = True - for evaluated_output in evaluated_outputs: - outputs.append({"value": evaluated_output, "meta": resource_change, "err": None}) + for resolved_path, evaluated_output in evaluated_outputs: + outputs.append( + { + "value": evaluated_output, + "meta": resource_change, + "err": None, + "context": _resource_context(input_type, resource_change, resolved_path), + } + ) # If we didn't find the attribute in this resource, raise the ProviderError so that the value # still gets evaluated @@ -119,6 +204,7 @@ def provide(provider_inputs, input_data): { "value": ProviderError(severity_value=2), "err": f"attribute: '{attribute}' is not found", + "context": _resource_context(input_type, resource_change), } ) else: @@ -126,6 +212,7 @@ def provide(provider_inputs, input_data): { "value": ProviderError(severity_value=0), "err": f"No Terraform changes found for resource type: '{resource_type}'", + "context": _resource_context(input_type, resource_change), } ) @@ -165,6 +252,10 @@ def provide(provider_inputs, input_data): "value": action, "meta": resource_change, "err": None, + # The evaluated value already is the action, so don't repeat it + "context": _resource_context( + input_type, resource_change, attribute="action", include_action=False + ), } ) if not is_resource_type_found: @@ -197,6 +288,10 @@ def provide(provider_inputs, input_data): "value": count, "meta": resource_meta, "err": None, + # A count belongs to a resource type rather than to a single resource + "context": _build_context( + operation_type=input_type, resource_type=resource_type, label=resource_type, attribute="count" + ), } ) return outputs @@ -267,6 +362,12 @@ def provider_config_operator(input_data: dict, provider_inputs: dict, outputs: l # FIXME: The region might not be in the constant_value, it can be in a variable attribute_value = provider_config_dict.get("expressions", {}).get("region", {}).get("constant_value") + context = _build_context( + operation_type="provider_config", + label=f"provider {terraform_provider_full_name}", + attribute=attribute_to_get, + ) + if attribute_value is None: severity_value = 2 outputs.append( @@ -274,6 +375,7 @@ def provider_config_operator(input_data: dict, provider_inputs: dict, outputs: l "value": ProviderError(severity_value=severity_value), "err": f"`{attribute_to_get}` is not found in the provider_config (severity_value: {severity_value})", "meta": provider_config_dict, + "context": context, } ) return @@ -281,6 +383,7 @@ def provider_config_operator(input_data: dict, provider_inputs: dict, outputs: l { "value": attribute_value, "meta": provider_config_dict, + "context": context, } ) @@ -303,7 +406,13 @@ def terraform_version_operator(input_data: dict, provider_inputs: dict, outputs: :param provider_inputs: The provider inputs :param outputs: The outputs """ - outputs.append({"value": input_data.get("terraform_version"), "meta": input_data}) + outputs.append( + { + "value": input_data.get("terraform_version"), + "meta": input_data, + "context": _build_context(operation_type="terraform_version", attribute="terraform_version"), + } + ) def direct_dependencies_operator(input_data: dict, provider_inputs: dict, outputs: list): @@ -327,7 +436,18 @@ def direct_dependencies_operator(input_data: dict, provider_inputs: dict, output continue is_resource_found = True deps_resource_type = {resource_id.split(".")[0] for resource_id in resource.get("depends_on", [])} - outputs.append({"value": list(deps_resource_type), "meta": config_resources}) + outputs.append( + { + "value": list(deps_resource_type), + "meta": config_resources, + "context": _build_context( + operation_type="direct_dependencies", + resource_type=resource_type, + resource_address=resource.get("address"), + attribute="depends_on", + ), + } + ) if not is_resource_found: outputs.append( @@ -393,12 +513,32 @@ def direct_references_operator_referenced_by(input_data: dict, provider_inputs: if reference_address in reference_target_addresses: reference_target_addresses.remove(reference_address) outputs.append( - {"value": True, "meta": {"address": reference_address, "referenced_by": resource_config}} + { + "value": True, + "meta": {"address": reference_address, "referenced_by": resource_config}, + "context": _build_context( + operation_type="direct_references", + resource_type=resource_type, + resource_address=reference_address, + attribute=f"referenced_by {referenced_by}", + ), + } ) # For all of the reference_target_addresses that don't have a reference for reference_target_address in reference_target_addresses: - outputs.append({"value": False, "meta": {"address": reference_target_address, "referenced_by": {}}}) + outputs.append( + { + "value": False, + "meta": {"address": reference_target_address, "referenced_by": {}}, + "context": _build_context( + operation_type="direct_references", + resource_type=resource_type, + resource_address=reference_target_address, + attribute=f"referenced_by {referenced_by}", + ), + } + ) def get_module_resources_by_type_recursive(module: dict, resource_type: str, current_module_path: str = "") -> iter: @@ -484,7 +624,19 @@ def direct_references_operator_references_to(input_data: dict, provider_inputs: return is_all_resource_type_references_to = resource_type_count == reference_count - outputs.append({"value": is_all_resource_type_references_to, "meta": config_resources}) + outputs.append( + { + "value": is_all_resource_type_references_to, + "meta": config_resources, + # The value covers every instance of the resource type, so there is no single address + "context": _build_context( + operation_type="direct_references", + resource_type=resource_type, + label=resource_type, + attribute=f"references_to {references_to_type}", + ), + } + ) def direct_references_operator(input_data: dict, provider_inputs: dict, outputs: list): @@ -536,7 +688,18 @@ def direct_references_operator(input_data: dict, provider_inputs: dict, outputs: # Only get the resource type resource_references.add(reference.split(".")[0]) - outputs.append({"value": list(resource_references), "meta": resource}) + outputs.append( + { + "value": list(resource_references), + "meta": resource, + "context": _build_context( + operation_type="direct_references", + resource_type=resource_type, + resource_address=resource.get("address"), + attribute="references", + ), + } + ) if not is_resource_found: outputs.append( diff --git a/tests/core/test_core.py b/tests/core/test_core.py index 3afdc41e..117c4750 100644 --- a/tests/core/test_core.py +++ b/tests/core/test_core.py @@ -69,7 +69,7 @@ def test_generate_evaluator_result_empty_inputs(): assert result["passed"] is False assert len(result["result"]) == 1 assert result["result"][0]["passed"] is False - assert result["result"][0]["message"] == "Could not find input value" + assert result["result"][0]["message"] == "Could not find input value for operation_type: 'attribute'" @mark.passing diff --git a/tests/example-output.jsonc b/tests/example-output.jsonc index e54c8df4..8fdb26c5 100644 --- a/tests/example-output.jsonc +++ b/tests/example-output.jsonc @@ -14,6 +14,15 @@ { // comes from provider and evaluator combination "error": "", "message": "", + // Optional. Set when the provider says where the value came from, + // and rendered into the front of "message" + "context": { + "operation_type": "", + "resource_type": "", + "resource_address": "", + "action": "", + "attribute": "" + }, "meta": {}, "passed": true // ... diff --git a/tests/providers/json/test_get_value.py b/tests/providers/json/test_get_value.py index a702d957..ad154ba5 100644 --- a/tests/providers/json/test_get_value.py +++ b/tests/providers/json/test_get_value.py @@ -15,6 +15,12 @@ def test_get_value(): result = start_policy_evaluation_from_dict(policy, input_data) assert result["final_result"] is True + # This provider does not attach a result context, so its messages carry no prefix + for evaluator in result["evaluators"]: + for item in evaluator["result"]: + assert "context" not in item + assert not item["message"].startswith("[") + def test_get_value_playbook(): """Test get_value with playbook YAML data using wildcard path""" diff --git a/tests/providers/kubernetes/test_attribute.py b/tests/providers/kubernetes/test_attribute.py index ee1d8d4b..f1439b1b 100644 --- a/tests/providers/kubernetes/test_attribute.py +++ b/tests/providers/kubernetes/test_attribute.py @@ -10,3 +10,8 @@ def test_get_value(): result = start_policy_evaluation(policy_path=policy_path, input_path=input_path) assert result["final_result"] is False + + # This provider does not attach a result context, so its messages carry no prefix + for item in result["evaluators"][0]["result"]: + assert "context" not in item + assert item["message"].startswith("Found ") diff --git a/tests/providers/terraform_plan/test_message_context.py b/tests/providers/terraform_plan/test_message_context.py new file mode 100644 index 00000000..66c01231 --- /dev/null +++ b/tests/providers/terraform_plan/test_message_context.py @@ -0,0 +1,196 @@ +""" +Tests for the resource context that the terraform_plan provider attaches to its results. + +Without it a message only states the comparison that was made, which reads the same for every +resource in a plan. These tests pin the resource address, the planned action and the attribute +into the message, and the same detail into the `context` of the result document. +""" + +import json +import os + +from pytest import mark + +from tirith.core.core import start_policy_evaluation_from_dict +from tirith.providers.terraform_plan import handler +from utils import load_terraform_plan_json + + +def load_policy_from_fixtures(json_path): + current_path = os.path.dirname(os.path.abspath(__file__)) + with open(os.path.join(current_path, "fixtures", json_path)) as f: + return json.load(f) + + +def evaluate(input_json, policy_json): + return start_policy_evaluation_from_dict( + load_policy_from_fixtures(policy_json), load_terraform_plan_json(input_json) + ) + + +def messages_of(result, evaluator_index=0): + return [item["message"] for item in result["evaluators"][evaluator_index]["result"]] + + +def contexts_of(result, evaluator_index=0): + return [item.get("context") for item in result["evaluators"][evaluator_index]["result"]] + + +@mark.passing +def test_attribute_message_names_the_resource_the_action_and_the_attribute(): + result = evaluate("input_multiple_resource_tag_check.json", "policy_multiple_resource_tag_check.json") + + assert messages_of(result) == [ + '[aws_s3_bucket.bucket_with_tag (create)] tags.a: `"true"` is not empty', + "[aws_s3_bucket.bucket_without_tag (create)] attribute: 'tags.a' is not found", + ] + + +@mark.passing +def test_attribute_context_carries_the_same_detail_as_the_message(): + result = evaluate("input_multiple_resource_tag_check.json", "policy_multiple_resource_tag_check.json") + + assert contexts_of(result) == [ + { + "operation_type": "attribute", + "resource_type": "aws_s3_bucket", + "resource_address": "aws_s3_bucket.bucket_with_tag", + "action": "create", + "attribute": "tags.a", + }, + { + "operation_type": "attribute", + "resource_type": "aws_s3_bucket", + "resource_address": "aws_s3_bucket.bucket_without_tag", + "action": "create", + }, + ] + + +@mark.passing +def test_wildcard_attribute_reports_the_index_it_resolved_to(): + # Every result of `ebs_block_device.*.tags.application_acronym` used to read identically, + # so there was no way to tell which block device was the one missing its tag + result = evaluate("input_aws_instance_ebs.json", "policy_aws_instance_ebs.json") + + assert messages_of(result, evaluator_index=1) == [ + "[aws_instance.example (create)] ebs_block_device.0.tags.application_acronym: `null` is empty", + '[aws_instance.example (create)] ebs_block_device.1.tags.application_acronym: `"TTO"` is not empty', + '[aws_instance.example (create)] ebs_block_device.2.tags.application_acronym: `"TTO"` is not empty', + ] + + +@mark.passing +def test_wildcard_attribute_keeps_the_unresolved_tail_when_an_item_lacks_the_attribute(): + values_with_paths = handler._wrapper_get_exp_attribute_with_paths( + "a.*.b.c", {"a": [{"b": {"c": "found"}}, {"no_b": True}]} + ) + + assert values_with_paths == [("a.0.b.c", "found"), ("a.1.b.c", None)] + + +@mark.passing +def test_wildcard_traversal_without_paths_is_unchanged(): + # `_wrapper_get_exp_attribute` is what the path-tracking traversal replaced, so it has to + # keep returning the bare values in the same order + input_data = {"a": [{"b": {"c": ["val1", "val3"]}}, {"b": {"c": ["val8", "val4"]}}, {"d": {}}]} + + assert handler._wrapper_get_exp_attribute("a.*.b.c.*", input_data) == ["val1", "val3", "val8", "val4", None] + + +@mark.passing +def test_attribute_not_found_names_each_resource_separately(): + # A `*` resource type reports the missing attribute once per resource, and those messages + # were previously indistinguishable from one another + result = evaluate("input_costcenter_tags.json", "policy_star_restype_should_skip.json") + + assert messages_of(result) == [ + "[aws_instance.web (create)] attribute: 'shouldnt_exist' is not found", + "[aws_s3_bucket.logs (create)] attribute: 'shouldnt_exist' is not found", + "[aws_vpc.main (create)] attribute: 'shouldnt_exist' is not found", + ] + + +@mark.passing +def test_count_and_action_messages(): + result = evaluate("input.json", "policy.json") + evaluators = {evaluator["id"]: evaluator for evaluator in result["evaluators"]} + + # A count belongs to a resource type rather than to any one resource, so it is labelled + # with the type and has no planned action + assert [item["message"] for item in evaluators["check111"]["result"]] == [ + "[aws_vpc] count: `2` is not greater than `10`" + ] + assert evaluators["check111"]["result"][0]["context"] == { + "operation_type": "count", + "resource_type": "aws_vpc", + "label": "aws_vpc", + "attribute": "count", + } + + # The evaluated value already is the action, so the action is not repeated next to the address + assert [item["message"] for item in evaluators["check11"]["result"]] == [ + '[aws_vpc.this[0]] action: Found `"create"` inside `["create", "update"]`', + '[aws_vpc.this[0]] action: Found `"create"` inside `["create", "update"]`', + ] + + +@mark.passing +def test_provider_config_and_terraform_version_messages(): + plan = load_terraform_plan_json("input_instance_deps_s3.json") + + def evaluate_provider_args(provider_args, condition): + policy = { + "meta": {"version": "v1", "required_provider": "stackguardian/terraform_plan"}, + "evaluators": [{"id": "check", "provider_args": provider_args, "condition": condition}], + "eval_expression": "check", + } + return start_policy_evaluation_from_dict(policy, plan) + + result = evaluate_provider_args( + { + "operation_type": "provider_config", + "terraform_provider_full_name": "registry.terraform.io/hashicorp/aws", + "attribute": "region", + }, + {"type": "Equals", "value": "us-east-1"}, + ) + assert messages_of(result) == [ + '[provider registry.terraform.io/hashicorp/aws] region: `"eu-central-1"` is not equal to `"us-east-1"`' + ] + + # There is no resource to name, so the attribute leads the message on its own + result = evaluate_provider_args({"operation_type": "terraform_version"}, {"type": "Equals", "value": "9.9.9"}) + assert messages_of(result) == ['terraform_version: `"1.4.5"` is not equal to `"9.9.9"`'] + + result = evaluate_provider_args( + {"operation_type": "direct_dependencies", "terraform_resource_type": "aws_instance"}, + {"type": "Contains", "value": "aws_kms_key"}, + ) + assert messages_of(result) == [ + '[aws_instance.example_c] depends_on: Failed to find `"aws_kms_key"` inside `["aws_s3_bucket"]`' + ] + + +@mark.passing +def test_provider_argument_errors_stay_uncontextualised(): + # These errors are about the policy rather than about a resource, so there is nothing to name + result = evaluate("input.json", "policy.json") + unsupported = start_policy_evaluation_from_dict( + { + "meta": {"version": "v1", "required_provider": "stackguardian/terraform_plan"}, + "evaluators": [ + { + "id": "check", + "provider_args": {"operation_type": "nope"}, + "condition": {"type": "Equals", "value": 1}, + } + ], + "eval_expression": "check", + }, + load_terraform_plan_json("input.json"), + ) + + assert messages_of(unsupported) == ["operation_type: 'nope' is not supported (severity_value: 99)"] + assert contexts_of(unsupported) == [None] + assert result["evaluators"][0]["result"][0]["context"] is not None diff --git a/tests/providers/test_common.py b/tests/providers/test_common.py index 3c65a1e6..d9236698 100644 --- a/tests/providers/test_common.py +++ b/tests/providers/test_common.py @@ -1,5 +1,5 @@ import pytest -from tirith.providers.common import get_path_value_from_input +from tirith.providers.common import format_context_prefix, get_path_value_from_input # Test data for simple path access @@ -242,3 +242,42 @@ def test_wildcard_primitive_no_remaining_paths(data, path, expected): """Test wildcard applied to primitive value with no remaining paths - covers lines 42-43""" result = get_path_value_from_input(path, data) assert result == expected + + +# Test data for the message prefix rendered out of a provider result context +context_prefix_cases = [ + # A resource attribute: the most common case + ( + {"resource_address": "aws_s3_bucket.example", "action": "create", "attribute": "acl"}, + "[aws_s3_bucket.example (create)] acl: ", + ), + # A replacement names both of its actions, in the order they happen + ( + {"resource_address": "aws_instance.web", "action": "delete/create", "attribute": "instance_type"}, + "[aws_instance.web (delete/create)] instance_type: ", + ), + # The action is left out when the evaluated value already is the action + ({"resource_address": "aws_vpc.main", "attribute": "action"}, "[aws_vpc.main] action: "), + # `label` stands in for the subject when there is no single resource address + ({"label": "aws_vpc", "attribute": "count"}, "[aws_vpc] count: "), + # An address always wins over a label + ( + {"resource_address": "aws_vpc.main", "label": "aws_vpc", "attribute": "cidr_block"}, + "[aws_vpc.main] cidr_block: ", + ), + # No subject at all, just the attribute + ({"attribute": "terraform_version"}, "terraform_version: "), + # No attribute to name, so no colon either + ({"resource_address": "aws_vpc.main", "action": "update"}, "[aws_vpc.main (update)] "), + ({"label": "aws_vpc"}, "[aws_vpc] "), + # Unrecognised keys are carried in the result document but never rendered + ({"operation_type": "attribute", "resource_type": "aws_vpc"}, ""), + # Nothing to render + ({}, ""), + (None, ""), +] + + +@pytest.mark.parametrize("context,expected", context_prefix_cases) +def test_format_context_prefix(context, expected): + assert format_context_prefix(context) == expected From ecfc90df1021dc946623cb2667ea3f99c7692705 Mon Sep 17 00:00:00 2001 From: Rafid Aslam Date: Fri, 28 Aug 2026 16:02:10 +0700 Subject: [PATCH 2/2] test(terraform_plan): build the context policies inline instead of using a fixture fixtures/policy.json has a single `&` in its eval_expression, which is not a valid operator. Two of these tests only needed a count and an action result, so they now build a one-evaluator policy for exactly that rather than depending on a fixture policy that happens to contain one. --- .../terraform_plan/test_message_context.py | 88 +++++++++---------- 1 file changed, 42 insertions(+), 46 deletions(-) diff --git a/tests/providers/terraform_plan/test_message_context.py b/tests/providers/terraform_plan/test_message_context.py index 66c01231..311b92b5 100644 --- a/tests/providers/terraform_plan/test_message_context.py +++ b/tests/providers/terraform_plan/test_message_context.py @@ -28,6 +28,16 @@ def evaluate(input_json, policy_json): ) +def evaluate_provider_args(provider_args, condition, input_json): + """Evaluate a single set of provider args, so a test does not need a policy fixture.""" + policy = { + "meta": {"version": "v1", "required_provider": "stackguardian/terraform_plan"}, + "evaluators": [{"id": "check", "provider_args": provider_args, "condition": condition}], + "eval_expression": "check", + } + return start_policy_evaluation_from_dict(policy, load_terraform_plan_json(input_json)) + + def messages_of(result, evaluator_index=0): return [item["message"] for item in result["evaluators"][evaluator_index]["result"]] @@ -112,24 +122,31 @@ def test_attribute_not_found_names_each_resource_separately(): @mark.passing -def test_count_and_action_messages(): - result = evaluate("input.json", "policy.json") - evaluators = {evaluator["id"]: evaluator for evaluator in result["evaluators"]} - - # A count belongs to a resource type rather than to any one resource, so it is labelled - # with the type and has no planned action - assert [item["message"] for item in evaluators["check111"]["result"]] == [ - "[aws_vpc] count: `2` is not greater than `10`" +def test_count_message_is_labelled_with_the_resource_type(): + # A count belongs to a resource type rather than to any one resource, so it is labelled with + # the type and has no planned action + result = evaluate_provider_args( + {"operation_type": "count", "terraform_resource_type": "aws_vpc"}, + {"type": "GreaterThan", "value": 10}, + "input.json", + ) + + assert messages_of(result) == ["[aws_vpc] count: `2` is not greater than `10`"] + assert contexts_of(result) == [ + {"operation_type": "count", "resource_type": "aws_vpc", "label": "aws_vpc", "attribute": "count"} ] - assert evaluators["check111"]["result"][0]["context"] == { - "operation_type": "count", - "resource_type": "aws_vpc", - "label": "aws_vpc", - "attribute": "count", - } - # The evaluated value already is the action, so the action is not repeated next to the address - assert [item["message"] for item in evaluators["check11"]["result"]] == [ + +@mark.passing +def test_action_message_does_not_repeat_the_action(): + # The evaluated value already is the action, so it is not also shown next to the address + result = evaluate_provider_args( + {"operation_type": "action", "terraform_resource_type": "aws_vpc"}, + {"type": "ContainedIn", "value": ["create", "update"]}, + "input.json", + ) + + assert messages_of(result) == [ '[aws_vpc.this[0]] action: Found `"create"` inside `["create", "update"]`', '[aws_vpc.this[0]] action: Found `"create"` inside `["create", "update"]`', ] @@ -137,16 +154,6 @@ def test_count_and_action_messages(): @mark.passing def test_provider_config_and_terraform_version_messages(): - plan = load_terraform_plan_json("input_instance_deps_s3.json") - - def evaluate_provider_args(provider_args, condition): - policy = { - "meta": {"version": "v1", "required_provider": "stackguardian/terraform_plan"}, - "evaluators": [{"id": "check", "provider_args": provider_args, "condition": condition}], - "eval_expression": "check", - } - return start_policy_evaluation_from_dict(policy, plan) - result = evaluate_provider_args( { "operation_type": "provider_config", @@ -154,18 +161,22 @@ def evaluate_provider_args(provider_args, condition): "attribute": "region", }, {"type": "Equals", "value": "us-east-1"}, + "input_instance_deps_s3.json", ) assert messages_of(result) == [ '[provider registry.terraform.io/hashicorp/aws] region: `"eu-central-1"` is not equal to `"us-east-1"`' ] # There is no resource to name, so the attribute leads the message on its own - result = evaluate_provider_args({"operation_type": "terraform_version"}, {"type": "Equals", "value": "9.9.9"}) + result = evaluate_provider_args( + {"operation_type": "terraform_version"}, {"type": "Equals", "value": "9.9.9"}, "input_instance_deps_s3.json" + ) assert messages_of(result) == ['terraform_version: `"1.4.5"` is not equal to `"9.9.9"`'] result = evaluate_provider_args( {"operation_type": "direct_dependencies", "terraform_resource_type": "aws_instance"}, {"type": "Contains", "value": "aws_kms_key"}, + "input_instance_deps_s3.json", ) assert messages_of(result) == [ '[aws_instance.example_c] depends_on: Failed to find `"aws_kms_key"` inside `["aws_s3_bucket"]`' @@ -174,23 +185,8 @@ def evaluate_provider_args(provider_args, condition): @mark.passing def test_provider_argument_errors_stay_uncontextualised(): - # These errors are about the policy rather than about a resource, so there is nothing to name - result = evaluate("input.json", "policy.json") - unsupported = start_policy_evaluation_from_dict( - { - "meta": {"version": "v1", "required_provider": "stackguardian/terraform_plan"}, - "evaluators": [ - { - "id": "check", - "provider_args": {"operation_type": "nope"}, - "condition": {"type": "Equals", "value": 1}, - } - ], - "eval_expression": "check", - }, - load_terraform_plan_json("input.json"), - ) + # This error is about the policy rather than about a resource, so there is nothing to name + result = evaluate_provider_args({"operation_type": "nope"}, {"type": "Equals", "value": 1}, "input.json") - assert messages_of(unsupported) == ["operation_type: 'nope' is not supported (severity_value: 99)"] - assert contexts_of(unsupported) == [None] - assert result["evaluators"][0]["result"][0]["context"] is not None + assert messages_of(result) == ["operation_type: 'nope' is not supported (severity_value: 99)"] + assert contexts_of(result) == [None]