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")
        assert results == []

    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 results == []

    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 results == []

    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)