Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
once per resource with identical text.
- `core`: "Could not find input value" now names the provider arguments that produced no value.

### Fixed
- **Verdict change.** A resource skipped through `error_tolerance` no longer overwrites the
verdict of the resources evaluated before it. An evaluator now fails if any resource fails,
passes if none fail and at least one was evaluated, and is skipped only when every resource
was tolerated away. Previously a skip reset the verdict to skipped, so a violating resource
followed by a destroyed one (severity 0, tolerated at every `error_tolerance`) disappeared
from `eval_expression` and the policy reported no verdict; under `!id` it passed. The result
depended on the order of `resource_changes`. Plans that mix compliant and destroyed
resources now pass instead of exiting 1, and plans that mix violating and destroyed
resources now exit 3 instead of 1. (#293)

### Notes
- The local evaluation surface is untouched. `ui` is dispatched before the flat parser, like
`platform`, so `--json` output remains byte-identical to the golden file.
Expand Down
23 changes: 14 additions & 9 deletions src/tirith/core/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -97,15 +97,18 @@

evaluator_instance = evaluator_class()
evaluation_results = []
has_evaluation_passed = True

# If there are no evaluator inputs, it means the provider didn't find any resources
# In this case, the evaluation should fail
if not evaluator_inputs:
has_evaluation_passed = False
evaluation_results = [{"passed": False, "message": _no_input_value_message(provider_inputs)}]
else:
# Track if we've had at least one valid evaluation (not skipped)
# Roll-up across resources: any failure fails the evaluator; otherwise it passes if at least
# one resource was actually evaluated; only when every resource was tolerated away is the
# verdict None (skipped). Tracked as two flags and decided once at the end, so a skip can
# never overwrite a failure or a pass that came before it (issue #293).
has_failure = False
has_valid_evaluation = False

for evaluator_input in evaluator_inputs:
Expand All @@ -117,25 +120,24 @@
# reads as a genuine violation.
if evaluator_input.get("err") and not isinstance(evaluator_input["value"], ProviderError):
evaluation_results.append({"passed": False, "message": evaluator_input["err"]})
has_evaluation_passed = False
has_failure = True
continue

if isinstance(evaluator_input["value"], ProviderError) and evaluator_input.get("err", None):
severity_value = evaluator_input["value"].severity_value
context = evaluator_input.get("context")
err_result = dict(message=format_context_prefix(context) + evaluator_input["err"])

Check warning on line 129 in src/tirith/core/core.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Replace this constructor call with a literal.

See more on https://sonarcloud.io/project/issues?id=StackGuardian_policy-framework&issues=AaBsLlpXqippGUiyMcbk&open=AaBsLlpXqippGUiyMcbk&pullRequest=365
if context:
err_result["context"] = context

if severity_value > evaluator_error_tolerance:
err_result.update(dict(passed=False))

Check warning on line 134 in src/tirith/core/core.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Replace this constructor call with a literal.

See more on https://sonarcloud.io/project/issues?id=StackGuardian_policy-framework&issues=AaBsLlpXqippGUiyMcbl&open=AaBsLlpXqippGUiyMcbl&pullRequest=365
evaluation_results.append(err_result)
has_evaluation_passed = False
has_failure = True
continue
# Mark as skipped evaluation
# Within tolerance: this resource is skipped and does not touch the verdict
err_result.update(dict(passed=None))

Check warning on line 139 in src/tirith/core/core.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Replace this constructor call with a literal.

See more on https://sonarcloud.io/project/issues?id=StackGuardian_policy-framework&issues=AaBsLlpXqippGUiyMcbm&open=AaBsLlpXqippGUiyMcbm&pullRequest=365
evaluation_results.append(err_result)
has_evaluation_passed = None
continue

evaluation_result = evaluator_instance.evaluate(evaluator_input["value"], evaluator_data)
Expand All @@ -150,10 +152,13 @@
has_valid_evaluation = True

if not evaluation_result["passed"]:
has_evaluation_passed = False
has_failure = True

# If all evaluations were skipped, we need to make sure the overall result is 'None'
if not has_valid_evaluation and has_evaluation_passed is None:
if has_failure:
has_evaluation_passed = False
elif has_valid_evaluation:
has_evaluation_passed = True
else:
has_evaluation_passed = None

result["result"] = evaluation_results
Expand All @@ -161,7 +166,7 @@
return result


def generate_compiled_code_without_none_and_variables(eval_str: str) -> Tuple[Optional[CodeType], List[str]]:

Check failure on line 169 in src/tirith/core/core.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this function to reduce its Cognitive Complexity from 18 to the 15 allowed.

See more on https://sonarcloud.io/project/issues?id=StackGuardian_policy-framework&issues=AaBsLlpXqippGUiyMcbn&open=AaBsLlpXqippGUiyMcbn&pullRequest=365
# To make sure that the AST tree loop doesn't run forever
MAX_TRIES = 2000

Expand Down Expand Up @@ -263,7 +268,7 @@
for key in eval_id_values:
regex_string = "\\b" + key + "\\b"
eval_string = re.sub(regex_string, str(eval_id_values[key]), eval_string)
# eval_string = eval_string.replace(key, str(eval_id_values[key]["passed"]))

Check warning on line 271 in src/tirith/core/core.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Remove this commented out code.

See more on https://sonarcloud.io/project/issues?id=StackGuardian_policy-framework&issues=AaBsLlpXqippGUiyMcbo&open=AaBsLlpXqippGUiyMcbo&pullRequest=365
# print (eval_string)

# TODO: shall we use and, or and not instead of symbols?
Expand Down Expand Up @@ -309,7 +314,7 @@
# TODO: validate policy_data against schema

with open(input_path) as f:
if input_path.endswith(".yaml") or input_path.endswith(".yml"):

Check warning on line 317 in src/tirith/core/core.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Replace chained "endswith" calls with a single call using a tuple argument.

See more on https://sonarcloud.io/project/issues?id=StackGuardian_policy-framework&issues=AaBsLlpXqippGUiyMcbp&open=AaBsLlpXqippGUiyMcbp&pullRequest=365
input_data = list(yaml.safe_load_all(f))
if len(input_data) == 1:
input_data = input_data[0]
Expand Down
72 changes: 72 additions & 0 deletions tests/core/test_core.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
# Maintain all core related tests here
from pytest import mark

Check warning on line 2 in tests/core/test_core.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Import "pytest" as a module.

See more on https://sonarcloud.io/project/issues?id=StackGuardian_policy-framework&issues=AaBsLlxtqippGUiyMcbq&open=AaBsLlxtqippGUiyMcbq&pullRequest=365
import pytest

from tirith.core.core import final_evaluator, generate_evaluator_result
Expand Down Expand Up @@ -215,3 +215,75 @@

assert result["passed"] is False, "a malformed provider call must not be skipped"
assert result["result"][0]["passed"] is False


def _skip(msg="tolerated"):
return {"value": ProviderError(severity_value=0), "err": msg}


def _rollup(inputs, error_tolerance=0):
"""Run generate_evaluator_result over `inputs` with MockEvaluator and return the verdict."""
evaluator_obj = {
"id": "test_evaluator",
"provider_args": {"operation_type": "attribute", "key": "value"},
"condition": {"type": "Equals", "value": "expected_value", "error_tolerance": error_tolerance},
}
with patch("tirith.core.core.get_evaluator_inputs_from_provider_inputs", return_value=inputs):
with patch("tirith.core.core.EVALUATORS_DICT", {"Equals": MockEvaluator}):
return generate_evaluator_result(evaluator_obj, {}, "test_provider")


@mark.passing
@pytest.mark.parametrize(
"inputs",
[
[{"value": "resource2"}, _skip()], # violation first, then a tolerated skip (issue #293)
[_skip(), {"value": "resource2"}], # the same two resources, swapped
],
ids=["fail-then-skip", "skip-then-fail"],
)
def test_generate_evaluator_result_skip_never_erases_a_failure(inputs):
"""A tolerated skip must not turn a failing evaluator into a skipped one, whatever the order.

Regression for #293: a destroyed resource (severity 0, tolerated at every error_tolerance)
after a violating resource used to reset the verdict to None, and the None id was then removed
from eval_expression, so the violation vanished.
"""
result = _rollup(inputs)
assert result["passed"] is False
assert sorted((r["passed"] for r in result["result"]), key=str) == [False, None]


@mark.passing
@pytest.mark.parametrize(
"inputs",
[
[{"value": "resource1"}, _skip()],
[_skip(), {"value": "resource1"}],
[{"value": "resource1"}, _skip(), {"value": "resource1"}],
],
ids=["pass-then-skip", "skip-then-pass", "pass-skip-pass"],
)
def test_generate_evaluator_result_skip_never_erases_a_pass(inputs):
"""A skipped resource alongside passing ones leaves a passing verdict, not a skipped one.

Before #293 was fixed, [PASS, skip] reported "Passed: 0 Failed: 0 Skipped: 1" and exited 1 for a
plan that had a compliant resource next to a destroyed one.
"""
result = _rollup(inputs)
assert result["passed"] is True


@mark.passing
def test_generate_evaluator_result_all_skipped_is_none():
"""Only when every resource was tolerated away is the evaluator skipped as a whole."""
result = _rollup([_skip(), _skip()])
assert result["passed"] is None
assert all(r["passed"] is None for r in result["result"])


@mark.passing
def test_generate_evaluator_result_skip_never_erases_a_bare_provider_error():
"""A malformed provider call fails hard even when a tolerated skip follows it."""
result = _rollup([{"value": None, "err": "attribute_to_get is not supported"}, _skip()])
assert result["passed"] is False
Loading