Jon Chery
ac18c98385
core/policy_engine.py: PolicyEngine Protocol (PEP 544, runtime_checkable)
+ PolicyEngineRegistry (selects from config.json.policy.engine) + NullEngine
fallback (NULL_ENGINE_INACTIVE when policy key absent).
adapters/kyverno-json/: KyvernoJsonEngine — shells to , translates
native output → list[dict] PCR records (engine: "kyverno", ruleId KJ_ prefix,
severity via nova.cloudinit.dev/severity annotation, default info).
is_configured() guards on → KJ_ENGINE_NOT_CONFIGURED SKIPPED PCR
(distinct from NullEngine). Defensive parsing (malformed → error PCR).
config.json: new object {engine: kyverno-json, policy_root}.
scripts/install-kyverno-json.sh: go install kj@latest (D-115).
CI (.gitea + .github): install Go + kj for policy-engine tests (best-effort;
tests skip when kj absent).
tests: 24 pass, 2 skip (kj not installed). 132 existing tests unchanged.
NullEngine satisfies PolicyEngine Protocol (G-Q8a — proves swap boundary).
---ci---
project: acdl
phase: 1
milestone: v1.25
status: execute
phase_role: execution
requirements:
covered: [REQ-291, REQ-292, REQ-293, REQ-294, REQ-308, REQ-309]
partial: []
---/ci---
125 lines
4.7 KiB
Python
125 lines
4.7 KiB
Python
"""Tests for core/policy_engine.py (REQ-308, v1.25).
|
|
|
|
Protocol conformance, registry selection, NullEngine fallback,
|
|
unknown-engine KeyError, and the NullEngine-satisfies-Protocol
|
|
assertion (G-Q8a — proves the swap boundary is real without
|
|
implementing OPA).
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
import core.policy_engine as pe
|
|
|
|
|
|
class TestPolicyEngineProtocol:
|
|
def test_null_engine_satisfies_protocol(self):
|
|
# G-Q8a: NullEngine satisfies the PolicyEngine Protocol — proves
|
|
# the swap boundary is real (a second engine implements it).
|
|
eng = pe.NullEngine()
|
|
assert isinstance(eng, pe.PolicyEngine)
|
|
|
|
def test_null_engine_is_configured_false(self):
|
|
assert pe.NullEngine().is_configured() is False
|
|
|
|
def test_null_engine_evaluate_returns_skipped(self):
|
|
out = pe.NullEngine().evaluate({}, Path("/tmp"), "cid-123")
|
|
assert len(out) == 1
|
|
pcr = out[0]
|
|
assert pcr["ruleId"] == "NULL_ENGINE_INACTIVE"
|
|
assert pcr["result"] == "skipped"
|
|
assert pcr["engine"] == "kyverno"
|
|
assert pcr["contractId"] == "cid-123"
|
|
|
|
def test_null_engine_severity_is_info(self):
|
|
out = pe.NullEngine().evaluate({}, Path("/tmp"), "cid")
|
|
assert out[0]["severity"] == "info"
|
|
|
|
|
|
class TestRegistry:
|
|
def test_register_and_get(self, tmp_path, monkeypatch):
|
|
# Register a stub engine and verify get_engine() returns it.
|
|
class StubEngine:
|
|
name = "stub"
|
|
|
|
def is_configured(self) -> bool:
|
|
return True
|
|
|
|
def evaluate(self, payload, policy_dir, contract_id):
|
|
return [{"contractId": contract_id, "engine": "kyverno",
|
|
"ruleId": "STUB", "result": "pass", "severity": "info",
|
|
"message": "", "evaluatedAt": "t", "resourceRef": "",
|
|
"evidence": {}}]
|
|
|
|
pe._REGISTRY.clear()
|
|
pe.register("stub", StubEngine)
|
|
monkeypatch.setattr(pe, "_load_config_policy", lambda: {"engine": "stub"})
|
|
eng = pe.get_engine()
|
|
assert eng.name == "stub"
|
|
pe._REGISTRY.clear()
|
|
pe._autoload_kyverno_json()
|
|
|
|
def test_unknown_engine_raises_keyerror(self, monkeypatch):
|
|
pe._REGISTRY.clear()
|
|
monkeypatch.setattr(pe, "_load_config_policy",
|
|
lambda: {"engine": "nonexistent"})
|
|
with pytest.raises(KeyError, match="Unknown policy engine"):
|
|
pe.get_engine()
|
|
pe._autoload_kyverno_json()
|
|
|
|
def test_null_engine_fallback_when_policy_key_absent(self, monkeypatch):
|
|
# G-Q4: policy key absent → NullEngine (distinct from kj-not-configured).
|
|
monkeypatch.setattr(pe, "_load_config_policy", lambda: None)
|
|
eng = pe.get_engine()
|
|
assert isinstance(eng, pe.NullEngine)
|
|
assert eng.is_configured() is False
|
|
|
|
def test_kyverno_json_registered_via_autoload(self):
|
|
# The autoload should register kyverno-json if the adapter file exists.
|
|
pe._autoload_kyverno_json()
|
|
assert "kyverno-json" in pe._REGISTRY or len(pe._REGISTRY) == 0
|
|
|
|
|
|
class TestConfigPolicyLoad:
|
|
def test_load_config_policy_returns_dict(self):
|
|
out = pe._load_config_policy()
|
|
if out is not None:
|
|
assert "engine" in out
|
|
assert out["engine"] == "kyverno-json"
|
|
|
|
def test_get_policy_root_is_path(self):
|
|
root = pe.get_policy_root()
|
|
assert isinstance(root, Path)
|
|
assert root.name == "policies" or str(root).endswith("policies")
|
|
|
|
|
|
class TestKjNotConfiguredPath:
|
|
"""G-Q4: when policy key is present but kj is absent, the engine
|
|
returns KJ_ENGINE_NOT_CONFIGURED (distinct from NullEngine's
|
|
NULL_ENGINE_INACTIVE)."""
|
|
|
|
def test_kj_not_configured_returns_distinct_ruleid(self, monkeypatch):
|
|
# Force the registry to return KyvernoJsonEngine, then mock
|
|
# `which kj` to return None.
|
|
pe._autoload_kyverno_json()
|
|
if "kyverno-json" not in pe._REGISTRY:
|
|
pytest.skip("kyverno-json adapter not loadable in this env")
|
|
monkeypatch.setattr(pe, "_load_config_policy",
|
|
lambda: {"engine": "kyverno-json"})
|
|
eng = pe.get_engine()
|
|
# Mock is_configured → False
|
|
with mock.patch.object(eng, "is_configured", return_value=False):
|
|
out = eng.evaluate({}, Path("/tmp"), "cid-456")
|
|
assert len(out) == 1
|
|
assert out[0]["ruleId"] == "KJ_ENGINE_NOT_CONFIGURED"
|
|
assert out[0]["result"] == "skipped"
|
|
assert out[0]["contractId"] == "cid-456"
|
|
# Distinct from NullEngine
|
|
assert out[0]["ruleId"] != "NULL_ENGINE_INACTIVE" |