acdl/tests/test_confidence_signal.py

import json
import sys
from pathlib import Path

import pytest

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

from acdl_platform.confidence_signal import (
    WEIGHTS, PENALTY, THRESHOLDS, compute, Signal, _per_input_score,
)


class TestWeights:
    def test_weights_sum_to_one(self):
        assert sum(WEIGHTS.values()) == pytest.approx(1.0)

    def test_policy_weight_highest(self):
        assert WEIGHTS["policy"] == 0.30

    def test_validation_weight(self):
        assert WEIGHTS["validation"] == 0.25


class TestThresholds:
    def test_dev_threshold(self):
        assert THRESHOLDS["dev"] == 0.50

    def test_qa_threshold(self):
        assert THRESHOLDS["qa"] == 0.75

    def test_prod_threshold(self):
        assert THRESHOLDS["prod"] == 0.90

    def test_dr_threshold(self):
        assert THRESHOLDS["dr"] == 0.95


class TestPenalty:
    def test_critical_is_none(self):
        assert PENALTY["critical"] is None

    def test_high_penalty(self):
        assert PENALTY["high"] == 0.20

    def test_medium_penalty(self):
        assert PENALTY["medium"] == 0.05

    def test_low_penalty(self):
        assert PENALTY["low"] == 0.01

    def test_info_no_penalty(self):
        assert PENALTY["info"] == 0.0


class TestPerInputScore:
    def test_missing_input_returns_half(self):
        score, reasons = _per_input_score("policy", None)
        assert score == 0.5
        assert "INPUT_MISSING:policy" in reasons

    def test_empty_policy_list(self):
        score, reasons = _per_input_score("policy", [])
        assert score == 0.5
        assert reasons == []

    def test_all_pass_policy(self):
        pcrs = [{"result": "pass"}, {"result": "pass"}]
        score, reasons = _per_input_score("policy", pcrs)
        assert score == 1.0
        assert reasons == []

    def test_mixed_policy(self):
        pcrs = [{"result": "pass"}, {"result": "fail"}]
        score, reasons = _per_input_score("policy", pcrs)
        assert score == 0.5

    def test_skipped_counts_as_pass(self):
        pcrs = [{"result": "skipped"}]
        score, reasons = _per_input_score("policy", pcrs)
        assert score == 1.0

    def test_validation_all_true(self):
        score, reasons = _per_input_score("validation", {
            "schema": True, "stack_resolved": True,
            "tf_validated": True, "tf_planned": True
        })
        assert score == 1.0

    def test_validation_partial(self):
        score, reasons = _per_input_score("validation", {
            "schema": True, "stack_resolved": True,
            "tf_validated": False, "tf_planned": False
        })
        assert score == 0.5

    def test_freshness_fresh(self):
        score, _ = _per_input_score("freshness", {"age_days": 0, "max_age_days": 7})
        assert score == 1.0

    def test_freshness_stale(self):
        score, _ = _per_input_score("freshness", {"age_days": 7, "max_age_days": 7})
        assert score == pytest.approx(0.0)

    def test_source_complete(self):
        score, _ = _per_input_score("source", {"submitter": "dev", "commit_sha": "abc"})
        assert score == 1.0

    def test_source_partial(self):
        score, _ = _per_input_score("source", {"submitter": "dev"})
        assert score == 0.5

    def test_history_clean(self):
        score, _ = _per_input_score("history", {"prior_rollbacks": 0, "prior_policy_fails": 0})
        assert score == 1.0

    def test_history_with_failures(self):
        score, _ = _per_input_score("history", {"prior_rollbacks": 2, "prior_policy_fails": 3})
        assert score == pytest.approx(0.3)

    def test_nfrs_none(self):
        score, _ = _per_input_score("nfrs", {"conformance": None})
        assert score == 0.5

    def test_nfrs_full(self):
        score, _ = _per_input_score("nfrs", {"conformance": 0.95})
        assert score == 0.95


class TestCompute:
    def _base_inputs(self):
        return {
            "policy": [{"result": "pass"}],
            "validation": {"schema": True, "stack_resolved": True,
                           "tf_validated": True, "tf_planned": True},
            "freshness": {"age_days": 0, "max_age_days": 7},
            "source": {"submitter": "dev", "commit_sha": "abc"},
            "history": {"prior_rollbacks": 0, "prior_policy_fails": 0},
            "nfrs": {"conformance": None},
        }

    def test_dev_pass(self):
        sig = compute("test-001", "dev", self._base_inputs())
        assert sig.band == "pass"
        assert sig.score >= 0.50

    def test_missing_input_blocks(self):
        inputs = self._base_inputs()
        del inputs["policy"]
        sig = compute("test-002", "dev", inputs)
        assert sig.band == "block"
        assert sig.score == 0.0
        assert any("INPUT_MISSING" in r for r in sig.reasonCodes)

    def test_critical_policy_blocks(self):
        inputs = self._base_inputs()
        inputs["policy"] = [{"result": "fail", "severity": "critical", "ruleId": "CKV_X"}]
        sig = compute("test-003", "dev", inputs)
        assert sig.band == "block"
        assert sig.score == 0.0
        assert any("CRITICAL_OVERRIDE" in r for r in sig.reasonCodes)

    def test_high_policy_lowers_score(self):
        inputs = self._base_inputs()
        inputs["policy"] = [{"result": "fail", "severity": "high", "ruleId": "CKV_Y"}]
        sig = compute("test-004", "dev", inputs)
        assert sig.score < 1.0

    def test_dev_warn_becomes_block(self):
        sig = compute("test-005", "dev", self._base_inputs())
        assert sig.band != "warn"

    def test_signal_has_per_input(self):
        sig = compute("test-006", "dev", self._base_inputs())
        assert "policy" in sig.perInput
        assert "validation" in sig.perInput
        assert "nfrs" in sig.perInput

    def test_all_six_inputs_present(self):
        sig = compute("test-007", "dev", self._base_inputs())
        assert len(sig.perInput) == 6