#!/usr/bin/env python3 """Lint a martech integration design: flow inventory, field map and test plan. Usage: python3 scripts/lint_integration.py INVENTORY.csv [FIELD_MAP.csv] [TEST_PLAN.csv] [--quiet] [--json] Inputs (columns and allowed values: references/csv-formats.md): INVENTORY.csv one row per flow, header of assets/flow-inventory-template.csv FIELD_MAP.csv one row per destination field and direction, assets/field-map-template.csv TEST_PLAN.csv one row per test case, assets/test-plan-template.csv Rules (meaning and fixes: references/contract-rules.md, delivery-reliability.md, operations.md and test-plan.md): CT01-CT06 completeness and allowed values MS01-MS02 method misuse ID01-ID04 identity keys IP01-IP03 idempotency and ledger retention RT01-RT02 retry layers, Retry-After unit WH01 webhook acknowledgement time RM01-RM02 removals OP01-OP05 operations FO01-FO09 field ownership (needs FIELD_MAP) TP01-TP04 test coverage (needs TEST_PLAN) Notes: inbound_webhook is a required cell (CT01). A blank retry_layers is RT01 WARN, because blank means unknown; write none for no retries. ID04 warns when an update-only flow matches on a field that does not enforce uniqueness. IP02 warns when a flow that creates records or has a side effect keys effects on a delivery ID alone, outside an event pipeline. RM01 also fails removal_policy not_applicable on a current-membership contract. FO01 warns when two flows write a field under version or effective-time rules, and FO02 fails when other writers are listed with no conflict rule, and warns when they are listed under a version or effective-time rule they may not compare. Output: "LEVEL RULE target: message" per finding in inventory order, then rule ID, then PASS lines per flow and family, then a summary line. --quiet leaves out the PASS lines (they are still counted). --json prints JSON. Exit 1 when any FAIL, 2 on bad input (missing file or column), 0 otherwise. Standard library only. Reads local files, calls no network, changes nothing. """ import argparse import csv import json import os import re import sys from collections import defaultdict INVENTORY_COLUMNS = ( "flow_id source_system source_object destination_system destination_object trigger " "inbound_webhook method method_rationale tier decision business_owner technical_owner " "on_call freshness_target_minutes volume_records_per_day peak_per_minute batch_size " "match_key match_key_kind destination_unique creates_records side_effect idempotency_key " "removal_policy membership_contract retry_layers retry_after_unit sender_timeout_seconds " "ack_p99_seconds ledger_retention_days max_replay_window_days exception_queue " "alert_oldest_age_minutes reconciliation_cadence credential_owner api_version runbook" ).split() REQUIRED_CELLS = ( "flow_id source_system source_object destination_system destination_object trigger " "inbound_webhook method tier decision business_owner technical_owner freshness_target_minutes match_key " "match_key_kind creates_records side_effect removal_policy membership_contract " "exception_queue reconciliation_cadence credential_owner" ).split() FIELD_MAP_COLUMNS = ("flow_id destination_system destination_object destination_field owner writer direction " "conflict_rule blank_policy type allowed_values unit_or_currency decision_driving pii other_writers").split() FIELD_MAP_CELLS = "flow_id destination_system destination_object destination_field owner writer type".split() FIELD_MAP_WRITER_CELLS = "direction conflict_rule blank_policy decision_driving pii".split() TEST_PLAN_COLUMNS = "test_id flow_id case environment evidence status approver".split() TEST_PLAN_CELLS = "test_id flow_id case environment status".split() YES_NO = {"yes", "no"} ALLOWED = { "trigger": {"event", "schedule", "manual", "file"}, "method": {"native", "ipaas", "event_pipeline", "reverse_etl", "custom", "file", "mcp"}, "tier": {"1", "2", "3"}, "credential_owner": {"service", "person"}, "match_key_kind": {"source_id", "email", "name", "domain", "destination_id", "composite"}, "inbound_webhook": YES_NO, "destination_unique": YES_NO, "creates_records": YES_NO, "exception_queue": YES_NO, "decision_driving": YES_NO, "pii": YES_NO, "side_effect": {"none", "task", "message", "order", "conversion", "audience", "other"}, "removal_policy": {"none", "clear_fields", "delete", "archive", "exit_audience", "not_applicable"}, "membership_contract": {"current", "add_only", "not_applicable"}, "writer": {"this_flow", "read_only"}, "retry_after_unit": {"seconds", "milliseconds", "http_date", "none"}, "direction": {"one_way", "two_way"}, "reconciliation_cadence": {"daily", "weekly", "monthly", "none"}, "conflict_rule": {"none", "version", "effective_time", "owner_wins", "manual_review"}, "blank_policy": {"preserve", "clear_on_explicit", "clear", "skip_and_flag"}, "type": {"string", "number", "integer", "boolean", "date", "datetime", "enum", "currency", "id"}, "environment": {"sandbox", "staging", "production_approved"}, "status": {"planned", "pass", "fail", "blocked"}, } INVENTORY_ENUMS = [c for c in INVENTORY_COLUMNS if c in ALLOWED] FIELD_MAP_ENUMS = "writer direction conflict_rule blank_policy type decision_driving pii".split() NUMERIC = ("freshness_target_minutes volume_records_per_day peak_per_minute batch_size sender_timeout_seconds " "ack_p99_seconds ledger_retention_days max_replay_window_days alert_oldest_age_minutes").split() CORE_CASES = ("new_record update_owned non_owned_untouched blank_semantics duplicate_delivery out_of_order " "unknown_outcome_retry rate_limited ambiguous_match environment_check").split() CONDITIONAL_CASES = ("signature_invalid unknown_event_type malformed_payload partial_batch_failure unmatched_record removal " "backfill_no_outreach replay_no_duplicate credential_revoked consent_change loop_check").split() MANUAL_CASES = "merge delete enum_unmapped schema_change cross_account_isolation permission_check late_change".split() KNOWN_CASES = set(CORE_CASES + CONDITIONAL_CASES + MANUAL_CASES) ACCOUNT_OBJECT = re.compile(r"company|account|organization|organisation|workspace|deal|opportunity", re.I) BAD_KEY = re.compile(r"attempt|delivery|received|arrival|\bnow\b|random|retry|uuid4", re.I) DELIVERY_KEY = re.compile(r"(event|message|webhook|delivery)_?id", re.I) FAMILIES = ["CT", "MS", "ID", "IP", "RT", "WH", "RM", "OP", "FO", "TP"] PASS_TEXT = { "CT": "required cells, owners and allowed values checked", "MS": "method fits the trigger and freshness target", "ID": "match key, uniqueness and creation rule checked", "IP": "keys and retention checked", "RT": "retry layers and Retry-After unit checked", "WH": "acknowledgement within half the sender timeout", "RM": "removal policy matches the membership contract", "FO": "field ownership rows checked", "OP": "exception queue, alerting, credentials, versions and reconciliation checked", "TP": "mandatory test cases present", } class InputError(Exception): pass def read_table(path, required, label): """Read a CSV into a list of dicts with lowercase headers and trimmed cells.""" if not os.path.isfile(path): raise InputError(f"{label} file not found: {os.path.basename(path)}") try: with open(path, encoding="utf-8-sig", newline="") as handle: reader = csv.reader(handle) header = [h.strip().lower() for h in next(reader, [])] missing = [c for c in required if c not in header] if missing: raise InputError(f"{label} {os.path.basename(path)} is missing column(s): {', '.join(missing)}") rows = [] for raw in reader: if not any(cell.strip() for cell in raw): continue rows.append({name: (raw[i].strip() if i < len(raw) else "") for i, name in enumerate(header)}) return rows except UnicodeDecodeError: raise InputError(f"{label} {os.path.basename(path)} is not UTF-8 text") def num(value): try: return float(value) except (TypeError, ValueError): return None # blank or text def yes(value): return value.strip().lower() == "yes" def parse_layers(text): """Return (product, attempts list, malformed parts) for 'name:attempts;...'.""" if not text or text.strip().lower() == "none": return 1, [], [] product, attempts, bad = 1, [], [] for part in text.split(";"): part = part.strip() if not part: continue name, _, count = part.rpartition(":") if not name or not count.strip().isdigit() or int(count) < 1: bad.append(part) continue attempts.append(int(count)) product *= int(count) return product, attempts, bad class Report: def __init__(self): self.items = [] def add(self, level, rule, target, message, flow_index): self.items.append({"level": level, "rule": rule, "target": target, "message": message, "flow_index": flow_index, "seq": len(self.items)}) def mandatory_cases(flow, map_rows): cases = list(CORE_CASES) if yes(flow["inbound_webhook"]): cases += ["signature_invalid", "unknown_event_type", "malformed_payload"] if (num(flow["batch_size"]) or 1) > 1: cases.append("partial_batch_failure") if yes(flow["creates_records"]): cases.append("unmatched_record") if flow["removal_policy"].lower() not in ("", "none", "not_applicable") or \ flow["membership_contract"].lower() == "current": cases.append("removal") if flow["tier"] == "1": cases += ["backfill_no_outreach", "replay_no_duplicate", "credential_revoked"] if flow["side_effect"].lower() == "message" or any(yes(r["pii"]) for r in map_rows): cases.append("consent_change") if any(r["direction"].lower() == "two_way" or r["other_writers"] for r in map_rows): cases.append("loop_check") return cases def lint_flow(flow, index, report): """Apply the CT, MS, ID, IP, RT, WH, RM and OP rules to one inventory row.""" target = f"flow {flow['flow_id'] or '(row ' + str(index + 1) + ')'}" evaluated = {"CT", "ID", "IP", "RT", "RM", "OP"} def add(level, rule, message): report.add(level, rule, target, message, index) method, tier = flow["method"].lower(), flow["tier"] fresh = num(flow["freshness_target_minutes"]) blank = [c for c in REQUIRED_CELLS if not flow[c]] if blank: add("FAIL", "CT01", "required cells blank: " + ", ".join(blank)) if tier == "1": missing = [c for c in ("on_call", "runbook") if not flow[c]] if missing: add("FAIL", "CT03", f"tier 1 flow without {' and '.join(missing)}") elif tier == "2" and not flow["runbook"]: add("WARN", "CT03", "tier 2 flow without a runbook") if method and method != "native" and not flow["method_rationale"]: add("WARN", "CT04", f"method {method} with blank method_rationale: record the clause each simpler method failed") if flow["freshness_target_minutes"] and (fresh is None or fresh <= 0): add("WARN", "CT05", "freshness_target_minutes must be a positive number of minutes") for col in INVENTORY_ENUMS: value = flow[col].lower() if value and value not in ALLOWED[col]: add("WARN", "CT06", f"{col} '{flow[col]}' is not one of: {', '.join(sorted(ALLOWED[col]))}") for col in NUMERIC: if flow[col] and num(flow[col]) is None: add("WARN", "CT06", f"{col} '{flow[col]}' is not a number") if method in ("mcp", "reverse_etl"): evaluated.add("MS") if method == "mcp" and flow["trigger"].lower() in ("event", "schedule"): add("FAIL", "MS01", "MCP access used as the scheduler or delivery path; MCP has no scheduler, retry ledger or delivery guarantee") if method == "reverse_etl" and fresh is not None and fresh < 15: add("WARN", "MS02", f"reverse ETL with a {fresh:g}-minute freshness target; below 15 minutes prefer an event-driven path") kind, dobj = flow["match_key_kind"].lower(), flow["destination_object"] creates = yes(flow["creates_records"]) if kind in ("name", "domain"): add("FAIL", "ID01", f"match key '{flow['match_key']}' is a {kind}; names collide and domains are shared, so match on a stable source ID in a unique destination field") elif kind == "email" and ACCOUNT_OBJECT.search(dobj): add("FAIL", "ID01", f"email matches account-level object '{dobj}'; never match accounts on email") elif kind == "email": add("WARN", "ID01", f"email matches person object '{dobj}'; emails change and move between people, keep a crosswalk to the destination ID") if creates and not yes(flow["destination_unique"]): add("FAIL", "ID02", "creates records but the destination match field does not enforce uniqueness") if kind == "destination_id" and creates: add("WARN", "ID03", "matches on the destination ID while creating records; only the writer that owns creation can know that ID") if not creates and not yes(flow["destination_unique"]): add("WARN", "ID04", "update-only flow matching on a field that does not enforce uniqueness: two destination records " "can match; state how the flow detects two matches and stops") key, side = flow["idempotency_key"], flow["side_effect"].lower() if (creates or side not in ("", "none")) and not key: add("FAIL", "IP01", f"creates records or has side effect '{side or 'unknown'}' with no idempotency key") if key and BAD_KEY.search(key): add("FAIL", "IP02", f"idempotency key '{key}' uses attempt or arrival data; build it from the flow, source object and business version") elif key and method != "event_pipeline" and (creates or side not in ("", "none")): parts = [p.strip() for p in key.split("+")] parts = [p for p in parts if p and p.lower() not in ("flow", "flow_id", flow["flow_id"].lower())] if parts and all(DELIVERY_KEY.search(p) for p in parts): add("WARN", "IP02", f"idempotency key '{key}' is a delivery key only; key effects on flow + source object + business version") ledger, window = num(flow["ledger_retention_days"]), num(flow["max_replay_window_days"]) if ledger is not None and window is not None and ledger < window: add("FAIL", "IP03", f"ledger kept {ledger:g} day{'' if ledger == 1 else 's'}, shorter than the {window:g}-day replay window") elif (ledger is None or window is None) and (creates or side not in ("", "none")): add("WARN", "IP03", "ledger_retention_days or max_replay_window_days blank; retention cannot be compared with replay windows") product, attempts, bad = parse_layers(flow["retry_layers"]) if not flow["retry_layers"]: add("WARN", "RT01", "retry_layers blank: list every layer that retries (SDK, HTTP client, iPaaS step, platform " "autoreplay) or write none") for part in bad: add("WARN", "CT06", f"retry_layers part '{part}' is not name:attempts; ignored") if product > 10: add("FAIL", "RT01", f"nested attempts multiply to {product} ({flow['retry_layers']}); keep the product at 10 or below by retrying at one layer") elif sum(1 for a in attempts if a > 1) > 1: add("WARN", "RT01", f"more than one layer retries ({flow['retry_layers']}, product {product}); retry at one layer") if method in ("custom", "ipaas") and not flow["retry_after_unit"]: add("WARN", "RT02", "retry_after_unit blank; record the vendor's Retry-After unit (seconds, milliseconds or HTTP date)") if yes(flow["inbound_webhook"]): evaluated.add("WH") timeout, ack = num(flow["sender_timeout_seconds"]), num(flow["ack_p99_seconds"]) if timeout is None or ack is None: add("WARN", "WH01", "sender_timeout_seconds or ack_p99_seconds blank; acknowledgement time cannot be checked") elif ack > timeout / 2: add("FAIL", "WH01", f"acknowledgement p99 {ack:g} s is above half the sender's {timeout:g} s timeout; store durably, then acknowledge, then process") membership, removal = flow["membership_contract"].lower(), flow["removal_policy"].lower() if membership == "current" and removal in ("none", "not_applicable"): add("FAIL", "RM01", f"current-membership contract with removal_policy {removal}; an add-only sync cannot remove people who no longer qualify") if removal == "delete": add("WARN", "RM02", "removal_policy delete: needs a minimum-population guard (10% or 1,000 records) and approval") queue = yes(flow["exception_queue"]) if not queue and tier == "1": add("FAIL", "OP01", "tier 1 flow without an exception queue") elif not queue and tier == "2": add("WARN", "OP01", "tier 2 flow without an exception queue") if tier in ("1", "2"): alert = num(flow["alert_oldest_age_minutes"]) if alert is None: add("FAIL", "OP02", f"tier {tier} flow with no oldest-unprocessed-item alert") elif fresh is not None and alert > 2 * fresh: add("WARN", "OP02", f"alert after {alert:g} minutes against a {fresh:g}-minute freshness target (more than twice)") if flow["credential_owner"].lower() == "person": add("FAIL", "OP03", "credential owned by a person; use a service identity in a business-owned account") if method in ("custom", "ipaas") and not flow["api_version"]: add("WARN", "OP04", "api_version blank; pin the version and track deprecations") cadence = flow["reconciliation_cadence"].lower() if tier == "1" and cadence in ("none", "monthly"): add("FAIL", "OP05", f"tier 1 flow reconciled {cadence}; reconcile tier 1 flows daily, weekly at the least") elif tier == "2" and cadence == "none": add("WARN", "OP05", "tier 2 flow with no reconciliation") return evaluated def lint_field_map(rows, flows, report, flow_index): """Apply FO01 to FO09 and field-map completeness checks.""" by_field = defaultdict(list) for n, row in enumerate(rows, 1): fid = row["flow_id"] index = flow_index.get(fid, len(flows)) label = f"{row['destination_system']} {row['destination_object']}.{row['destination_field']}" target = f"field-map row {n} ({fid} {label})" writes = row["writer"].lower() == "this_flow" decision = yes(row["decision_driving"]) def add(level, rule, message): report.add(level, rule, target, message, index) blank = [c for c in FIELD_MAP_CELLS if not row[c]] if writes: blank += [c for c in FIELD_MAP_WRITER_CELLS if not row[c]] if blank: add("FAIL", "CT01", "required cells blank: " + ", ".join(blank)) for col in FIELD_MAP_ENUMS: value = row[col].lower() if value and value not in ALLOWED[col]: add("WARN", "CT06", f"{col} '{row[col]}' is not one of: {', '.join(sorted(ALLOWED[col]))}") if fid not in flow_index: add("FAIL", "FO08", f"flow_id '{fid}' is not in the inventory") conflict, direction = row["conflict_rule"].lower(), row["direction"].lower() if writes: by_field[(row["destination_system"].lower(), row["destination_object"].lower(), row["destination_field"].lower())].append((n, row)) if row["other_writers"] and conflict in ("", "none"): add("FAIL", "FO02", f"other writers ({row['other_writers']}) and no conflict rule") elif row["other_writers"] and conflict in ("version", "effective_time"): add("WARN", "FO02", f"other writers under a {conflict} rule: {row['other_writers']}. A version rule only " "binds writers that compare it; remove the other writer's write access or show it " "compares the same version") if row["blank_policy"].lower() == "clear" and decision: add("FAIL", "FO04", "blank_policy clear on a decision-driving field; use skip_and_flag or clear_on_explicit") if direction == "two_way" and conflict in ("", "none"): add("FAIL", "FO03", "two-way field with no conflict rule") elif direction == "two_way" and conflict == "owner_wins" and decision: add("WARN", "FO03", "two-way decision-driving field resolved by a global winner; set the owner and direction per field") if row["type"].lower() == "enum" and not row["allowed_values"]: add("FAIL", "FO05", "enum field with no allowed_values; unknown values need a quarantine rule") if row["type"].lower() == "currency" and not row["unit_or_currency"]: add("FAIL" if decision else "WARN", "FO06", "currency field with no unit_or_currency") if yes(row["pii"]): add("WARN", "FO07", "personal data: confirm the consent basis, minimise, and keep it out of logs and alerts") if writes and direction == "one_way" and fid in flow_index: owner, source = row["owner"].lower(), flows[flow_index[fid]]["source_system"].lower() if owner and source and owner not in source and source not in owner: add("WARN", "FO09", f"the flow writes a field owned by {row['owner']}, not by its source {flows[flow_index[fid]]['source_system']}") for (system, obj, field), entries in by_field.items(): writers = sorted({r["flow_id"] for _, r in entries}, key=lambda f: flow_index.get(f, len(flows))) safe = all(r["conflict_rule"].lower() in ("version", "effective_time") for _, r in entries) if len(writers) > 1: first = entries[0][1] label = f"{first['destination_system']} {first['destination_object']}.{first['destination_field']}" rows_text = ", ".join(str(n) for n, _ in entries) where = flow_index.get(writers[0], len(flows)) if safe: report.add("WARN", "FO01", f"field {label}", f"written by {' and '.join(writers)} (rows {rows_text}) under version or " "effective-time rules; confirm every writer compares the same version field", where) else: report.add("FAIL", "FO01", f"field {label}", f"written by {' and '.join(writers)} (rows {rows_text}) without version or effective-time conflict rules on every row", where) def lint_tests(rows, flows, map_by_flow, report, flow_index): """Apply TP01 to TP04 and test-plan completeness checks.""" cases_by_flow = defaultdict(set) for row in rows: target = f"test {row['test_id'] or '?'} ({row['flow_id']} {row['case']})" index = flow_index.get(row["flow_id"], len(flows)) cases_by_flow[row["flow_id"]].add(row["case"].lower()) blank = [c for c in TEST_PLAN_CELLS if not row[c]] if blank: report.add("FAIL", "CT01", target, "required cells blank: " + ", ".join(blank), index) for col in ("environment", "status"): value = row[col].lower() if value and value not in ALLOWED[col]: report.add("WARN", "CT06", target, f"{col} '{row[col]}' is not one of: {', '.join(sorted(ALLOWED[col]))}", index) if row["case"] and row["case"].lower() not in KNOWN_CASES: report.add("WARN", "CT06", target, f"case '{row['case']}' is not in the case catalogue (references/test-plan.md)", index) if row["flow_id"] not in flow_index: report.add("WARN", "CT06", target, f"flow_id '{row['flow_id']}' is not in the inventory", index) if row["environment"].lower() == "production_approved" and not row["approver"]: report.add("FAIL", "TP02", target, "production test with no named approver", index) if row["status"].lower() in ("fail", "blocked"): report.add("WARN", "TP03", target, f"status {row['status'].lower()}", index) if not row["evidence"]: report.add("WARN", "TP04", target, f"evidence blank (status {row['status'].lower() or 'blank'})", index) for index, flow in enumerate(flows): fid = flow["flow_id"] present = cases_by_flow.get(fid, set()) if not present: report.add("FAIL", "TP01", f"flow {fid}", "no test cases", index) continue missing = [c for c in mandatory_cases(flow, map_by_flow.get(fid, [])) if c not in present] if missing: report.add("FAIL", "TP01", f"flow {fid}", "missing mandatory cases: " + ", ".join(missing), index) def main(argv=None): parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("inventory") parser.add_argument("field_map", nargs="?") parser.add_argument("test_plan", nargs="?") parser.add_argument("--quiet", action="store_true", help="leave PASS lines out of the printed report") parser.add_argument("--json", action="store_true", help="print findings as JSON") args = parser.parse_args(argv) try: flows = read_table(args.inventory, INVENTORY_COLUMNS, "Flow inventory") fmap = read_table(args.field_map, FIELD_MAP_COLUMNS, "Field map") if args.field_map else None tests = read_table(args.test_plan, TEST_PLAN_COLUMNS, "Test plan") if args.test_plan else None except InputError as exc: print(f"error: {exc}", file=sys.stderr) return 2 if not flows: print("error: the flow inventory has no data rows", file=sys.stderr) return 2 report, flow_index, counts = Report(), {}, defaultdict(list) for index, flow in enumerate(flows): flow_index.setdefault(flow["flow_id"], index) counts[flow["flow_id"]].append(index + 1) evaluated = [lint_flow(flow, index, report) for index, flow in enumerate(flows)] for fid, rows in counts.items(): if fid and len(rows) > 1: report.add("FAIL", "CT02", f"flow {fid}", f"flow_id appears {len(rows)} times (rows {', '.join(map(str, rows))})", flow_index[fid]) map_by_flow = defaultdict(list) not_checked = [] if fmap is not None: for row in fmap: map_by_flow[row["flow_id"]].append(row) lint_field_map(fmap, flows, report, flow_index) for index, flow in enumerate(flows): if map_by_flow.get(flow["flow_id"]): evaluated[index].add("FO") elif flow["method"].lower() in ("native", "ipaas", "reverse_etl", "custom"): report.add("WARN", "FO08", f"flow {flow['flow_id']}", f"method {flow['method'].lower()} with no field-map rows", index) evaluated[index].add("FO") else: not_checked.append("FO01-FO09: no field map given") if tests is not None: lint_tests(tests, flows, map_by_flow, report, flow_index) for item in evaluated: item.add("TP") else: not_checked.append("TP01-TP04: no test plan given") findings = sorted(report.items, key=lambda f: (f["flow_index"], f["rule"], f["seq"])) passes = [] for index, flow in enumerate(flows): dirty = {f["rule"][:2] for f in findings if f["flow_index"] == index} for family in FAMILIES: if family in evaluated[index] and family not in dirty: passes.append({"level": "PASS", "rule": family, "target": f"flow {flow['flow_id']}", "message": PASS_TEXT[family], "flow_index": index}) ordered = [] for index in range(len(flows) + 1): ordered += [f for f in findings if f["flow_index"] == index] ordered += [p for p in passes if p["flow_index"] == index] total = {level: sum(1 for f in ordered if f["level"] == level) for level in ("FAIL", "WARN", "PASS")} summary = f"{total['FAIL']} FAIL, {total['WARN']} WARN, {total['PASS']} PASS" if not_checked: summary += f", {len(not_checked)} NOT CHECKED" summary += ". Nothing was changed." tail = ("Not checked by this script: method fitness beyond MS01 and MS02, vendor limits, actual " "configuration, runtime behaviour, signature schemes, vendor key windows.") if args.json: print(json.dumps({"findings": [{k: f[k] for k in ("level", "rule", "target", "message")} for f in ordered], "not_checked": not_checked, "summary": total}, indent=2)) else: parts = [f"{len(flows)} flow{'' if len(flows) == 1 else 's'} in {os.path.basename(args.inventory)}"] if fmap is not None: parts.append(f"{len(fmap)} field-map rows in {os.path.basename(args.field_map)}") if tests is not None: parts.append(f"{len(tests)} test cases in {os.path.basename(args.test_plan)}") print("Integration lint: " + ", ".join(parts)) for f in ordered: if not (args.quiet and f["level"] == "PASS"): print(f"{f['level']} {f['rule']} {f['target']}: {f['message']}") for item in not_checked: print(f"NOT CHECKED {item}") print(summary) print(tail) return 1 if total["FAIL"] else 0 if __name__ == "__main__": sys.exit(main())