diff --git a/cli/ponens/registry.py b/cli/ponens/registry.py index 03f81a8..dbe6052 100644 --- a/cli/ponens/registry.py +++ b/cli/ponens/registry.py @@ -21,6 +21,7 @@ from .formatting import ( bold, gray, red, green, yellow, blue, cyan, magenta, underline, table, heading, ) +from .policy_compiler import CheckError, check_policy DEFAULT_GALLERY_URL = "https://ponens.dev/gallery/policies" @@ -450,6 +451,62 @@ def cmd_policies_add(args): print(gray(f" run: ponens trace check {path}")) +def cmd_policies_lint(args): + """Lint policy definitions locally: required fields + formula syntax. + + Runs the same oracle `trace check` applies before evaluating + (policy_compiler.check_policy), so a policy that lints valid cannot be + marked syntax-invalid by check later. With --json, exits 0 whenever + linting ran — per-policy verdicts are in the records; without it, exits 1 + if any policy is invalid. An unreadable or malformed input file exits 1. + """ + try: + with open(args.policy_file) as f: + policy_data = json.load(f) + except (OSError, json.JSONDecodeError) as e: + print(f"Error: cannot read policy file: {e}", file=sys.stderr) + return 1 + policies = policy_data if isinstance(policy_data, list) else policy_data.get("policies", []) + if not isinstance(policies, list) or not all(isinstance(p, dict) for p in policies): + print('Error: policy file must be a JSON array of policy objects, or {"policies": [...]}', + file=sys.stderr) + return 1 + + as_json = getattr(args, "json", False) + records = [] + invalid = 0 + for p in policies: + pid = p.get("policy_id", p.get("name", "?")) + if "name" not in p: + # check_policy indexes policy['name'] before it can report it missing; + # produce the message it would have produced. + errors, warnings = [CheckError("Missing required field 'name'", pid)], [] + else: + try: + _, errors, warnings = check_policy(p) + except Exception as e: # a policy the checker cannot inspect is invalid, not a crash + errors, warnings = [CheckError(f"Policy could not be checked: {e!r}", pid)], [] + record = {"policy_id": pid, "status": "invalid" if errors else "valid"} + if errors: + record["errors"] = [{"message": e.message, "path": e.path} for e in errors] + invalid += 1 + if warnings: + record["warnings"] = [{"message": w.message, "path": w.path} for w in warnings] + records.append(record) + if not as_json: + print(f" {'INVALID' if errors else 'OK '} {pid}") + for e in errors: + print(f" {e.message}") + for w in warnings: + print(f" warning: {w.message}") + + if as_json: + print(json.dumps(records, indent=2, ensure_ascii=False)) + return 0 + print(f"\n {len(records)} policies linted, {invalid} invalid") + return 1 if invalid else 0 + + # ---------------------------------------------------------------------------- # Commands — unified search (policies, packs, organizations) # ---------------------------------------------------------------------------- @@ -611,3 +668,9 @@ def register(subparsers): p.add_argument("--into", required=True, help="Path to the trace JSON file") p.add_argument("--refresh", action="store_true", help="Force re-fetch, bypassing the cache") p.set_defaults(func=cmd_policies_add) + + p = pol_sub.add_parser("lint", help="Lint policy definitions locally (required fields + formula syntax)") + p.add_argument("policy_file", help='Policy JSON file: an array of policies, or {"policies": [...]}') + p.add_argument("--json", action="store_true", + help="Emit per-policy verdicts as JSON (machine-readable; exits 0 whenever linting ran)") + p.set_defaults(func=cmd_policies_lint) diff --git a/cli/tests/unit/test_policies_lint.py b/cli/tests/unit/test_policies_lint.py new file mode 100644 index 0000000..605735d --- /dev/null +++ b/cli/tests/unit/test_policies_lint.py @@ -0,0 +1,94 @@ +"""`ponens policies lint` — local policy validation (required fields + formula +syntax), the same oracle `trace check` applies before evaluating.""" + +import json +from argparse import Namespace + +from ponens.registry import cmd_policies_lint + +VALID = { + "policy_id": "tests_before_commit", + "name": "tests_before_commit", + "severity": "error", + "scope": "trace", + "kind": "temporal", + "formula": "G(GitCommit → P(RunTests ∧ completed))", +} + + +def _lint(tmp_path, payload, as_json=True): + f = tmp_path / "policies.json" + f.write_text(json.dumps(payload, ensure_ascii=False)) + return cmd_policies_lint(Namespace(policy_file=str(f), json=as_json)) + + +def _records(capsys): + return json.loads(capsys.readouterr().out) + + +def test_valid_policy_lints_valid(tmp_path, capsys): + assert _lint(tmp_path, [VALID]) == 0 + (rec,) = _records(capsys) + assert rec == {"policy_id": "tests_before_commit", "status": "valid"} + + +def test_wrapped_policies_key_is_accepted(tmp_path, capsys): + assert _lint(tmp_path, {"policies": [VALID]}) == 0 + (rec,) = _records(capsys) + assert rec["status"] == "valid" + + +def test_missing_scope_is_invalid_with_the_check_message(tmp_path, capsys): + p = {k: v for k, v in VALID.items() if k != "scope"} + assert _lint(tmp_path, [p]) == 0 # --json: verdicts in records, not the exit code + (rec,) = _records(capsys) + assert rec["status"] == "invalid" + assert any("Missing required field 'scope'" in e["message"] for e in rec["errors"]) + + +def test_formula_that_does_not_parse_is_invalid(tmp_path, capsys): + p = dict(VALID, formula="G(GitCommit →") + assert _lint(tmp_path, [p]) == 0 + (rec,) = _records(capsys) + assert rec["status"] == "invalid" + assert rec["errors"] + + +def test_missing_name_is_invalid_not_a_crash(tmp_path, capsys): + p = {k: v for k, v in VALID.items() if k != "name"} + assert _lint(tmp_path, [p]) == 0 + (rec,) = _records(capsys) + assert rec["policy_id"] == "tests_before_commit" + assert rec["status"] == "invalid" + assert any("Missing required field 'name'" in e["message"] for e in rec["errors"]) + + +def test_human_mode_exits_nonzero_on_invalid(tmp_path, capsys): + p = {k: v for k, v in VALID.items() if k != "scope"} + assert _lint(tmp_path, [p], as_json=False) == 1 + out = capsys.readouterr().out + assert "INVALID" in out and "1 invalid" in out + + +def test_human_mode_exits_zero_when_all_valid(tmp_path, capsys): + assert _lint(tmp_path, [VALID], as_json=False) == 0 + assert "0 invalid" in capsys.readouterr().out + + +def test_malformed_file_is_refused(tmp_path, capsys): + f = tmp_path / "nope.json" + f.write_text("{not json") + assert cmd_policies_lint(Namespace(policy_file=str(f), json=True)) == 1 + captured = capsys.readouterr() + assert captured.out == "" # no records on refusal + assert "cannot read policy file" in captured.err + + +def test_non_object_entries_are_refused(tmp_path, capsys): + assert _lint(tmp_path, ["just a string"]) == 1 + assert "array of policy objects" in capsys.readouterr().err + + +def test_empty_list_lints_clean(tmp_path, capsys): + assert _lint(tmp_path, []) == 0 + assert _records(capsys) == []