From abe5dda50bdd8b5a4d1d8ce025d3a76f92a343a2 Mon Sep 17 00:00:00 2001 From: stxkxs Date: Wed, 2 Sep 2026 02:03:23 -0700 Subject: [PATCH 1/7] Prove the gates a positive control cannot reach MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `scripts/tests/controls.py` proves a gate rejects by planting the violation it names and requiring a non-zero exit. It cannot do that for a gate whose input arrives over a network, so `NEEDS_NETWORK_PY` exempts seven. Six of those seven also carried no unit tests, which left 1,107 of 3,072 statements in the gate corpus with no proof of any kind — and not a random sixth of it. The set is `check-policy-admission.py`, which decides whether any addon would be denied at admission on an Enforce cluster; `render-addons.py`, whose `discover()` is the corpus two other gates import rather than re-derive; `check-platform-crs.py`, `validate-dashboards.py`, `check-image-pins.py` and `check-log-volume-budget.py`. An exemption from one kind of proof is not an exemption from proof. ─────────────────────── The assertion, not the number ─────────────────────── `scripts/tests/run.py` gains a check that does not depend on a percentage: every gate `controls.py` exempts must carry unit coverage. The exemption list is read out of `controls.py` at run time rather than copied, because a second copy of an exemption list drifts permissively — the gate added to one and not the other is excused by both. Three ways that check can quietly stop meaning anything, each of which now fails: a control-exempt gate arriving with no tests FAIL 1 gate(s) have neither a positive control nor unit coverage: check-hardcoded-org.py. controls.py exempts them because their input arrives over the network, so a unit test on the half that decides the verdict is the only proof available that they say anything at all the exemption list renamed out from under the reader FAIL controls.py could not be read for its control exemptions (AttributeError: module 'controls' has no attribute 'NEEDS_NETWORK_PY'), so no gate could be checked for having neither kind of proof the exemption list emptied FAIL controls.py exempts no gate from a positive control, so the assertion that every exempt gate carries unit tests holds over an empty set and states nothing The coverage floors move too — combined 12 to 29, the no-unit-coverage ceiling 17 to 11, and seven new per-gate floors — but they are the weaker half and the comment in `run.py` says so. A count loose enough to survive an honest refactor is loose enough to miss a decision going untested. ─────────────────────────── What the tests assert ─────────────────────────── 157 tests across six modules, each over the decision that produces a verdict rather than over parsing or setup. Every one of them was verified by reverting the behaviour it names and watching the suite refuse: scripts/tests/test_policy_admission.py (29) Manifest normalisation, rule-name derivation and the two-sided verdict. try the shorter autogen prefix first FAIL: test_the_cronjob_prefix_is_stripped_whole keep kind-less documents in the resource file ERROR: test_a_kindless_document_is_dropped admit the runtime pod on absence of denials alone FAIL: test_a_pod_no_rule_evaluated_is_not_admitted FAIL: test_a_skip_is_neither_a_pass_nor_a_denial FAIL: test_a_clean_canary_cannot_carry_a_missing_runtime_pod count the runtime pod as a foreign addon FAIL: test_the_runtime_pod_is_not_counted_among_flagged_addons scripts/tests/test_platform_crs.py (29) The four rules the API server enforces and no other gate here checks: required-with-defaulting, pruning, declared types, list identity. let a bool satisfy integer, as Python does FAIL: test_a_bool_does_not_satisfy_integer read `required` without the defaulting exemption FAIL: test_a_missing_required_property_that_declares_a_default_is_admitted check list identity only at the top level FAIL: test_a_set_list_identifies_a_scalar_by_itself FAIL: test_an_absent_key_participates_in_the_identity FAIL: test_three_entries_sharing_an_identity_report_each_repeat scripts/tests/test_dashboards.py (28) The offline extractors: variable references in all four Grafana spellings, the `json: |` block, nested panels, and which datasources count as wired. match only the ${name} variable form FAIL: test_the_bare_form_every_dashboard_here_uses FAIL: test_the_bracket_form FAIL: test_a_builtin_beside_a_real_variable_leaves_the_real_one scan panels flat, without descending into a collapsed row FAIL: test_an_alert_inside_a_collapsed_row_is_found FAIL: test_a_top_level_alert_panel_is_found FAIL: test_an_untitled_alert_panel_is_still_reported glob the datasource directory instead of reading the kustomization FAIL: test_a_datasource_file_absent_from_resources_is_not_wired scripts/tests/test_render_addons.py (29) Unit discovery and the exit-code classification that separates a pin which does not resolve from a registry that could not be reached. call any missing-chart message a finding about this repo FAIL: test_a_message_naming_both_is_read_as_unreachable append the chart name to every OCI repoURL FAIL: test_a_repo_url_ending_in_the_chart_name_is_used_as_is take the addon path from any values file FAIL: test_a_per_environment_file_alone_does_not_supply_the_path scripts/tests/test_image_pins.py (23) Reference classification and both directions of the exemption check. look for the tag colon anywhere in the reference FAIL: test_a_registry_port_is_not_read_as_a_tag stop re-checking exemptions against the render FAIL: test_an_exemption_the_fleet_no_longer_renders_mutably_fails FAIL: test_an_exemption_for_an_image_now_pinned_by_tag_fails ERROR: test_the_recorded_reason_is_quoted_back scripts/tests/test_log_volume_budget.py (19) The relation the gate exists for — the fill alert leads the ingestion cutoff — plus every way the alert can stop supplying its operand. let the alert fire at the cutoff rather than before it FAIL: test_a_warning_at_the_cutoff_leaves_no_window accept a rule keyed on the edge counter FAIL: test_a_rule_querying_the_edge_counter_fails_twice FAIL: test_a_rule_querying_both_still_fails_on_the_edge_counter Bytecode caching is disabled while a mutant is under test. It keys on mtime and size, so a restore and a re-mutation inside the same second can hand the next run the previous mutant's module — which reads as the suite catching a mutant it never saw. ──────────────────────── Three verdicts made reachable ──────────────────────── Three gates computed their verdict inline with the network call that fed it, so the decision could only be exercised through a round trip. Each is extracted whole, with no change to what it decides or to what it prints: scripts/check-log-volume-budget.py `environment_verdict(cfg, warn, rel)` returns the problems found and whether the alert leads a declared cutoff. The second value is what the closing line counts, so an environment that failed the comparison is not also reported as one the comparison covered. scripts/check-image-pins.py `verdict(images, allowed)`, `bare_name(ref)` `bare_name` splits on the last colon only when it sits in the final path segment, so a registry port is not read as a tag. scripts/render-addons.py `registry_answered(err)` true when helm's complaint says the registry replied and refused the pin. A message naming both a missing chart and a broken connection is not one: an unreachable registry cannot testify about what it holds. The summary lines these gates print are unchanged — 55 rendered images across 26 charts, four Loki environments checked against an alert at 0.75. ────────────────────────────── Why not a number ────────────────────────────── Raising the coverage floor alone would have left the same hole. The floor is a ratchet over the whole corpus, so a gate arriving with no tests dilutes it by a percentage point and passes; and the six gates in question were the largest files in the tree, so the number they were absent from was the number they dominated. `.coveragerc` said as much and the figure stayed low anyway. What closes it is a per-gate property with the exemption list as its operand. Co-authored-by: stxkxsbot <275011021+stxkxsbot@users.noreply.github.com> --- .coveragerc | 19 +- scripts/check-image-pins.py | 65 +++-- scripts/check-log-volume-budget.py | 76 ++++-- scripts/render-addons.py | 15 +- scripts/tests/run.py | 133 ++++++++-- scripts/tests/test_dashboards.py | 227 ++++++++++++++++ scripts/tests/test_image_pins.py | 166 ++++++++++++ scripts/tests/test_log_volume_budget.py | 210 +++++++++++++++ scripts/tests/test_platform_crs.py | 275 +++++++++++++++++++ scripts/tests/test_policy_admission.py | 338 ++++++++++++++++++++++++ scripts/tests/test_render_addons.py | 222 ++++++++++++++++ 11 files changed, 1663 insertions(+), 83 deletions(-) create mode 100644 scripts/tests/test_dashboards.py create mode 100644 scripts/tests/test_image_pins.py create mode 100644 scripts/tests/test_log_volume_budget.py create mode 100644 scripts/tests/test_platform_crs.py create mode 100644 scripts/tests/test_policy_admission.py create mode 100644 scripts/tests/test_render_addons.py diff --git a/.coveragerc b/.coveragerc index 506e627..220aacd 100644 --- a/.coveragerc +++ b/.coveragerc @@ -1,13 +1,18 @@ # Coverage over the gate scripts, measured by the unit-test suite in -# scripts/tests. The positive controls in scripts/tests/controls.py exercise MOST +# scripts/tests. The positive controls in scripts/tests/controls.py exercise most # gates end to end as a subprocess, which this figure does not see — so the -# number below is the floor under the UNIT tier specifically. +# number in scripts/tests/run.py is the floor under the UNIT tier specifically, +# and it is not a floor under the correctness of a gate the controls cover. # -# Most, not every. controls.py exempts the gates that reach a chart registry or an -# API, and prints the split on every run. Those gates therefore have neither unit -# coverage nor a behavioural control, and they are the largest ones — so the -# figure below is not a floor under their correctness at all, and reading it as -# one would be reading a number measured over a different population. +# Most, not every: controls.py exempts the gates whose input arrives over the +# network, and prints the split on every run. Those are the largest files in the +# tree, and for them a unit test is the only proof available — so run.py asserts +# that each one carries unit coverage, reading the exemption list out of +# controls.py rather than repeating it. +# +# The two tiers answer different questions and neither substitutes for the other. +# A control proves a gate rejects a supplied violation; a unit test proves the +# gate computes the right verdict on a case the real tree does not contain. [run] source = ./scripts omit = diff --git a/scripts/check-image-pins.py b/scripts/check-image-pins.py index d6f97d7..8599548 100755 --- a/scripts/check-image-pins.py +++ b/scripts/check-image-pins.py @@ -142,6 +142,49 @@ def classify(ref: str) -> str: return "mutable" if tag.lower() in MUTABLE_TAGS else "tag" +def bare_name(ref: str) -> str: + """A reference with its tag removed, which is what an exemption names. + + Split on the last colon only when it sits in the final path segment: a + registry with a port (`registry:5000/x/y`) carries a colon that is not a tag + separator, and cutting there would produce a key no exemption can match and + no reader can recognise. + """ + name = ref.rsplit("/", 1)[-1] + return ref.rsplit(":", 1)[0] if ":" in name else ref + + +def verdict(images: dict[str, set[str]], allowed: dict[str, str]) -> list[str]: + """Every image-pin problem, over an inventory and an exemption list. + + Two directions, and the second is the one that rots. A mutable reference with + no exemption is the defect the gate exists for. An exemption the fleet no + longer renders mutably is a description that outlived its reason, and an + exemption list nobody re-checks only ever widens. + """ + failures = [] + mutable_seen: set[str] = set() + + for ref in sorted(images): + if classify(ref) != "mutable": + continue + bare = bare_name(ref) + mutable_seen.add(bare) + if bare in allowed: + continue + failures.append( + f"{ref} (via {', '.join(sorted(images[ref]))}) resolves to a moving target. " + f"Pin it in the addon's values.yaml to the chart's appVersion or a digest.") + + for bare, reason in sorted(allowed.items()): + if bare not in mutable_seen: + failures.append( + f"{bare} is on the mutable-tag exemption list but the fleet no longer " + f"renders it mutably — the exemption outlived its reason. Delete it. " + f"(recorded: {reason[:100]})") + return failures + + def main() -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--list", action="store_true", help="print the image inventory") @@ -165,27 +208,7 @@ def main() -> int: print(f" {path}: {err}") return 2 - failures = [] - mutable_seen: set[str] = set() - - for ref in sorted(images): - if classify(ref) != "mutable": - continue - name = ref.rsplit("/", 1)[-1] - bare = ref.rsplit(":", 1)[0] if ":" in name else ref - mutable_seen.add(bare) - if bare in ALLOWED_MUTABLE: - continue - failures.append( - f"{ref} (via {', '.join(sorted(images[ref]))}) resolves to a moving target. " - f"Pin it in the addon's values.yaml to the chart's appVersion or a digest.") - - for bare, reason in sorted(ALLOWED_MUTABLE.items()): - if bare not in mutable_seen: - failures.append( - f"{bare} is on the mutable-tag exemption list but the fleet no longer " - f"renders it mutably — the exemption outlived its reason. Delete it. " - f"(recorded: {reason[:100]})") + failures = verdict(images, ALLOWED_MUTABLE) # Reported whatever the verdict: a chart that did not render contributed no # images, and counting the rest as the whole fleet is how a partial scan diff --git a/scripts/check-log-volume-budget.py b/scripts/check-log-volume-budget.py index a09896f..6e4b254 100755 --- a/scripts/check-log-volume-budget.py +++ b/scripts/check-log-volume-budget.py @@ -172,6 +172,53 @@ def alert_threshold() -> float | None: return None +def environment_verdict(cfg: dict, warn: float | None, + rel: str) -> tuple[list[str], bool]: + """Everything wrong with one environment's rendered Loki config. + + Separate from the render because the render reaches a chart repository and + this does not. The three assertions are what decide the gate's outcome, so + they are reachable with a config supplied directly rather than only through a + network round trip. + + Returns the problems found and whether the alert leads a declared cutoff — + the second is what the closing line counts, so an environment that failed the + comparison is not also reported as one the comparison covered. + """ + wal = ((cfg.get("ingester") or {}).get("wal") or {}) + limits = cfg.get("limits_config") or {} + comp = cfg.get("compactor") or {} + problems: list[str] = [] + leads = False + + cutoff = wal.get("disk_full_threshold") + if cutoff is None: + problems.append( + f"{rel}: the render sets no ingester.wal.disk_full_threshold, so the " + f"fraction at which Loki stops accepting every push is an upstream " + f"default. The alert's lead time then depends on a number that can " + f"move under a chart bump with nothing here to compare against.") + elif warn is not None and not warn < float(cutoff): + problems.append( + f"{rel}: the fill alert fires at {warn} but ingestion stops at " + f"{cutoff}. There is no window in which to act — and no remedy is " + f"fast: a retention cut must sync, wait a compaction interval, then " + f"clear retention_delete_delay before a byte is freed, and the volume " + f"cannot be grown through this repo at all.") + else: + leads = True + + if limits.get("retention_period") is None: + problems.append( + f"{rel}: the render sets no limits_config.retention_period. Nothing " + f"then deletes on a schedule, and the volume reaches the cutoff.") + if comp.get("retention_enabled") is not True: + problems.append( + f"{rel}: compactor.retention_enabled is not true, so retention_period " + f"deletes nothing however it is set and the cutoff arrives regardless.") + return problems, leads + + def main() -> int: repo, version = chart_pin() warn = alert_threshold() @@ -183,32 +230,11 @@ def main() -> int: checked = 0 for env in envs: cfg = render(repo, version, env) - wal = ((cfg.get("ingester") or {}).get("wal") or {}) - limits = cfg.get("limits_config") or {} - comp = cfg.get("compactor") or {} rel = f"addons/observability/loki/values-{env}.yaml" - - cutoff = wal.get("disk_full_threshold") - if cutoff is None: - fail(f"{rel}: the render sets no ingester.wal.disk_full_threshold, so the " - f"fraction at which Loki stops accepting every push is an upstream " - f"default. The alert's lead time then depends on a number that can " - f"move under a chart bump with nothing here to compare against.") - elif warn is not None and not warn < float(cutoff): - fail(f"{rel}: the fill alert fires at {warn} but ingestion stops at " - f"{cutoff}. There is no window in which to act — and no remedy is " - f"fast: a retention cut must sync, wait a compaction interval, then " - f"clear retention_delete_delay before a byte is freed, and the volume " - f"cannot be grown through this repo at all.") - else: - checked += 1 - - if limits.get("retention_period") is None: - fail(f"{rel}: the render sets no limits_config.retention_period. Nothing " - f"then deletes on a schedule, and the volume reaches the cutoff.") - if comp.get("retention_enabled") is not True: - fail(f"{rel}: compactor.retention_enabled is not true, so retention_period " - f"deletes nothing however it is set and the cutoff arrives regardless.") + problems, leads = environment_verdict(cfg, warn, rel) + for problem in problems: + fail(problem) + checked += 1 if leads else 0 if failures: for f in failures: diff --git a/scripts/render-addons.py b/scripts/render-addons.py index 54e1231..ea063d7 100755 --- a/scripts/render-addons.py +++ b/scripts/render-addons.py @@ -239,14 +239,25 @@ def discover() -> list[Unit]: ) +def registry_answered(err: str) -> bool: + """Whether helm's complaint says the registry replied and refused the pin. + + Separate from the exit path so the two worlds can be told apart without + running helm. A message naming both a missing chart and a broken connection + is the second: an unreachable registry cannot testify about what it holds. + """ + low = err.lower() + return (any(t in low for t in _NOT_FOUND) + and not any(t in low for t in _UNREACHABLE)) + + def helm_or_exit(cmd: list[str], what: str) -> subprocess.CompletedProcess: """Run a helm command; on failure exit with the RIGHT kind of complaint.""" proc = subprocess.run(cmd, capture_output=True, text=True, timeout=NETWORK_TIMEOUT) if proc.returncode == 0: return proc err = ((proc.stderr or "") + (proc.stdout or "")).strip() - low = err.lower() - if any(t in low for t in _NOT_FOUND) and not any(t in low for t in _UNREACHABLE): + if registry_answered(err): print(f"{what}: the registry answered and the pinned chart is not there.") print(err) print("This is a pin that does not resolve — a finding about this repo.") diff --git a/scripts/tests/run.py b/scripts/tests/run.py index 75ef305..062e255 100755 --- a/scripts/tests/run.py +++ b/scripts/tests/run.py @@ -19,6 +19,7 @@ from __future__ import annotations +import importlib.util import os import pathlib import sys @@ -36,21 +37,56 @@ "test_named_things", "test_falco_rule_floor", "test_gatelib", + # The gates the positive-control sweep exempts, because they reach a chart + # registry or an API. A control cannot be written for them, so a unit test on + # the half that decides the verdict is the only thing that proves they say + # anything — and they are the largest files in the tree. + "test_policy_admission", + "test_platform_crs", + "test_dashboards", + "test_render_addons", + "test_image_pins", + "test_log_volume_budget", ) # A floor well under the real count. It catches "discovery found almost nothing", # not "somebody removed one test". MIN_TESTS = 31 -# Line coverage floors over the gate scripts, enforced when `coverage` is -# available. A ratchet, not a target: each is set just under what the suite -# reaches, so coverage cannot fall silently, and raising one is what lands with +# THE ASSERTION THAT IS NOT A NUMBER +# +# Every gate carries at least one kind of proof that it says something: +# +# * a positive control in controls.py, which supplies the violation the gate +# names and requires a rejection; or +# * unit tests over the decision that produces the verdict. +# +# A gate with neither has been read by nobody and asserts nothing that anything +# has checked. controls.py exempts every gate that reaches a chart registry or an +# API — the exemption is honest, because a control cannot be written for a gate +# whose input arrives over the network — and those are the largest files in the +# tree. Among them is check-policy-admission.py, an approval gate in the sense +# testing-rubric's `security-critical-100` uses the term. For exactly these, a +# unit test is the only proof available. +# +# `control_exempt_gates` below reads that exemption list out of controls.py +# rather than repeating it, and check_coverage fails on any gate in both it and +# the uncovered set. A gate added to the exemption must arrive with tests, and +# cannot be excused by a percentage that moves a point. +# +# The floors that follow are the weaker half. They stop a covered file +# regressing; they cannot state what is covered, and a count loose enough to +# survive an honest refactor is loose enough to miss a decision going untested. + +# Combined statement+branch coverage floor over the gate scripts, enforced when +# `coverage` is available. A ratchet, not a target: set just under what the suite +# reaches, so coverage cannot fall silently, and raising it is what lands with # the tests that earn it. # -# The org testing-rubric asks for 75% lines and 60% branches. COMBINED_FLOOR -# sitting far below that is the record of the gap, kept rather than papered over: -# most gate files carry no unit tests at all, and the floor cannot rise without -# writing them. Read the current standing with +# The org testing-rubric asks for 75% lines and 60% branches, and this sits under +# that. What the gap is made of is readable rather than asserted here: the gates +# still carrying no unit tests are the ones controls.py DOES cover, so each of +# them has been observed to reject the violation it names. Read the standing with # # COVERAGE_REQUIRED=1 coverage run --rcfile=.coveragerc scripts/tests/run.py # coverage report --rcfile=.coveragerc --include="scripts/*.py" @@ -60,32 +96,18 @@ # # COMBINED_FLOOR is compared against what coverage.report() returns, which with # branch=True is a COMBINED statement-and-branch figure, not line coverage. The -# two differ enough to matter, so the name below says combined and the printed +# two differ enough to matter, so the name here says combined and the printed # line says so too. Calling a combined figure "line coverage" would be a # measurement mislabelled as the one the rubric asks about. -# -# What the number does NOT capture: scripts/tests/controls.py runs most gates end -# to end as a subprocess against a mutated tree, so those gates have behavioural -# coverage this line count cannot see. Neither figure substitutes for the other — -# the controls prove a gate rejects, the unit tests prove it computes the right -# answer on a case the real tree does not contain. -# -# MOST, and the exceptions are the ones that matter to this floor. controls.py -# exempts every gate that reaches a chart registry or an API, and its own run -# prints the split. Those gates have neither kind of coverage, they are the -# largest in the tree, and among them are the gates on the paths testing-rubric -# calls security-critical. So this figure being low is not offset by behavioural -# coverage for precisely the files where that offset was being claimed. -COMBINED_FLOOR = 12 +COMBINED_FLOOR = 29 # A ceiling on gate scripts carrying NO unit coverage at all, complementing the # floors below. The floors stop a covered file regressing; nothing stopped a NEW -# gate arriving with no tests, and most of this tree arrived that way. +# gate arriving with no tests. # # Ratchets downward only. Adding a gate without tests fails here rather than -# diluting the combined figure by a percentage point nobody notices, which is how -# a suite reaches this state one honest commit at a time. -MAX_UNCOVERED_GATES = 17 +# diluting the combined figure by a percentage point nobody notices. +MAX_UNCOVERED_GATES = 11 PER_GATE_FLOORS = { "scripts/check-named-things.py": 35, @@ -93,6 +115,15 @@ "scripts/check-label-values.py": 33, "scripts/check-sync-waves.py": 10, "scripts/gatelib.py": 55, + # The control-exempt gates. Their floors are what keeps the verdict half + # covered after the module that covers it is edited. + "scripts/check-policy-admission.py": 50, + "scripts/check-platform-crs.py": 50, + "scripts/validate-dashboards.py": 45, + "scripts/render-addons.py": 40, + "scripts/check-image-pins.py": 38, + "scripts/check-log-volume-budget.py": 58, + "scripts/check-falco-rule-floor.py": 28, } @@ -119,6 +150,36 @@ def main() -> int: return check_coverage() +def control_exempt_gates() -> tuple[list[str], str]: + """Python gates controls.py exempts from a positive control, and why it might not. + + Read out of controls.py rather than repeated here. A second copy of an + exemption list is a copy that drifts, and the direction it drifts is + permissive: the gate added to one list and not the other is excused by both. + + A list that cannot be read is reported rather than treated as empty — an + empty exemption set makes the assertion above vacuously true, which is the + shape this whole suite exists to reject. + """ + path = HERE / "controls.py" + try: + spec = importlib.util.spec_from_file_location("controls", path) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + sys.modules["controls"] = module + spec.loader.exec_module(module) + exempt = sorted(module.NEEDS_NETWORK_PY) + except Exception as exc: # noqa: BLE001 — reported, not raised + return [], (f"{path.name} could not be read for its control exemptions " + f"({type(exc).__name__}: {exc}), so no gate could be checked for " + f"having neither kind of proof") + if not exempt: + return [], (f"{path.name} exempts no gate from a positive control, so the " + f"assertion that every exempt gate carries unit tests holds over " + f"an empty set and states nothing") + return exempt, "" + + def check_coverage() -> int: """Enforce the coverage ratchet, when this run was measured. @@ -201,6 +262,19 @@ def check_coverage() -> int: f"{', '.join(uncovered[:4])}{' …' if len(uncovered) > 4 else ''}" ) + exempt, exempt_problem = control_exempt_gates() + if exempt_problem: + failures.append(exempt_problem) + unproven = sorted(set(exempt) & set(uncovered)) + if unproven: + failures.append( + f"{len(unproven)} gate(s) have neither a positive control nor unit " + f"coverage: {', '.join(unproven)}. controls.py exempts them because " + f"their input arrives over the network, so a unit test on the half " + f"that decides the verdict is the only proof available that they say " + f"anything at all" + ) + if failures: print() for f in failures: @@ -212,8 +286,11 @@ def check_coverage() -> int: print(f"coverage ratchet OK: {total:.1f}% combined statement+branch " f"(floor {COMBINED_FLOOR}%), {len(PER_GATE_FLOORS)} per-gate floor(s) held, " f"{len(uncovered)} of {len(gates)} gate script(s) with no unit coverage " - f"(ceiling {MAX_UNCOVERED_GATES}). Well under the rubric's 75% lines / " - f"60% branches — see the note in this file.") + f"(ceiling {MAX_UNCOVERED_GATES}). Under the rubric's 75% lines / 60% " + f"branches — see the note in this file.") + print(f"control-exempt gates: all {len(exempt)} carry unit coverage. " + f"scripts/tests/controls.py holds the other half of the claim — that " + f"every gate it does NOT exempt rejects the violation it names.") return 0 diff --git a/scripts/tests/test_dashboards.py b/scripts/tests/test_dashboards.py new file mode 100644 index 0000000..60f2f05 --- /dev/null +++ b/scripts/tests/test_dashboards.py @@ -0,0 +1,227 @@ +"""Unit tests for the dashboard gate's extractors. + +The gate resolves grafana.com ids over the network, so the positive-control +sweep exempts it. Its offline half is where the failures are quiet: a dashboard +is data, so a panel naming a datasource nothing wires renders an error forever +in front of whoever opened the board, and nothing in the sync path reports it. + +Every extractor below has the same failure mode — matching less than it should +and comparing an empty set, which passes. The variable regex had exactly that: +anchored on `${name}` alone it matched none of the bare `$name` references every +dashboard here actually writes, so the undeclared-variable check compared an +always-empty set against the declared list and passed over every board. +""" + +from __future__ import annotations + +import json +import pathlib +import tempfile +import textwrap +import unittest + +from gateloader import load + +gate = load("validate-dashboards") + +ROOT = pathlib.Path(__file__).resolve().parent.parent.parent + + +class TemplateVariableReferences(unittest.TestCase): + """Grafana writes a variable reference four ways; all four must be seen.""" + + def test_the_braced_form(self): + self.assertEqual(gate.template_vars("SELECT ${cost_database}.t"), + {"cost_database"}) + + def test_the_braced_form_with_a_format_suffix(self): + self.assertEqual(gate.template_vars("${namespace:csv}"), {"namespace"}) + + def test_the_bare_form_every_dashboard_here_uses(self): + self.assertEqual(gate.template_vars('{namespace="$namespace"}'), {"namespace"}) + + def test_the_bracket_form(self): + self.assertEqual(gate.template_vars("[[tenant]] and [[tenant:csv]]"), + {"tenant"}) + + def test_grafana_builtins_are_not_reported(self): + text = "rate(x[$__rate_interval]) offset $__range and $__from to $__to" + self.assertEqual(gate.template_vars(text), set()) + + def test_a_builtin_beside_a_real_variable_leaves_the_real_one(self): + self.assertEqual( + gate.template_vars('sum(rate(x{ns="$namespace"}[$__rate_interval]))'), + {"namespace"}) + + def test_every_builtin_in_the_allowlist_is_suppressed(self): + for name in sorted(gate.GRAFANA_BUILTINS): + with self.subTest(builtin=name): + self.assertEqual(gate.template_vars(f"${name}"), set()) + + +class ExtractingTheDashboardBody(unittest.TestCase): + """Two structurally different failures, told apart so the fix is findable.""" + + def manifest(self, body, indent=4): + pad = " " * indent + block = "\n".join(f"{pad} {line}" for line in body.splitlines()) + return (f"kind: GrafanaDashboard\nspec:\n{pad}json: |\n{block}\n") + + def test_a_well_formed_block_parses(self): + dash, reason = gate.extract_dashboard_json( + self.manifest(json.dumps({"title": "T", "panels": []}, indent=1))) + self.assertIsNone(reason) + self.assertEqual(dash["title"], "T") + + def test_a_manifest_with_no_block_says_so(self): + dash, reason = gate.extract_dashboard_json("kind: GrafanaDashboard\nspec: {}\n") + self.assertIsNone(dash) + self.assertIn("no `json: |` literal block", reason) + + def test_invalid_json_is_reported_as_a_syntax_error_not_a_missing_block(self): + dash, reason = gate.extract_dashboard_json(self.manifest('{"title": }')) + self.assertIsNone(dash) + self.assertIn("not valid JSON", reason) + + def test_the_reported_line_is_one_the_reader_can_open(self): + """A line number relative to the block sends the reader to the wrong place.""" + text = self.manifest('{\n "title": "T",\n "panels": [ }\n') + _, reason = gate.extract_dashboard_json(text) + line = int(reason.rsplit("file line ", 1)[1]) + self.assertLessEqual(line, len(text.splitlines())) + self.assertGreater(line, 3) + + def test_content_after_the_block_is_not_swallowed(self): + text = (self.manifest(json.dumps({"title": "T"})) + + " instanceSelector:\n matchLabels:\n app: grafana\n") + dash, reason = gate.extract_dashboard_json(text) + self.assertIsNone(reason) + self.assertEqual(dash, {"title": "T"}) + + +class DatasourceReferencesInAPanel(unittest.TestCase): + """Grafana accepts both a bare string and a `{type, uid}` object.""" + + def refs(self, node): + out: list[str] = [] + gate.walk_datasources(node, out) + return sorted(out) + + def test_a_string_datasource_is_collected(self): + self.assertEqual(self.refs({"datasource": "athena-cur"}), ["athena-cur"]) + + def test_an_object_datasource_yields_its_uid(self): + self.assertEqual( + self.refs({"datasource": {"type": "prometheus", "uid": "amp"}}), ["amp"]) + + def test_a_datasource_object_with_no_uid_yields_nothing(self): + self.assertEqual(self.refs({"datasource": {"type": "prometheus"}}), []) + + def test_nested_panels_and_targets_are_reached(self): + """A collapsed row is a panel whose children hang off a nested list.""" + dash = {"panels": [{"collapsed": True, "panels": [ + {"datasource": "loki", "targets": [{"datasource": {"uid": "tempo"}}]}]}]} + self.assertEqual(self.refs(dash), ["loki", "tempo"]) + + +class DeclaredVariables(unittest.TestCase): + def test_names_are_read_off_the_templating_list(self): + dash = {"templating": {"list": [{"name": "namespace"}, {"name": "tenant"}]}} + self.assertEqual(gate.declared_template_vars(dash), {"namespace", "tenant"}) + + def test_a_dashboard_with_no_templating_block_declares_nothing(self): + self.assertEqual(gate.declared_template_vars({}), set()) + self.assertEqual(gate.declared_template_vars({"templating": {}}), set()) + + def test_a_malformed_entry_is_skipped_rather_than_raising(self): + dash = {"templating": {"list": ["oops", {"name": 3}, {"name": "ok"}]}} + self.assertEqual(gate.declared_template_vars(dash), {"ok"}) + + +class LegacyAlertPanels(unittest.TestCase): + """A grafana.com dashboard carrying legacy alerts is not AMG-saveable.""" + + def titles(self, node): + out: list[str] = [] + gate.alert_panels(node, out) + return out + + def test_a_top_level_alert_panel_is_found(self): + self.assertEqual(self.titles({"panels": [{"title": "CPU", "alert": {}}]}), + ["CPU"]) + + def test_an_alert_inside_a_collapsed_row_is_found(self): + """A flat scan misses every alert panel a collapsed row contains.""" + dash = {"panels": [{"title": "row", "collapsed": True, + "panels": [{"title": "Disk", "alert": {"x": 1}}]}]} + self.assertEqual(self.titles(dash), ["Disk"]) + + def test_a_null_alert_key_is_not_an_alert(self): + self.assertEqual(self.titles({"panels": [{"title": "CPU", "alert": None}]}), []) + + def test_an_untitled_alert_panel_is_still_reported(self): + self.assertEqual(self.titles({"panels": [{"alert": {}}]}), + [""]) + + +class WiredDatasources(unittest.TestCase): + """Read from the kustomization, not the directory. + + A GrafanaDatasource file that exists but is not in `resources` is never + applied, so globbing the directory would call a datasource wired when it is + not — and the panel referencing it would pass the check and fail at view time. + """ + + def tree(self, kustomization: str, files: dict[str, str]): + root = pathlib.Path(tempfile.mkdtemp()) + base = root / "dashboards" / "base" + (base / "datasources").mkdir(parents=True) + (base / "kustomization.yaml").write_text(textwrap.dedent(kustomization)) + for name, body in files.items(): + (base / "datasources" / name).write_text(textwrap.dedent(body)) + return root + + def test_both_the_uid_and_the_name_are_accepted(self): + root = self.tree( + """ + resources: + - datasources/loki.yaml + """, + {"loki.yaml": """ + spec: + datasource: + uid: managed-loki + name: Loki + """}) + self.assertEqual(gate.wired_datasource_refs(root), {"managed-loki", "Loki"}) + + def test_a_datasource_file_absent_from_resources_is_not_wired(self): + root = self.tree( + """ + resources: + - datasources/loki.yaml + """, + {"loki.yaml": "spec:\n datasource:\n uid: managed-loki\n", + "athena.yaml": "spec:\n datasource:\n uid: athena-cur\n"}) + self.assertEqual(gate.wired_datasource_refs(root), {"managed-loki"}) + + def test_a_tree_with_no_kustomization_wires_nothing(self): + self.assertEqual(gate.wired_datasource_refs(pathlib.Path(tempfile.mkdtemp())), + set()) + + +class TheShippedDashboards(unittest.TestCase): + """The offline half, over the corpus it actually governs.""" + + def test_the_local_checks_find_no_problem_in_the_tree(self): + self.assertEqual(gate.check_local_dashboards(ROOT), []) + + def test_the_tree_wires_at_least_one_datasource(self): + """An empty wired set makes every panel reference a reference to nothing — + which the check would report on all of them, or on none, depending only on + whether any dashboard names a datasource at all.""" + self.assertTrue(gate.wired_datasource_refs(ROOT)) + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/tests/test_image_pins.py b/scripts/tests/test_image_pins.py new file mode 100644 index 0000000..cfa6e99 --- /dev/null +++ b/scripts/tests/test_image_pins.py @@ -0,0 +1,166 @@ +"""Unit tests for the image-pin gate's classification and verdict. + +The gate renders every chart the catalog pins to learn which images actually +reach a cluster, so it reaches a chart registry and the positive-control sweep +exempts it. What that leaves untested is the rule that decides the outcome: +whether a reference resolves to a moving target. + +The rule is where the interesting inputs are. A reference can carry a digest, a +version tag, a mutable tag, a registry port that looks like a tag, or no tag at +all — and the fleet contains one example of some of those and none of others. +These supply all of them. +""" + +from __future__ import annotations + +import unittest + +from gateloader import load + +gate = load("check-image-pins") + + +class ClassifyingAReference(unittest.TestCase): + """'digest', 'tag' or 'mutable'. Only the third fails the gate.""" + + def test_a_digest_is_immutable(self): + self.assertEqual( + gate.classify("ghcr.io/opencost/opencost:1.121.1@sha256:" + "5" * 64), + "digest") + + def test_a_digest_without_a_tag_is_immutable(self): + self.assertEqual(gate.classify("quay.io/cilium/cilium@sha256:" + "0" * 64), + "digest") + + def test_a_version_tag_is_accepted(self): + for ref in ("quay.io/jetstack/cert-manager-controller:v1.21.1", + "memcached:1.6.45-alpine", + "docker.io/envoyproxy/ratelimit:1e50889b"): + with self.subTest(ref=ref): + self.assertEqual(gate.classify(ref), "tag") + + def test_an_untagged_reference_is_mutable(self): + """No tag resolves to :latest, which is the defect wearing no clothes.""" + self.assertEqual(gate.classify("ghcr.io/kyverno/readiness-checker"), "mutable") + self.assertEqual(gate.classify("alpine"), "mutable") + + def test_every_moving_tag_is_mutable(self): + for tag in sorted(gate.MUTABLE_TAGS): + with self.subTest(tag=tag): + self.assertEqual(gate.classify(f"docker.io/library/nginx:{tag}"), + "mutable") + + def test_a_moving_tag_is_recognised_whatever_its_case(self): + self.assertEqual(gate.classify("docker.io/library/nginx:LATEST"), "mutable") + + def test_a_registry_port_is_not_read_as_a_tag(self): + """`registry:5000/x/y` carries a colon that is not a tag separator. + + Read as one, an untagged image behind a private registry would be + classified as pinned — the gate would pass exactly the reference it + exists to catch. + """ + self.assertEqual(gate.classify("registry:5000/nanohype/agent"), "mutable") + self.assertEqual(gate.classify("registry:5000/nanohype/agent:1.2.3"), "tag") + + def test_a_tag_containing_a_moving_word_is_not_itself_moving(self): + self.assertEqual(gate.classify("docker.io/x/y:v1-stable-2"), "tag") + + +class TheBareNameAnExemptionMatches(unittest.TestCase): + """An exemption names the image, not the tag it happened to carry.""" + + def test_the_tag_is_dropped(self): + self.assertEqual(gate.bare_name("docker.io/library/nginx:latest"), + "docker.io/library/nginx") + + def test_an_untagged_reference_is_its_own_bare_name(self): + self.assertEqual(gate.bare_name("ghcr.io/kyverno/readiness-checker"), + "ghcr.io/kyverno/readiness-checker") + + def test_a_registry_port_survives(self): + self.assertEqual(gate.bare_name("registry:5000/nanohype/agent"), + "registry:5000/nanohype/agent") + + +class TheVerdict(unittest.TestCase): + """Both directions fail: an unpinned image, and an exemption that rotted.""" + + def test_an_all_pinned_fleet_has_no_problems(self): + images = {"quay.io/cilium/cilium@sha256:" + "0" * 64: {"cilium"}, + "registry.k8s.io/metrics-server/metrics-server:v0.8.1": {"metrics-server"}} + self.assertEqual(gate.verdict(images, {}), []) + + def test_a_mutable_reference_is_reported_with_the_chart_that_renders_it(self): + images = {"ghcr.io/kyverno/readiness-checker": {"kyverno"}} + problems = gate.verdict(images, {}) + self.assertEqual(len(problems), 1) + self.assertIn("via kyverno", problems[0]) + self.assertIn("moving target", problems[0]) + + def test_one_image_rendered_by_several_charts_names_all_of_them(self): + images = {"docker.io/library/nginx:latest": {"loki", "tempo"}} + self.assertIn("via loki, tempo", gate.verdict(images, {})[0]) + + def test_an_exemption_suppresses_exactly_its_own_image(self): + images = {"docker.io/library/nginx:latest": {"loki"}, + "docker.io/library/redis:latest": {"tempo"}} + problems = gate.verdict(images, {"docker.io/library/nginx": "recorded reason"}) + self.assertEqual(len(problems), 1) + self.assertIn("redis", problems[0]) + + def test_an_exemption_the_fleet_no_longer_renders_mutably_fails(self): + """An exemption list nobody re-checks only ever widens.""" + images = {"quay.io/cilium/cilium@sha256:" + "0" * 64: {"cilium"}} + problems = gate.verdict(images, {"docker.io/library/nginx": "recorded reason"}) + self.assertEqual(len(problems), 1) + self.assertIn("outlived its reason", problems[0]) + + def test_an_exemption_for_an_image_now_pinned_by_tag_fails(self): + """Pinning the image is the fix; leaving the exemption behind is the rot.""" + images = {"docker.io/library/nginx:1.31-alpine": {"loki"}} + problems = gate.verdict(images, {"docker.io/library/nginx": "recorded reason"}) + self.assertEqual(len(problems), 1) + self.assertIn("outlived its reason", problems[0]) + + def test_the_recorded_reason_is_quoted_back(self): + problems = gate.verdict({}, {"docker.io/library/nginx": "waiting on chart 2.0"}) + self.assertIn("waiting on chart 2.0", problems[0]) + + +class TheImageExtractor(unittest.TestCase): + """A regex that stops matching turns a full fleet into a clean one.""" + + def refs(self, text): + return [m.group(1) for m in gate.IMAGE.finditer(text)] + + def test_a_plain_image_line_matches(self): + self.assertEqual(self.refs(" image: nginx:1.27\n"), ["nginx:1.27"]) + + def test_a_quoted_image_line_matches_without_the_quotes(self): + self.assertEqual(self.refs(' image: "nginx:1.27"\n'), ["nginx:1.27"]) + + def test_a_key_merely_ending_in_image_does_not_match(self): + for line in (" initImage: nginx:1.27\n", " image_pull_policy: Always\n"): + with self.subTest(line=line.strip()): + self.assertEqual(self.refs(line), []) + + def test_every_image_in_a_multi_document_render_is_found(self): + text = ("kind: Deployment\n image: a:1\n---\n" + "kind: DaemonSet\n image: b:2\n") + self.assertEqual(self.refs(text), ["a:1", "b:2"]) + + +class TheShippedExemptionList(unittest.TestCase): + """An empty allowlist grants nothing, and that is the state to hold.""" + + def test_every_entry_carries_a_recorded_reason(self): + for bare, reason in gate.ALLOWED_MUTABLE.items(): + with self.subTest(image=bare): + self.assertTrue(reason.strip(), + f"{bare} is exempted with no reason recorded, so " + f"nothing states what would let it be removed") + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/tests/test_log_volume_budget.py b/scripts/tests/test_log_volume_budget.py new file mode 100644 index 0000000..8379d7f --- /dev/null +++ b/scripts/tests/test_log_volume_budget.py @@ -0,0 +1,210 @@ +"""Unit tests for the Loki log-volume gate's verdict. + +The gate holds one relation: the alert that warns about the volume must fire +strictly before the fraction at which Loki stops accepting every push. Both +operands come from elsewhere — the cutoff from a rendered chart, the threshold +from a shipped alert file — so the gate reaches a chart repository and the +positive-control sweep exempts it. + +What that leaves untested is the comparison itself, and the comparison is the +gate. These supply both operands directly. + +The alert-reading half is exercised against fixture files rather than the shipped +one: the shipped rule is a single sample of a shape with several failure modes, +and the ones that matter are the shapes it does not have. +""" + +from __future__ import annotations + +import pathlib +import tempfile +import unittest + +import yaml +from gateloader import load + +gate = load("check-log-volume-budget") + +REL = "addons/observability/loki/values-production.yaml" + + +def config(cutoff=0.9, retention="2160h", retention_enabled=True): + """A rendered Loki config carrying only the keys this gate reads.""" + cfg: dict = {"ingester": {"wal": {}}, "limits_config": {}, "compactor": {}} + if cutoff is not None: + cfg["ingester"]["wal"]["disk_full_threshold"] = cutoff + if retention is not None: + cfg["limits_config"]["retention_period"] = retention + if retention_enabled is not None: + cfg["compactor"]["retention_enabled"] = retention_enabled + return cfg + + +def alert_file(rules) -> pathlib.Path: + """Write a Grafana alerting manifest carrying `rules` and return its path.""" + tmp = pathlib.Path(tempfile.mkdtemp()) / "loki-disk.yaml" + tmp.write_text(yaml.safe_dump({"apiVersion": "grafana.integreatly.org/v1beta1", + "kind": "GrafanaAlertRuleGroup", + "spec": {"rules": rules}})) + return tmp + + +def fill_rule(expr=None, threshold=0.75, uid=None): + return { + "uid": uid or gate.FILL_RULE, + "data": [ + {"refId": "A", + "model": {"expr": expr or f"max({gate.GAUGE})"}}, + {"refId": "B", + "model": {"conditions": [ + {"evaluator": {"type": "gt", "params": [threshold]}}]}}, + ], + } + + +class ReadingTheAlert(unittest.TestCase): + """The alert supplies the left operand. Every way it can fail to is a pass.""" + + def setUp(self): + self.original = (gate.ALERT, gate.ROOT) + gate.failures.clear() + + def tearDown(self): + gate.ALERT, gate.ROOT = self.original + gate.failures.clear() + + def threshold(self, rules): + # ROOT moves with ALERT: the gate reports the alert path relative to the + # repo root, so a fixture outside it would raise instead of failing. + gate.ALERT = alert_file(rules) + gate.ROOT = gate.ALERT.parent + return gate.alert_threshold() + + def test_the_evaluator_threshold_is_returned(self): + self.assertEqual(self.threshold([fill_rule(threshold=0.75)]), 0.75) + self.assertEqual(gate.failures, []) + + def test_a_file_with_no_such_rule_fails_rather_than_returning_nothing(self): + """Deleting the rule must not read as 'no threshold to compare against'.""" + self.assertIsNone(self.threshold([fill_rule(uid="something-else")])) + self.assertEqual(len(gate.failures), 1) + self.assertIn("no rule with uid", gate.failures[0]) + + def test_a_file_with_no_rules_at_all_fails(self): + self.assertIsNone(self.threshold([])) + self.assertIn("no rule with uid", gate.failures[0]) + + def test_a_rule_querying_the_edge_counter_fails_twice(self): + """It increments once per transition into the throttled state. + + A cluster throttled for a week increments it once, so a rate() alert on it + reads as quiet through exactly the outage it exists to report. + """ + self.threshold([fill_rule(expr=f"max(rate({gate.EDGE_COUNTER}[5m]))")]) + joined = " ".join(gate.failures) + self.assertIn("does not query", joined) + self.assertIn("increments once per transition", joined) + + def test_a_rule_querying_both_still_fails_on_the_edge_counter(self): + self.threshold([fill_rule( + expr=f"max({gate.GAUGE}) or max({gate.EDGE_COUNTER})")]) + self.assertEqual(len(gate.failures), 1) + self.assertIn("increments once per transition", gate.failures[0]) + + def test_a_rule_with_no_evaluator_fails(self): + rule = fill_rule() + rule["data"] = [{"refId": "A", "model": {"expr": f"max({gate.GAUGE})"}}] + self.assertIsNone(self.threshold([rule])) + self.assertIn("carries no evaluator threshold", gate.failures[0]) + + +class TheCutoffComparison(unittest.TestCase): + """The gate's whole claim: the warning leads the cutoff, with room to act.""" + + def test_a_warning_ahead_of_the_cutoff_passes(self): + problems, leads = gate.environment_verdict(config(cutoff=0.9), 0.75, REL) + self.assertEqual(problems, []) + self.assertTrue(leads) + + def test_a_warning_at_the_cutoff_leaves_no_window(self): + """Equal is not ahead: the alert and the outage arrive together.""" + problems, leads = gate.environment_verdict(config(cutoff=0.75), 0.75, REL) + self.assertFalse(leads) + self.assertIn("no window in which to act", problems[0]) + + def test_a_warning_past_the_cutoff_fails(self): + problems, leads = gate.environment_verdict(config(cutoff=0.7), 0.75, REL) + self.assertFalse(leads) + self.assertIn("fires at 0.75 but ingestion stops at 0.7", problems[0]) + + def test_an_inherited_cutoff_fails(self): + """Unset, the lead time depends on a chart default that can move.""" + problems, leads = gate.environment_verdict(config(cutoff=None), 0.75, REL) + self.assertFalse(leads) + self.assertIn("sets no ingester.wal.disk_full_threshold", problems[0]) + + def test_a_cutoff_written_as_a_string_is_still_compared(self): + problems, leads = gate.environment_verdict(config(cutoff="0.9"), 0.75, REL) + self.assertEqual(problems, []) + self.assertTrue(leads) + + def test_an_environment_that_failed_the_comparison_is_not_counted_as_covered(self): + """The closing line reports how many environments the comparison covered. + + Counting a failed one there would say the relation was checked and held. + """ + _, leads = gate.environment_verdict(config(cutoff=0.5), 0.75, REL) + self.assertFalse(leads) + + +class RetentionMustActuallyDelete(unittest.TestCase): + """A cutoff nothing deletes toward is a cutoff that arrives on a schedule.""" + + def test_a_missing_retention_period_fails(self): + problems, _ = gate.environment_verdict(config(retention=None), 0.75, REL) + self.assertIn("sets no limits_config.retention_period", problems[0]) + + def test_retention_disabled_in_the_compactor_fails(self): + """`retention_period` applies only when the compactor enforces it.""" + problems, _ = gate.environment_verdict( + config(retention_enabled=False), 0.75, REL) + self.assertIn("compactor.retention_enabled is not true", problems[0]) + + def test_an_absent_compactor_block_fails(self): + problems, _ = gate.environment_verdict( + config(retention_enabled=None), 0.75, REL) + self.assertIn("compactor.retention_enabled is not true", problems[0]) + + def test_a_truthy_non_true_value_does_not_satisfy_it(self): + """Loki reads the key as a bool; `"true"` is a string the chart passes on.""" + problems, _ = gate.environment_verdict( + config(retention_enabled="true"), 0.75, REL) + self.assertIn("compactor.retention_enabled is not true", problems[0]) + + def test_every_failure_is_reported_from_one_pass(self): + """A gate that returns on the first problem hides the rest behind a fix.""" + problems, _ = gate.environment_verdict( + config(cutoff=None, retention=None, retention_enabled=False), 0.75, REL) + self.assertEqual(len(problems), 3) + + +class TheEnvironmentCorpus(unittest.TestCase): + """The gate reads the environments off the tree, so an added one is checked.""" + + def test_every_shipped_values_file_is_an_environment(self): + envs = gate.environments() + self.assertTrue(envs, "the addon carries no values-.yaml, so this gate " + "would examine no environment") + for env in envs: + with self.subTest(env=env): + self.assertTrue((gate.ADDON / f"values-{env}.yaml").exists()) + + def test_the_shipped_alert_names_the_gauge_the_ingester_sets_every_tick(self): + gate.failures.clear() + threshold = gate.alert_threshold() + self.assertEqual(gate.failures, []) + self.assertIsNotNone(threshold) + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/tests/test_platform_crs.py b/scripts/tests/test_platform_crs.py new file mode 100644 index 0000000..b2af7de --- /dev/null +++ b/scripts/tests/test_platform_crs.py @@ -0,0 +1,275 @@ +"""Unit tests for the platform-CR admissibility walk. + +The gate fetches the operator chart to read the CRD schemas the catalog's own +custom resources are checked against, so the positive-control sweep exempts it. +That leaves the walk — which is the whole gate — resting on nobody having read it. + +Each rule below is one the API server enforces and no other gate here does. +kubeconform skips these kinds entirely; kustomize builds them happily; ArgoCD +applies them and the Application simply never reaches Healthy. A rule that +under-matches produces the same output as a compliant manifest, and there is no +second signal anywhere to contradict it. + +The schemas here are written inline rather than fetched, so the assertions are +about the walk rather than about whatever the pinned chart currently ships. +""" + +from __future__ import annotations + +import unittest + +from gateloader import load + +gate = load("check-platform-crs") + +KIND = "Platform" +SOURCE = "addons/ai-platform/agent-platform/base/platform.yaml" + + +def problems_for(value, schema): + found: list[str] = [] + gate.walk(value, schema, "", KIND, SOURCE, found) + return found + + +class RequiredProperties(unittest.TestCase): + """`required` is a rejection of the whole object, but not always.""" + + SCHEMA = { + "type": "object", + "required": ["tenant", "persona"], + "properties": {"tenant": {"type": "string"}, + "persona": {"type": "string", "default": "generic"}}, + } + + def test_a_missing_required_property_with_no_default_is_a_rejection(self): + found = problems_for({"persona": "ops"}, self.SCHEMA) + self.assertEqual(len(found), 1) + self.assertIn("tenant: Required value", found[0]) + + def test_a_missing_required_property_that_declares_a_default_is_admitted(self): + """Structural-schema defaulting runs BEFORE validation. + + Reading `required` alone reports a manifest as refused over a property + the API server fills in itself. + """ + self.assertEqual(problems_for({"tenant": "platform-ops"}, self.SCHEMA), []) + + def test_a_complete_object_has_no_problems(self): + self.assertEqual( + problems_for({"tenant": "t", "persona": "ops"}, self.SCHEMA), []) + + +class ExcessProperties(unittest.TestCase): + """A property the CRD does not carry is pruned — it never reaches a cluster.""" + + SCHEMA = {"type": "object", "properties": {"tenant": {"type": "string"}}} + + def test_an_unknown_property_is_reported_as_pruned(self): + found = problems_for({"tenant": "t", "budgetUsd": "50"}, self.SCHEMA) + self.assertEqual(len(found), 1) + self.assertIn("is pruned at admission", found[0]) + + def test_a_schema_preserving_unknown_fields_prunes_nothing(self): + schema = dict(self.SCHEMA, **{"x-kubernetes-preserve-unknown-fields": True}) + self.assertEqual(problems_for({"anything": 1}, schema), []) + + def test_a_schema_declaring_no_properties_is_not_walked(self): + """The API server does not prune where the schema declines to describe.""" + self.assertEqual(problems_for({"anything": 1}, {"type": "object"}), []) + + +class DeclaredTypes(unittest.TestCase): + """YAML decides the type for you, and the CRD does not negotiate.""" + + def test_a_fractional_quantity_left_unquoted_is_rejected(self): + """Kubernetes serialises fractional quantities as strings. + + `minACU: 0.5` reads as a YAML float, every property is present, none is + excess, and the API server refuses the whole object. + """ + found = problems_for({"minACU": 0.5}, + {"type": "object", + "properties": {"minACU": {"type": "string"}}}) + self.assertEqual(len(found), 1) + self.assertIn("must be of type string", found[0]) + self.assertIn("quote it", found[0]) + + def test_a_quoted_quantity_is_accepted(self): + self.assertEqual( + problems_for({"minACU": "0.5"}, + {"type": "object", + "properties": {"minACU": {"type": "string"}}}), []) + + def test_a_bool_does_not_satisfy_integer(self): + """In Python a bool IS an int, so a naive check passes an unquoted `true`.""" + found = problems_for({"replicas": True}, + {"type": "object", + "properties": {"replicas": {"type": "integer"}}}) + self.assertEqual(len(found), 1) + self.assertIn("is a boolean and the CRD declares integer", found[0]) + + def test_an_integer_satisfies_number(self): + self.assertEqual( + problems_for({"ratio": 3}, + {"type": "object", + "properties": {"ratio": {"type": "number"}}}), []) + + def test_a_float_does_not_satisfy_integer(self): + found = problems_for({"replicas": 1.5}, + {"type": "object", + "properties": {"replicas": {"type": "integer"}}}) + self.assertEqual(len(found), 1) + self.assertIn("must be of type integer", found[0]) + + def test_an_unset_value_is_not_a_type_error(self): + self.assertEqual( + problems_for({"tenant": None}, + {"type": "object", + "properties": {"tenant": {"type": "string"}}}), []) + + def test_a_schema_declaring_no_type_checks_nothing(self): + self.assertEqual( + problems_for({"x": 1}, {"type": "object", "properties": {"x": {}}}), []) + + def test_the_reported_type_name_is_the_one_the_api_server_prints(self): + for value, name in ((True, "boolean"), ("s", "string"), (1, "integer"), + (1.5, "number"), ({}, "object"), ([], "array")): + with self.subTest(value=value): + self.assertEqual(gate._json_type_name(value), name) + + +class ListIdentity(unittest.TestCase): + """x-kubernetes-list-type is a validation rule, not documentation. + + A duplicate is a hard rejection of the whole object — not a warning, not a + merge of the two entries. + """ + + MAP_SCHEMA = { + "type": "object", + "properties": {"datastores": { + "type": "array", + "x-kubernetes-list-type": "map", + "x-kubernetes-list-map-keys": ["name"], + "items": {"type": "object", "properties": { + "name": {"type": "string"}, "kind": {"type": "string"}}}, + }}, + } + + def test_two_entries_sharing_the_map_key_are_rejected(self): + found = problems_for( + {"datastores": [{"name": "main"}, {"name": "main", "kind": "cache"}]}, + self.MAP_SCHEMA) + self.assertEqual(len(found), 1) + self.assertIn("Duplicate value", found[0]) + self.assertIn("repeats name=main", found[0]) + + def test_distinct_map_keys_are_accepted(self): + self.assertEqual(problems_for( + {"datastores": [{"name": "main"}, {"name": "cache"}]}, + self.MAP_SCHEMA), []) + + def test_an_absent_key_participates_in_the_identity(self): + """Reading only the fields that happen to be set misses this pair.""" + found = problems_for( + {"datastores": [{"kind": "a"}, {"kind": "b"}]}, self.MAP_SCHEMA) + self.assertEqual(len(found), 1) + self.assertIn("", found[0]) + + def test_a_set_list_identifies_a_scalar_by_itself(self): + schema = {"type": "object", "properties": {"zones": { + "type": "array", "x-kubernetes-list-type": "set", + "items": {"type": "string"}}}} + found = problems_for({"zones": ["us-east-1a", "us-east-1a"]}, schema) + self.assertEqual(len(found), 1) + self.assertIn("repeats value=us-east-1a", found[0]) + + def test_an_atomic_list_imposes_no_uniqueness(self): + schema = {"type": "object", "properties": {"args": { + "type": "array", "x-kubernetes-list-type": "atomic", + "items": {"type": "string"}}}} + self.assertEqual(problems_for({"args": ["-v", "-v"]}, schema), []) + + def test_a_list_with_no_declared_type_imposes_no_uniqueness(self): + schema = {"type": "object", "properties": {"args": { + "type": "array", "items": {"type": "string"}}}} + self.assertEqual(problems_for({"args": ["-v", "-v"]}, schema), []) + + def test_a_map_list_declaring_no_keys_imposes_no_uniqueness(self): + schema = {"type": "object", "properties": {"x": { + "type": "array", "x-kubernetes-list-type": "map", + "items": {"type": "object"}}}} + self.assertEqual(problems_for({"x": [{"a": 1}, {"a": 1}]}, schema), []) + + def test_three_entries_sharing_an_identity_report_each_repeat(self): + found = problems_for( + {"datastores": [{"name": "m"}, {"name": "m"}, {"name": "m"}]}, + self.MAP_SCHEMA) + self.assertEqual(len(found), 2) + + +class TheWalkDescends(unittest.TestCase): + """A rule that fires only at the top level checks the least interesting layer.""" + + SCHEMA = { + "type": "object", + "properties": {"identity": { + "type": "object", + "required": ["allowedModels"], + "properties": {"allowedModels": { + "type": "array", + "x-kubernetes-list-type": "set", + "items": {"type": "string"}}}, + }}, + } + + def test_a_nested_required_property_is_reached(self): + found = problems_for({"identity": {}}, self.SCHEMA) + self.assertIn("`identity.allowedModels: Required value`", found[0]) + + def test_a_duplicate_inside_a_nested_list_is_reached(self): + found = problems_for( + {"identity": {"allowedModels": ["us.anthropic.claude-sonnet-5", + "us.anthropic.claude-sonnet-5"]}}, + self.SCHEMA) + self.assertEqual(len(found), 1) + self.assertIn(".identity.allowedModels", found[0]) + + def test_a_type_error_inside_an_array_item_names_its_index(self): + schema = {"type": "object", "properties": {"routes": { + "type": "array", + "items": {"type": "object", + "properties": {"rateLimit": {"type": "integer"}}}}}} + found = problems_for({"routes": [{"rateLimit": 60}, {"rateLimit": "60"}]}, + schema) + self.assertEqual(len(found), 1) + self.assertIn(".routes[1].rateLimit", found[0]) + + def test_a_mistyped_container_is_reported_before_it_is_descended_into(self): + schema = {"type": "object", "properties": {"identity": { + "type": "object", "properties": {"allowedModels": {"type": "array"}}}}} + found = problems_for({"identity": ["a", "b"]}, schema) + self.assertEqual(len(found), 1) + self.assertIn("must be of type object", found[0]) + + def test_a_non_mapping_schema_is_not_walked(self): + self.assertEqual(problems_for({"a": 1}, "not-a-schema"), []) + + +class TheManifestCorpus(unittest.TestCase): + """A walk over no manifests reports the same as a walk over compliant ones.""" + + def test_the_corpus_is_not_empty(self): + self.assertTrue(list(gate.manifests())) + + def test_no_skipped_directory_swallows_the_catalog_crs(self): + found = [f for f in gate.manifests() + if "kind: Platform" in f.read_text(encoding="utf-8", errors="replace")] + self.assertTrue(found, + "no manifest in the walk declares a Platform, so this gate " + "examined none of the CRs it exists to check") + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/tests/test_policy_admission.py b/scripts/tests/test_policy_admission.py new file mode 100644 index 0000000..28b0a32 --- /dev/null +++ b/scripts/tests/test_policy_admission.py @@ -0,0 +1,338 @@ +"""Unit tests for the policy-admission gate's verdict. + +This gate answers the question the fleet cannot answer anywhere else: would any +addon be DENIED at admission on a cluster running the Enforce-tier policies? It +reaches a chart registry, so the positive-control sweep exempts it — which left +the whole verdict resting on nobody having read it. + +The verdict is where it matters. Every branch below is one where a wrong answer +prints the sentence a healthy fleet prints: a canary the run never evaluated, a +runtime pod nothing matched, a namespace receiving workloads that no exclusion +list covers. Each of those produces an empty result set, and an empty result set +is what "no addon was denied" looks like from the outside. + +Everything here is offline. Rendering and the kyverno invocation are not +exercised from here; the report they produce is supplied directly, in the shapes +kyverno emits. +""" + +from __future__ import annotations + +import contextlib +import io +import subprocess +import unittest + +import yaml +from gateloader import load + +gate = load("check-policy-admission") + + +def quietly(fn, *args, **kwargs): + """Run a gate function that narrates to stdout, returning its verdict.""" + buf = io.StringIO() + with contextlib.redirect_stdout(buf): + result = fn(*args, **kwargs) + return result, buf.getvalue() + + +def result(policy="best-practices", rule="require-probes", verdict="fail", + namespace="workloads", name="app", kind="Deployment"): + """One entry as kyverno writes it into a policy report.""" + return { + "policy": policy, + "rule": rule, + "result": verdict, + "resources": [{"kind": kind, "namespace": namespace, "name": name}], + } + + +def canary(rule="require-probes", verdict="fail"): + return result(rule=rule, verdict=verdict, + namespace=gate.CANARY_NAMESPACE, name=gate.CANARY_NAME) + + +def runtime_pod(verdict="pass", rule="require-probes"): + return result(rule=rule, verdict=verdict, kind="Pod", + namespace=gate.RUNTIME_POD_NAMESPACE, name=gate.RUNTIME_POD_NAME) + + +def completed(returncode=0, stdout="", stderr=""): + return subprocess.CompletedProcess(["kyverno"], returncode, stdout, stderr) + + +class NormalisingARender(unittest.TestCase): + """`_prepare` decides which manifests reach kyverno and in which namespace. + + Both halves change the verdict silently. A workload left unqualified misses + the namespace exclusion and is reported as a denial that would not happen; a + manifest dropped here is one the run cannot deny at all. + """ + + def prepare(self, docs, namespace="monitoring"): + landed: set[str] = set() + rendered = "\n---\n".join(yaml.safe_dump(d, sort_keys=False) for d in docs) + out = list(yaml.safe_load_all(gate._prepare(rendered, namespace, landed))) + return [d for d in out if d], landed + + def test_a_workload_without_a_namespace_is_qualified_with_the_release_one(self): + docs, landed = self.prepare([{"apiVersion": "apps/v1", "kind": "Deployment", + "metadata": {"name": "loki"}}]) + self.assertEqual(docs[0]["metadata"]["namespace"], "monitoring") + self.assertEqual(landed, {"monitoring"}) + + def test_a_namespace_the_chart_sets_is_left_alone_and_is_what_lands(self): + """The coverage claim is about where pods END UP, not where helm aimed.""" + docs, landed = self.prepare([{"apiVersion": "apps/v1", "kind": "DaemonSet", + "metadata": {"name": "agent", + "namespace": "kube-system"}}]) + self.assertEqual(docs[0]["metadata"]["namespace"], "kube-system") + self.assertEqual(landed, {"kube-system"}, + "a chart placing a workload outside the release namespace " + "must be recorded where it lands, or the exclusion " + "coverage check compares against the wrong namespace") + + def test_a_non_workload_kind_is_neither_qualified_nor_recorded(self): + docs, landed = self.prepare([{"apiVersion": "v1", "kind": "ConfigMap", + "metadata": {"name": "cfg"}}]) + self.assertNotIn("namespace", docs[0]["metadata"]) + self.assertEqual(landed, set()) + + def test_a_helm_test_hook_is_dropped(self): + """ArgoCD skips test hooks, so a chart's test Pod never hits admission.""" + docs, landed = self.prepare([ + {"apiVersion": "v1", "kind": "Pod", + "metadata": {"name": "probe", + "annotations": {"helm.sh/hook": "test"}}}, + {"apiVersion": "v1", "kind": "Pod", "metadata": {"name": "real"}}, + ]) + self.assertEqual([d["metadata"]["name"] for d in docs], ["real"]) + self.assertEqual(landed, {"monitoring"}) + + def test_a_kindless_document_is_dropped(self): + """helm v4 prints OCI pull progress ahead of the manifests on a fresh pull. + + Those parse as kind-less mappings. Written back, kyverno rejects the whole + resource file with "Object 'Kind' is missing" and the run evaluates + nothing — which reports as no addon being denied. + """ + docs, _ = self.prepare([{"Pulled": "ghcr.io/x/y:1.0"}, + {"apiVersion": "v1", "kind": "Pod", + "metadata": {"name": "real"}}]) + self.assertEqual([d["metadata"]["name"] for d in docs], ["real"]) + + def test_every_workload_kind_is_qualified(self): + for kind in sorted(gate.WORKLOAD_KINDS): + with self.subTest(kind=kind): + docs, landed = self.prepare( + [{"apiVersion": "v1", "kind": kind, "metadata": {"name": "w"}}]) + self.assertEqual(docs[0]["metadata"]["namespace"], "monitoring") + self.assertEqual(landed, {"monitoring"}) + + +class RuleNamesFromAReport(unittest.TestCase): + """Kyverno never names a rule the way the policy file does. + + The canary is checked against the rule set read off the policy YAML, so a + prefix left on turns every autogen variant into an unexercised rule and the + run fails for a reason that is not true. + """ + + def test_a_plain_rule_is_unchanged(self): + self.assertEqual(gate._rule_key(result(rule="require-probes")), + "best-practices/require-probes") + + def test_the_pod_controller_prefix_is_stripped(self): + self.assertEqual(gate._rule_key(result(rule="autogen-require-probes")), + "best-practices/require-probes") + + def test_the_cronjob_prefix_is_stripped_whole(self): + """`autogen-cronjob-` must be tried before `autogen-`. + + Stripping the shorter prefix first leaves `cronjob-require-probes`, which + matches no rule in any policy file. + """ + self.assertEqual(gate._rule_key(result(rule="autogen-cronjob-require-probes")), + "best-practices/require-probes") + + def test_a_rule_whose_name_begins_with_autogen_text_is_not_truncated(self): + self.assertEqual(gate._rule_key(result(rule="autogenerate-labels")), + "best-practices/autogenerate-labels") + + +class IdentifyingTheProbes(unittest.TestCase): + """Namespace AND name. Either alone matches an addon that happens to share it.""" + + def test_the_canary_is_matched_on_both_coordinates(self): + self.assertTrue(gate._is_canary(canary())) + self.assertFalse(gate._is_canary( + result(namespace="workloads", name=gate.CANARY_NAME))) + self.assertFalse(gate._is_canary( + result(namespace=gate.CANARY_NAMESPACE, name="something-else"))) + + def test_the_runtime_pod_is_matched_on_both_coordinates(self): + self.assertTrue(gate._is_runtime_pod(runtime_pod())) + self.assertFalse(gate._is_runtime_pod( + result(namespace="workloads", name=gate.RUNTIME_POD_NAME))) + + def test_a_result_carrying_no_resources_matches_neither(self): + bare = {"policy": "p", "rule": "r", "result": "fail"} + self.assertFalse(gate._is_canary(bare)) + self.assertFalse(gate._is_runtime_pod(bare)) + + +class TheRuntimePodVerdictIsTwoSided(unittest.TestCase): + """An Argo workflow step pod carries no probes and cannot. + + "No rule denied it" and "no rule ever saw it" are the same empty result set, + and the second is the failure being guarded against. + """ + + def test_a_pod_no_rule_evaluated_is_not_admitted(self): + verdict, out = quietly(gate.judge_runtime_pod, [result(verdict="pass")]) + self.assertFalse(verdict) + self.assertIn("evaluated by NO rule", out) + + def test_a_pod_at_least_one_rule_passed_and_none_denied_is_admitted(self): + verdict, _ = quietly(gate.judge_runtime_pod, + [runtime_pod("pass"), runtime_pod("pass", "require-labels")]) + self.assertTrue(verdict) + + def test_a_denial_alongside_a_pass_still_fails(self): + for denial in ("fail", "warn", "error"): + with self.subTest(denial=denial): + verdict, out = quietly( + gate.judge_runtime_pod, + [runtime_pod("pass"), runtime_pod(denial, "require-non-root")]) + self.assertFalse(verdict) + self.assertIn("denied by 1 rule", out) + + def test_a_skip_is_neither_a_pass_nor_a_denial(self): + """A skipped rule proves nothing was evaluated, so the pass is missing.""" + verdict, out = quietly(gate.judge_runtime_pod, [runtime_pod("skip")]) + self.assertFalse(verdict) + self.assertIn("evaluated by NO rule", out) + + +class TheRunVerdict(unittest.TestCase): + """`judge` fails on both sides: what was denied, and what was never evaluated.""" + + RULES = {"best-practices/require-probes", "best-practices/require-labels"} + + def clean_report(self, canary_result="fail"): + return { + "summary": {"pass": 1, "fail": 2}, + "results": [canary("require-probes", canary_result), + canary("require-labels", canary_result), + runtime_pod("pass")], + } + + def judge(self, report, proc=None, canary_result="fail"): + return quietly(gate.judge, "test", report, proc or completed(), + self.RULES, canary_result) + + def test_a_run_that_exercised_every_rule_and_flagged_nothing_passes(self): + verdict, out = self.judge(self.clean_report()) + self.assertTrue(verdict) + self.assertIn("canary failed by all 2 rules", out) + self.assertIn("no addon flagged", out) + + def test_an_empty_report_at_exit_zero_is_named_as_matching_nothing(self): + """Kyverno emits nothing and exits 0 when its rules matched no resource. + + That is indistinguishable from success to an exit-code check, which is + why the verdict is read out of the report instead. + """ + verdict, out = self.judge(None, completed(returncode=0, stdout="")) + self.assertFalse(verdict) + self.assertIn("matched NO resource", out) + + def test_an_unparseable_report_reports_the_child_failure(self): + verdict, out = self.judge(None, completed(returncode=2, stderr="panic: boom")) + self.assertFalse(verdict) + self.assertIn("panic: boom", out) + + def test_a_rule_that_never_reported_the_canary_fails_the_run(self): + report = self.clean_report() + report["results"] = [r for r in report["results"] + if r["rule"] != "require-labels"] + verdict, out = self.judge(report) + self.assertFalse(verdict) + self.assertIn("require-labels", out) + self.assertIn("did not evaluate", out) + + def test_a_canary_reported_at_the_wrong_result_does_not_count_as_exercised(self): + """An Enforce run that only warns has not enforced anything.""" + verdict, out = self.judge(self.clean_report(canary_result="warn"), + canary_result="fail") + self.assertFalse(verdict) + self.assertIn("did not evaluate", out) + + def test_a_flagged_addon_fails_the_run(self): + report = self.clean_report() + report["results"].append(result(namespace="monitoring", name="loki", + verdict="fail")) + verdict, out = self.judge(report) + self.assertFalse(verdict) + self.assertIn("monitoring/loki", out) + + def test_the_runtime_pod_is_not_counted_among_flagged_addons(self): + """Its verdict is judged separately and names what actually broke.""" + report = self.clean_report() + report["results"].append(runtime_pod("fail", "require-non-root")) + verdict, out = self.judge(report) + self.assertFalse(verdict) + self.assertIn("no addon flagged", out) + self.assertIn("runtime pod was denied", out) + + def test_a_clean_canary_cannot_carry_a_missing_runtime_pod(self): + report = self.clean_report() + report["results"] = [r for r in report["results"] + if not gate._is_runtime_pod(r)] + verdict, out = self.judge(report) + self.assertFalse(verdict) + self.assertIn("evaluated by NO rule", out) + + +class ExclusionCoverage(unittest.TestCase): + """"Uncovered" and "compliant" must stop being the same signal. + + A new addon in a new namespace is a workload the policies WILL evaluate on a + vended cluster. The canary run alone reports it only if it also happens to + violate a rule. + """ + + def test_a_namespace_on_no_exclusion_list_fails(self): + verdict, out = quietly(gate.check_namespace_coverage, + {"monitoring", "brand-new"}, {"monitoring"}) + self.assertFalse(verdict) + self.assertIn("brand-new", out) + + def test_every_landed_namespace_excluded_passes(self): + verdict, _ = quietly(gate.check_namespace_coverage, + {"monitoring"}, {"monitoring", "kube-system"}) + self.assertTrue(verdict) + + def test_an_exclusion_list_wider_than_the_fleet_is_not_this_check(self): + """Parity between the four lists is asserted elsewhere; breadth is not a + failure here — a namespace excluded before its addon arrives is normal.""" + verdict, _ = quietly(gate.check_namespace_coverage, set(), {"monitoring"}) + self.assertTrue(verdict) + + +class TheExclusionCorpusIsReal(unittest.TestCase): + """The four policies the parity check compares must exist to be compared.""" + + def test_each_named_exclusion_policy_is_a_file_in_the_tree(self): + for group, name in gate.EXCLUSION_POLICIES: + path = gate.POLICY_DIR / group / "base" / name + with self.subTest(policy=f"{group}/{name}"): + self.assertTrue(path.exists(), + f"{path} is named as an exclusion-bearing policy " + f"but does not exist, so the parity check compares " + f"three lists and reports agreement") + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/tests/test_render_addons.py b/scripts/tests/test_render_addons.py new file mode 100644 index 0000000..86b00e8 --- /dev/null +++ b/scripts/tests/test_render_addons.py @@ -0,0 +1,222 @@ +"""Unit tests for the render gate's discovery and failure classification. + +render-addons is the corpus every other render-based gate reads: the image-pin +gate and the policy-admission gate both import its `discover()` rather than +re-deriving which chart a path belongs to. A unit it does not find is a unit +three gates never examine, and all three print a count derived from the same +short list — so the omission reports itself as a complete run. + +It templates every chart against a registry, so the positive-control sweep +exempts it. These exercise the two halves that decide what happens without a +network: which units exist, and what a helm failure means. +""" + +from __future__ import annotations + +import unittest + +from gateloader import load + +gate = load("render-addons") + + +def appset(sources, generators=None, destination=None): + doc = {"kind": "ApplicationSet", + "spec": {"template": {"spec": {"sources": sources}}}} + if generators is not None: + doc["spec"]["generators"] = generators + if destination is not None: + doc["spec"]["template"]["spec"]["destination"] = destination + return doc + + +class OciReferences(unittest.TestCase): + """ArgoCD resolves the OCI digest from repoURL, so the shape of it decides. + + Appending the chart name to a repoURL that already ends in it produces + `.../karpenter/karpenter`, which is not a package. + """ + + def unit(self, repo, chart): + return gate.Unit(appset="a.yaml", chart=chart, version="1.0.0", + repo=repo, path="addons/x/y") + + def test_a_repo_url_ending_in_the_chart_name_is_used_as_is(self): + u = self.unit("oci://public.ecr.aws/karpenter/karpenter", "karpenter") + self.assertEqual(u.oci_ref(), "oci://public.ecr.aws/karpenter/karpenter") + + def test_a_repo_url_naming_the_enclosing_namespace_gets_the_chart_appended(self): + u = self.unit("oci://ghcr.io/nanohype/charts", "operator") + self.assertEqual(u.oci_ref(), "oci://ghcr.io/nanohype/charts/operator") + + def test_an_https_repo_is_not_an_oci_one(self): + self.assertFalse(self.unit("https://kyverno.github.io/kyverno", "kyverno").is_oci) + self.assertTrue(self.unit("oci://ghcr.io/x/y", "y").is_oci) + + +class RecognisingATemplatedField(unittest.TestCase): + """A templated `chart` is what says an appset renders many units, not one.""" + + def test_a_go_template_is_a_template(self): + self.assertTrue(gate._is_template("{{ .chart }}")) + self.assertTrue(gate._is_template('{{ index .metadata.labels "environment" }}')) + + def test_a_literal_is_not(self): + self.assertFalse(gate._is_template("kyverno")) + self.assertFalse(gate._is_template("3.8.2")) + + def test_a_non_string_is_not(self): + for value in (None, 3.82, ["{{ .x }}"], {"a": "{{ .x }}"}): + with self.subTest(value=value): + self.assertFalse(gate._is_template(value)) + + +class FindingTheChartSource(unittest.TestCase): + """Multi-source appsets carry a `$values` ref alongside the chart.""" + + def test_the_source_carrying_a_chart_key_is_returned(self): + values_ref = {"repoURL": "{{ .repo }}", "ref": "values"} + chart = {"repoURL": "https://x", "chart": "loki", "targetRevision": "1"} + self.assertIs(gate._chart_source([values_ref, chart]), chart) + + def test_an_appset_with_no_chart_source_is_not_ours(self): + """Kustomize, git-sourced and local-chart appsets render elsewhere.""" + self.assertIsNone(gate._chart_source([{"repoURL": "https://x", "path": "p"}])) + + def test_a_non_mapping_source_is_skipped_rather_than_read(self): + chart = {"repoURL": "https://x", "chart": "loki"} + self.assertIs(gate._chart_source(["not-a-mapping", chart]), chart) + + +class SynthesisingChartParameters(unittest.TestCase): + """ArgoCD injects the real per-cluster value; the render needs a valid one.""" + + def test_a_templated_parameter_takes_its_synthetic_value(self): + params = gate._synth_params( + {"parameters": [{"name": "clusterName", "value": "{{ .name }}"}]}) + self.assertEqual(params, [("clusterName", "ci-cluster")]) + + def test_a_templated_parameter_with_no_synthetic_value_takes_a_placeholder(self): + params = gate._synth_params( + {"parameters": [{"name": "someOther", "value": "{{ .x }}"}]}) + self.assertEqual(params, [("someOther", "ci")]) + + def test_a_literal_parameter_is_passed_through(self): + params = gate._synth_params( + {"parameters": [{"name": "replicaCount", "value": 3}]}) + self.assertEqual(params, [("replicaCount", "3")]) + + def test_a_valueless_parameter_takes_a_synthetic_value(self): + params = gate._synth_params({"parameters": [{"name": "vpcId"}]}) + self.assertEqual(params, [("vpcId", "vpc-00000000000000000")]) + + def test_a_nameless_parameter_is_dropped(self): + self.assertEqual(gate._synth_params({"parameters": [{"value": "x"}]}), []) + + def test_no_parameters_block_yields_none(self): + self.assertEqual(gate._synth_params({}), []) + self.assertEqual(gate._synth_params({"parameters": None}), []) + + +class TheAddonPathFromValueFiles(unittest.TestCase): + """A single-source appset states its addon directory only in `valueFiles`.""" + + def test_the_base_values_path_is_extracted(self): + helm = {"valueFiles": [ + "$values/addons/observability/loki/values.yaml", + "$values/addons/observability/loki/values-production.yaml"]} + self.assertEqual(gate._path_from_valuefiles(helm), + "addons/observability/loki") + + def test_a_per_environment_file_alone_does_not_supply_the_path(self): + """Matching it would yield a directory that ends in the environment name.""" + helm = {"valueFiles": ["$values/addons/x/y/values-production.yaml"]} + self.assertIsNone(gate._path_from_valuefiles(helm)) + + def test_no_value_files_yields_nothing(self): + self.assertIsNone(gate._path_from_valuefiles({})) + self.assertIsNone(gate._path_from_valuefiles({"valueFiles": None})) + + +class UnreachableIsNotAVerdict(unittest.TestCase): + """Exit 1 is a finding about a pin; exit 2 is a fact about the network. + + Collapsing them makes an outage read as a defect in the catalogue, and + trains readers to re-run a red gate rather than read it. + """ + + def test_a_missing_chart_version_is_a_finding_about_this_repo(self): + for err in ('Error: chart "loki" version "9.9.9" not found', + "Error: no chart version found for kyverno-3.99.0", + "Error: failed to fetch ...: 404 Not Found"): + with self.subTest(err=err): + self.assertTrue(gate.registry_answered(err)) + + def test_a_connection_failure_is_a_fact_about_the_network(self): + for err in ("Error: dial tcp 1.2.3.4:443: i/o timeout", + "Get https://x: dial tcp: lookup x: no such host", + "Error: TLS handshake timeout", + "Error: connection refused"): + with self.subTest(err=err): + self.assertFalse(gate.registry_answered(err)) + + def test_a_message_naming_both_is_read_as_unreachable(self): + """An unreachable registry cannot testify about what it holds.""" + self.assertFalse(gate.registry_answered( + "Error: chart not found: dial tcp 1.2.3.4:443: i/o timeout")) + + def test_an_unrecognised_failure_is_not_treated_as_a_finding(self): + self.assertFalse(gate.registry_answered("Error: something entirely new")) + + def test_the_classification_ignores_case(self): + self.assertTrue(gate.registry_answered("Error: NO CHART VERSION FOUND")) + self.assertFalse(gate.registry_answered("Error: I/O Timeout")) + + +class TheDiscoveredCorpus(unittest.TestCase): + """Every unit three gates share comes from here.""" + + @classmethod + def setUpClass(cls): + cls.units = gate.discover() + + def test_the_walk_finds_units(self): + self.assertTrue(self.units, + "discovery matched no ApplicationSet, so this gate and the " + "two that import it would each report a complete run over " + "an empty corpus") + + def test_every_unit_names_a_directory_that_exists(self): + """A typo in an element's `path` removes the addon from three gates.""" + for u in self.units: + with self.subTest(unit=f"{u.appset}:{u.chart}"): + self.assertTrue((gate.REPO_ROOT / u.path).is_dir(), + f"{u.appset} points {u.chart} at '{u.path}', " + f"which is not a directory") + + def test_no_unit_carries_an_unrendered_template(self): + for u in self.units: + with self.subTest(unit=f"{u.appset}:{u.chart}"): + for field in (u.chart, u.version, u.repo, u.path): + self.assertNotIn("{{", field) + + def test_every_unit_carries_a_version(self): + for u in self.units: + with self.subTest(unit=f"{u.appset}:{u.chart}"): + self.assertTrue(u.version.strip()) + + def test_units_are_unique_per_appset_and_chart(self): + """Two units for one chart double-count the corpus the floor is read against.""" + seen = [(u.appset, u.chart, u.path) for u in self.units] + self.assertEqual(len(seen), len(set(seen))) + + def test_every_skipped_chart_is_one_the_corpus_contains(self): + """A skip naming a chart no appset pins excuses nothing and reads as care.""" + charts = {u.chart for u in self.units} + for chart in gate.SKIP_CHARTS: + with self.subTest(chart=chart): + self.assertIn(chart, charts) + + +if __name__ == "__main__": + unittest.main() From 6e3a2936c3c3360c1bfbc2a3c7f6abd06b20e9f5 Mon Sep 17 00:00:00 2001 From: stxkxs Date: Wed, 2 Sep 2026 02:07:29 -0700 Subject: [PATCH 2/7] Commit the harness that proves those tests reject MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous commit's mutation results were produced out of tree, so the note about bytecode caching named a procedure a reader cannot run and the seventeen probes existed nowhere. A proof nobody can re-run is the shape this repo rejects everywhere else. `scripts/tests/reverify-tests.sh` is that procedure, as a sibling to `scripts/tests/reverify-gates.sh`. The existing harness proves a GATE rejects a planted defect; this proves the TESTS reject a reverted behaviour, which a passing suite cannot establish on its own — a test that asserts what it just constructed passes forever, and so does one whose subject was rewritten underneath it. ─────────────────────────── Naming, not just failing ─────────────────────────── Each probe declares the test ids that must fail, and a run that goes red without them is reported as a MISS rather than a pass: MISS classify: find the tag colon anywhere in the reference rejected, but not by test_a_version_tag_is_accepted RESULT pass=16 fail=1 A non-zero exit alone proves the suite noticed something, not that the assertion which noticed is the one describing the behaviour. A mutant caught through an unrelated fixture has demonstrated detection without demonstrating coverage, and the two are indistinguishable from an exit code. The other two ways it can stop meaning anything fail as well: MISS alert: accept a rule keyed on the edge counter the suite passed the mutant RESULT pass=16 fail=1 RESULT pass=4 fail=0 FAIL ran 4 probe(s), under the floor of 17 — this harness planted almost nothing, which is not the same as every test holding. ──────────────────────────────── What it plants ──────────────────────────────── Seventeen probes across the six modules, asserting the clean tree is green before anything is planted and green again after. On the tree as it stands: ── Clean tree: the suite must PASS before anything is planted ── ok all six modules green on the unmodified tree ── One behaviour reverted at a time; the suite must name it ── ok autogen: strip the shorter prefix first FAIL: test_the_cronjob_prefix_is_stripped_whole ok prepare: keep kind-less documents ERROR: test_a_kindless_document_is_dropped ok runtime pod: admit on absence of denials alone FAIL: test_a_pod_no_rule_evaluated_is_not_admitted FAIL: test_a_skip_is_neither_a_pass_nor_a_denial FAIL: test_a_clean_canary_cannot_carry_a_missing_runtime_pod ok judge: count the runtime pod as a foreign addon FAIL: test_the_runtime_pod_is_not_counted_among_flagged_addons ok types: let a bool satisfy integer, as Python does FAIL: test_a_bool_does_not_satisfy_integer ok required: drop the defaulting exemption FAIL: test_a_missing_required_property_that_declares_a_default_is_admitted ok list identity: check only the top level FAIL: test_a_set_list_identifies_a_scalar_by_itself FAIL: test_an_absent_key_participates_in_the_identity FAIL: test_three_entries_sharing_an_identity_report_each_repeat ok variables: match only the braced form FAIL: test_the_bare_form_every_dashboard_here_uses FAIL: test_the_bracket_form FAIL: test_a_builtin_beside_a_real_variable_leaves_the_real_one ok alerts: scan panels flat, skipping collapsed rows FAIL: test_an_alert_inside_a_collapsed_row_is_found FAIL: test_a_top_level_alert_panel_is_found FAIL: test_an_untitled_alert_panel_is_still_reported ok datasources: glob the directory, not the kustomization FAIL: test_a_datasource_file_absent_from_resources_is_not_wired ok helm failure: call every missing-chart message a finding FAIL: test_a_message_naming_both_is_read_as_unreachable ok oci: append the chart name to every repoURL FAIL: test_a_repo_url_ending_in_the_chart_name_is_used_as_is ok addon path: take it from any values file FAIL: test_a_per_environment_file_alone_does_not_supply_the_path ok classify: find the tag colon anywhere in the reference FAIL: test_a_registry_port_is_not_read_as_a_tag ok exemptions: stop re-checking them against the render FAIL: test_an_exemption_the_fleet_no_longer_renders_mutably_fails FAIL: test_an_exemption_for_an_image_now_pinned_by_tag_fails ok cutoff: let the alert fire AT it rather than before FAIL: test_a_warning_at_the_cutoff_leaves_no_window ok alert: accept a rule keyed on the edge counter FAIL: test_a_rule_querying_the_edge_counter_fails_twice FAIL: test_a_rule_querying_both_still_fails_on_the_edge_counter ── Restored tree must be green again ── ok all six modules green again RESULT pass=17 fail=0 ──────────────────────────── Two properties it needs ──────────────────────────── Bytecode is cleared and disabled around every probe. The cache keys on mtime and size, so a restore and the next mutation inside the same second can hand the run the previous mutant's module — and the suite then reports the previous mutant's failing test against this mutant's label. That is a green line naming the wrong assertion, which is worse than a red one, and the naming requirement above is what surfaces it. Backups live outside the scratch directory the cleanup removes, and restoration runs on INT and TERM as well as EXIT. The probes plant defects in tracked gate scripts; between planting and restoring the backup is the only copy of the original, and `scripts/tests/reverify-gates.sh` records what happens when that copy sits inside the directory the teardown deletes. Co-authored-by: stxkxsbot <275011021+stxkxsbot@users.noreply.github.com> --- scripts/tests/reverify-tests.sh | 381 ++++++++++++++++++++++++++++++++ 1 file changed, 381 insertions(+) create mode 100755 scripts/tests/reverify-tests.sh diff --git a/scripts/tests/reverify-tests.sh b/scripts/tests/reverify-tests.sh new file mode 100755 index 0000000..6e766f5 --- /dev/null +++ b/scripts/tests/reverify-tests.sh @@ -0,0 +1,381 @@ +#!/usr/bin/env bash +# Revert one gate behaviour at a time and require the suite to NAME it. +# +# scripts/tests/reverify-gates.sh proves a GATE rejects a planted defect. This +# proves the TESTS reject a reverted behaviour, which is the other half and the +# one a passing suite cannot supply on its own: a test that asserts what it just +# constructed passes forever, and so does a test whose subject was quietly +# rewritten underneath it. +# +# Each probe below names the test ids that must fail. Requiring a non-zero exit +# alone is not enough — it proves the suite noticed something, not that the +# assertion which noticed is the one that describes the behaviour. A suite can +# catch a mutant by accident through an unrelated fixture, and then the mutant +# has demonstrated detection without demonstrating coverage. +set -uo pipefail +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +cd "$ROOT" +SP="$(mktemp -d)" + +# Backups live outside $SP, for the reason reverify-gates.sh records: between +# planting and restoring, the backup is the only copy of the original, and +# keeping it in the directory the cleanup removes means an interrupt destroys +# the means of restoration. +BK="$(mktemp -d)" +OUT="$SP/rt.out" + +MUT_FILES=() + +_slot() { printf '%s' "$1" | tr '/' '_'; } + +mut() { + local f="$1" s + s="$(_slot "$f")" + [ -e "$BK/$s" ] || cp "$f" "$BK/$s" + case " ${MUT_FILES[*]:-} " in + *" $f "*) ;; + *) MUT_FILES+=("$f") ;; + esac +} + +res() { + local f="$1" s + s="$(_slot "$f")" + [ -e "$BK/$s" ] && cp "$BK/$s" "$f" + return 0 +} + +restore_all() { + [ "${#MUT_FILES[@]}" -eq 0 ] && return 0 + local f + for f in "${MUT_FILES[@]}"; do res "$f"; done +} + +cleanup() { restore_all; rm -rf "$SP" "$BK"; } +trap cleanup EXIT +trap 'cleanup; exit 130' INT +trap 'cleanup; exit 143' TERM + +pass=0; fail=0 + +# Every module this harness plants against. Named rather than discovered: a +# module that stops being loaded would otherwise drop out of the clean-tree and +# restored-tree checks without either of them failing. +ALL_MODULES="test_policy_admission test_platform_crs test_dashboards \ + test_render_addons test_image_pins test_log_volume_budget" + +# Bytecode caching keys on (mtime, size). A restore and the next mutation inside +# the same second can hand the run the PREVIOUS mutant's module, and the suite +# then reports the previous mutant's failing test against this mutant's label — +# a green line naming the wrong assertion, which is worse than a red one. +_suite() { + rm -rf "$ROOT/scripts/__pycache__" "$ROOT/scripts/tests/__pycache__" + ( cd "$ROOT/scripts/tests" \ + && PYTHONDONTWRITEBYTECODE=1 python3 -m unittest "$@" ) >"$OUT" 2>&1 +} + +# rejects