acdl/tests/test_kyverno_json_engine.py

"""Tests for adapters/kyverno-json/kyverno_json_engine.py (REQ-309, v1.25).

PCR schema validity (jsonschema validation), defensive parsing
(malformed output → error PCR, never exception), is_configured()
guard, severity annotation reading (G-Q10a), and pytest.skip when
kj is absent.
"""

import json
import os
import sys
from pathlib import Path
from unittest import mock

import jsonschema
import pytest

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

# Load the engine module by file path (the dir has a hyphen).
import importlib.util
_ENGINE_PATH = Path(__file__).resolve().parent.parent / "adapters" / "kyverno-json" / "kyverno_json_engine.py"
_spec = importlib.util.spec_from_file_location("kyverno_json_engine", _ENGINE_PATH)
_mod = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_mod)
KyvernoJsonEngine = _mod.KyvernoJsonEngine
_to_pcr = _mod._to_pcr
_load_policy_severities = _mod._load_policy_severities

PCR_SCHEMA_PATH = Path(__file__).resolve().parent.parent / "schemas" / "policy_check_result.schema.json"


def _load_pcr_schema():
    with open(PCR_SCHEMA_PATH, "r", encoding="utf-8") as fh:
        return json.load(fh)


PCR_SCHEMA = _load_pcr_schema()


def _kj_installed() -> bool:
    """Return True if the kj binary is on PATH."""
    return _mod._which_kj() is not None


def _smoke_policy_dir() -> Path:
    return Path(__file__).resolve().parent.parent / "adapters" / "kyverno-json" / "policies"


class TestToPcr:
    def test_pass_entry(self):
        entry = {"policy": "require-contract-id", "rule": "require-id",
                 "result": "pass", "message": "ok", "resource": "res-1"}
        pcr = _to_pcr(entry, "cid", "high")
        assert pcr["contractId"] == "cid"
        assert pcr["engine"] == "kyverno"
        assert pcr["ruleId"] == "KJ_require-contract-id/require-id"
        assert pcr["result"] == "pass"
        assert pcr["severity"] == "high"
        assert pcr["resourceRef"] == "res-1"

    def test_fail_entry(self):
        entry = {"policy": "forbid-public-ingress", "rule": "no-public",
                 "result": "fail", "message": "public ingress not allowed",
                 "resource": "s3/x"}
        pcr = _to_pcr(entry, "cid", "critical")
        assert pcr["result"] == "fail"
        assert pcr["severity"] == "critical"
        assert pcr["message"] == "public ingress not allowed"

    def test_skip_entry(self):
        entry = {"policy": "p", "rule": "r", "result": "skip"}
        pcr = _to_pcr(entry, "cid", "info")
        assert pcr["result"] == "skipped"

    def test_unknown_result_becomes_error(self):
        entry = {"policy": "p", "rule": "r", "result": "garbled"}
        pcr = _to_pcr(entry, "cid", "info")
        assert pcr["result"] == "error"

    def test_pcr_validates_against_schema(self):
        entry = {"policy": "p", "rule": "r", "result": "pass",
                 "message": "ok", "resource": "r"}
        pcr = _to_pcr(entry, "cid-uuid", "medium")
        jsonschema.validate(pcr, PCR_SCHEMA)


class TestSeverityAnnotation:
    """G-Q10a: severity is read from the policy's metadata.annotation."""

    def test_policy_with_severity_annotation(self, tmp_path):
        policy = {
            "apiVersion": "json.kyverno.io/v1alpha1",
            "kind": "ValidatingPolicy",
            "metadata": {
                "name": "test-sev",
                "annotations": {"nova.cloudinit.dev/severity": "high"},
            },
            "spec": {"rules": [{"name": "r", "validate": {"assert": {"all": []}}}]},
        }
        p = tmp_path / "test-sev.json"
        p.write_text(json.dumps(policy))
        sevs = _load_policy_severities(tmp_path)
        assert sevs.get("test-sev") == "high"

    def test_policy_without_severity_defaults_info(self, tmp_path):
        policy = {
            "apiVersion": "json.kyverno.io/v1alpha1",
            "kind": "ValidatingPolicy",
            "metadata": {"name": "no-sev"},
            "spec": {"rules": [{"name": "r", "validate": {"assert": {"all": []}}}]},
        }
        p = tmp_path / "no-sev.json"
        p.write_text(json.dumps(policy))
        sevs = _load_policy_severities(tmp_path)
        assert sevs.get("no-sev") == "info"

    def test_underscore_files_skipped(self, tmp_path):
        # _smoke.json starts with _ — should be skipped.
        (tmp_path / "_smoke.json").write_text("{}")
        sevs = _load_policy_severities(tmp_path)
        assert sevs == {}


class TestIsConfigured:
    def test_is_configured_returns_bool(self):
        eng = KyvernoJsonEngine()
        assert isinstance(eng.is_configured(), bool)

    def test_is_configured_false_when_kj_absent(self, monkeypatch):
        monkeypatch.setattr(_mod, "_which_kj", lambda: None)
        eng = KyvernoJsonEngine()
        assert eng.is_configured() is False


class TestEvaluateNotConfigured:
    """When kj is absent, evaluate() returns KJ_ENGINE_NOT_CONFIGURED."""

    def test_evaluate_returns_skipped_when_not_configured(self, monkeypatch):
        monkeypatch.setattr(_mod, "_which_kj", lambda: None)
        eng = KyvernoJsonEngine()
        out = eng.evaluate({"id": "x"}, Path("/tmp/policies"), "cid-1")
        assert len(out) == 1
        assert out[0]["ruleId"] == "KJ_ENGINE_NOT_CONFIGURED"
        assert out[0]["result"] == "skipped"
        jsonschema.validate(out[0], PCR_SCHEMA)


class TestEvaluateWithKj:
    """Tests that run the real kj binary. Skip when kj is not installed."""

    @pytest.fixture(autouse=True)
    def _require_kj(self):
        if not _kj_installed():
            pytest.skip("kj not installed (scripts/install-kyverno-json.sh)")

    def test_smoke_policy_round_trip(self, tmp_path):
        eng = KyvernoJsonEngine()
        if not eng.is_configured():
            pytest.skip("kj not configured")
        # Use the real smoke policy dir.
        out = eng.evaluate({"id": "msvc"}, _smoke_policy_dir(), "cid-smoke")
        assert isinstance(out, list)
        assert len(out) >= 1
        for pcr in out:
            jsonschema.validate(pcr, PCR_SCHEMA)
            assert pcr["engine"] == "kyverno"
            assert pcr["contractId"] == "cid-smoke"

    def test_no_results_returns_pass(self, tmp_path):
        # An empty policy dir → no results → KJ_NO_RESULTS pass PCR.
        eng = KyvernoJsonEngine()
        empty_dir = tmp_path / "empty"
        empty_dir.mkdir()
        out = eng.evaluate({"id": "x"}, empty_dir, "cid-empty")
        assert len(out) == 1
        assert out[0]["ruleId"] == "KJ_NO_RESULTS"
        assert out[0]["result"] == "pass"


class TestDefensiveParsing:
    """Malformed kyverno-json output → error PCR, never exception."""

    def test_malformed_output_produces_error_pcr(self, monkeypatch):
        eng = KyvernoJsonEngine()
        # Mock is_configured → True, then mock subprocess to return
        # garbage output.
        monkeypatch.setattr(_mod, "_which_kj", lambda: "/fake/kj")
        monkeypatch.setattr(eng, "is_configured", lambda: True)

        class FakeProc:
            returncode = 0
            stdout = "not valid json {"
            stderr = ""

        def fake_run(*a, **kw):
            return FakeProc()

        monkeypatch.setattr(_mod.subprocess, "run", fake_run)
        out = eng.evaluate({"id": "x"}, _smoke_policy_dir(), "cid-bad")
        assert len(out) == 1
        assert out[0]["result"] == "error"
        assert out[0]["ruleId"] == "KJ_ENGINE_ERROR"
        jsonschema.validate(out[0], PCR_SCHEMA)

    def test_missing_policy_dir_produces_error_pcr(self, monkeypatch):
        eng = KyvernoJsonEngine()
        monkeypatch.setattr(_mod, "_which_kj", lambda: "/fake/kj")
        monkeypatch.setattr(eng, "is_configured", lambda: True)
        out = eng.evaluate({"id": "x"}, Path("/nonexistent/dir"), "cid-miss")
        assert len(out) == 1
        assert out[0]["result"] == "error"
        assert "not found" in out[0]["message"]