acdl/tests/test_kyverno_adapter.py

import json
import sys
from pathlib import Path

import jsonschema
import pytest

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

from adapters.kyverno.kyverno_adapter import (
    SEVERITY_MAP, RESULT_MAP, _to_pcr, adapt,
)


FIXTURES = Path(__file__).resolve().parent.parent / "tests" / "fixtures"


class TestSeverityResultMaps:
    def test_severity_map(self):
        assert SEVERITY_MAP["critical"] == "critical"
        assert SEVERITY_MAP["high"] == "high"
        assert SEVERITY_MAP["medium"] == "medium"
        assert SEVERITY_MAP["low"] == "low"
        assert SEVERITY_MAP["info"] == "info"

    def test_result_map(self):
        assert RESULT_MAP["pass"] == "pass"
        assert RESULT_MAP["fail"] == "fail"
        assert RESULT_MAP["warn"] == "skipped"
        assert RESULT_MAP["error"] == "error"
        assert RESULT_MAP["skip"] == "skipped"


class TestToPcr:
    def test_translates_pass(self):
        entry = {"policy": "p1", "severity": "high", "result": "pass",
                 "message": "ok", "resource": "ns/Pod/x"}
        pcr = _to_pcr(entry, "c-1")
        assert pcr["engine"] == "kyverno"
        assert pcr["ruleId"] == "p1"
        assert pcr["severity"] == "high"
        assert pcr["result"] == "pass"
        assert pcr["resourceRef"] == "ns/Pod/x"

    def test_warn_maps_to_skipped(self):
        entry = {"policy": "p1", "severity": "medium", "result": "warn",
                 "resource": "r"}
        pcr = _to_pcr(entry, "c-1")
        assert pcr["result"] == "skipped"

    def test_unknown_severity_defaults_info(self):
        entry = {"policy": "p1", "severity": "BOGUS", "result": "fail",
                 "resource": "r"}
        pcr = _to_pcr(entry, "c-1")
        assert pcr["severity"] == "info"

    def test_unknown_result_defaults_error(self):
        entry = {"policy": "p1", "severity": "low", "result": "BOGUS",
                 "resource": "r"}
        pcr = _to_pcr(entry, "c-1")
        assert pcr["result"] == "error"

    def test_missing_policy_defaults_unknown(self):
        entry = {"severity": "low", "result": "pass", "resource": "r"}
        pcr = _to_pcr(entry, "c-1")
        assert pcr["ruleId"] == "KYVERNO_UNKNOWN"

    def test_pcr_validates_against_schema(self, policy_check_result_schema):
        entry = {"policy": "p1", "severity": "high", "result": "fail",
                 "message": "m", "resource": "ns/Pod/x", "namespace": "ns",
                 "kind": "Pod", "name": "x"}
        pcr = _to_pcr(entry, "11111111-1111-1111-1111-111111111111")
        jsonschema.validate(pcr, policy_check_result_schema)


class TestAdapt:
    def test_translates_fixture(self, tmp_path, policy_check_result_schema):
        src = FIXTURES / "kyverno_policyreport.json"
        f = tmp_path / "policyreport.json"
        f.write_text(src.read_text())
        results = adapt(str(f), "11111111-1111-1111-1111-111111111111")
        assert len(results) == 3

        # result 1: pass/high
        assert results[0]["ruleId"] == "disallow-privileged-containers"
        assert results[0]["severity"] == "high"
        assert results[0]["result"] == "pass"

        # result 2: fail/medium
        assert results[1]["ruleId"] == "require-resource-labels"
        assert results[1]["severity"] == "medium"
        assert results[1]["result"] == "fail"

        # result 3: warn/high -> skipped/high
        assert results[2]["ruleId"] == "require-image-digests"
        assert results[2]["severity"] == "high"
        assert results[2]["result"] == "skipped"

        for pcr in results:
            jsonschema.validate(pcr, policy_check_result_schema)

    def test_empty_results(self, tmp_path):
        f = tmp_path / "empty.json"
        f.write_text(json.dumps({"results": []}))
        results = adapt(str(f), "11111111-1111-1111-1111-111111111111")
        # Empty results emit the inactive-for-TF guard record (REQ-111).
        assert len(results) == 1
        assert results[0]["ruleId"] == "KYVERNO_INACTIVE_TF_STACK"

    def test_missing_results_key(self, tmp_path):
        f = tmp_path / "noresults.json"
        f.write_text(json.dumps({"apiVersion": "x", "kind": "PolicyReport"}))
        results = adapt(str(f), "11111111-1111-1111-1111-111111111111")
        assert len(results) == 1
        assert results[0]["ruleId"] == "KYVERNO_INACTIVE_TF_STACK"

    def test_non_list_results_treated_as_empty(self, tmp_path):
        f = tmp_path / "bad.json"
        f.write_text(json.dumps({"results": "not-a-list"}))
        results = adapt(str(f), "11111111-1111-1111-1111-111111111111")
        assert len(results) == 1
        assert results[0]["ruleId"] == "KYVERNO_INACTIVE_TF_STACK"

    def test_missing_fields_in_entry(self, tmp_path, policy_check_result_schema):
        f = tmp_path / "sparse.json"
        f.write_text(json.dumps({"results": [{}]}))
        results = adapt(str(f), "11111111-1111-1111-1111-111111111111")
        assert len(results) == 1
        pcr = results[0]
        assert pcr["ruleId"] == "KYVERNO_UNKNOWN"
        assert pcr["severity"] == "info"
        # entry.get("result", "skip") -> default "skip" -> "skipped"
        assert pcr["result"] == "skipped"
        jsonschema.validate(pcr, policy_check_result_schema)

    def test_unknown_result_string_defaults_error(self, tmp_path, policy_check_result_schema):
        # An explicit but unmapped result string falls back to "error".
        f = tmp_path / "unknownresult.json"
        f.write_text(json.dumps({"results": [
            {"policy": "p1", "severity": "low", "result": "BOGUS",
             "resource": "r"},
        ]}))
        results = adapt(str(f), "11111111-1111-1111-1111-111111111111")
        assert results[0]["result"] == "error"
        jsonschema.validate(results[0], policy_check_result_schema)

# --- v1.9 REQ-111: fleshed-out translator tests ---

class TestFleshedOutTranslator:
    def test_pass_result_emits_pass(self, tmp_path):
        f = tmp_path / "pass.json"
        f.write_text(json.dumps({"results": [
            {"policy": "require-labels", "rule": "check-app-label", "severity": "medium",
             "result": "pass", "resource": "pod/x", "message": "label present"},
        ]}))
        results = adapt(str(f), "c1")
        assert results[0]["result"] == "pass"
        assert results[0]["ruleId"] == "require-labels/check-app-label"

    def test_fail_result_with_severity(self, tmp_path):
        f = tmp_path / "fail.json"
        f.write_text(json.dumps({"results": [
            {"policy": "disallow-privileged", "rule": "no-priv", "severity": "critical",
             "result": "fail", "resource": "pod/y", "message": "privileged container"},
        ]}))
        results = adapt(str(f), "c2")
        assert results[0]["result"] == "fail"
        assert results[0]["severity"] == "critical"
        assert results[0]["ruleId"] == "disallow-privileged/no-priv"

    def test_skip_with_reason(self, tmp_path):
        f = tmp_path / "skip.json"
        f.write_text(json.dumps({"results": [
            {"policy": "require-image-digests", "rule": "digest", "severity": "low",
             "result": "skip", "resource": "pod/z", "skipReason": "no image"},
        ]}))
        results = adapt(str(f), "c3")
        assert results[0]["result"] == "skipped"
        assert "no image" in results[0]["message"]

    def test_warn_result_maps_to_skipped(self, tmp_path):
        f = tmp_path / "warn.json"
        f.write_text(json.dumps({"results": [
            {"policy": "p", "rule": "r", "severity": "info", "result": "warn", "resource": "x"},
        ]}))
        results = adapt(str(f), "c4")
        assert results[0]["result"] == "skipped"

    def test_informational_severity_maps_to_info(self, tmp_path):
        f = tmp_path / "info.json"
        f.write_text(json.dumps({"results": [
            {"policy": "p", "rule": "r", "severity": "informational", "result": "pass", "resource": "x"},
        ]}))
        results = adapt(str(f), "c5")
        assert results[0]["severity"] == "info"

    def test_resource_ref_constructed_from_kind_name(self, tmp_path):
        f = tmp_path / "res.json"
        f.write_text(json.dumps({"results": [
            {"policy": "p", "rule": "r", "severity": "low", "result": "fail",
             "kind": "Pod", "namespace": "default", "name": "my-pod"},
        ]}))
        results = adapt(str(f), "c6")
        assert "my-pod" in results[0]["resourceRef"]

    def test_inactive_guard_directly(self):
        from adapters.kyverno.kyverno_adapter import adapt_inactive
        pcrs = adapt_inactive("c7")
        assert len(pcrs) == 1
        assert pcrs[0]["ruleId"] == "KYVERNO_INACTIVE_TF_STACK"
        assert pcrs[0]["result"] == "skipped"
        assert "Terraform" in pcrs[0]["message"]

    def test_kube_version_removed(self, tmp_path):
        """v1.14 (G-103): --kube-version flag removed; adapt() no longer
        accepts kube_version parameter. Version-aware policy selection
        deferred to GitOps reconciler (D-053)."""
        f = tmp_path / "k.json"
        f.write_text(json.dumps({"results": [
            {"policy": "p", "rule": "r", "severity": "low", "result": "pass", "resource": "x"},
        ]}))
        results = adapt(str(f), "c8")
        assert len(results) == 1