breakpilot-compliance/backend-compliance/tests/test_transition_reasoning.py

"""Tests for Transition Reasoning v0 (RS-005) — the Transition Planning Engine.

Acceptance: from a TransitionGoal + the Company Capability Profile (2A, „have") +
INJECTED TargetRequirements (Execution-owned „required"), the engine emits ranked
`TransitionQuestionRequest`s (information gaps) — NOT rendered questions. A
certification-derived capability is „probably_covered" (Welt 1), never „already_covered".

The cert->capability mapping below is a MOCK (Execution-owned in reality), only here.
"""

from __future__ import annotations

from compliance.company import (
    CapabilityMappingEntry, Certification, CompanyContext, Declaration,
    ExistingEvidence, build_company_profile,
)
from compliance.reasoning.enums import Confidence
from compliance.transition_reasoning import (
    CoverageStatus, InformationGain, RequestPriority, TargetRequirement, TargetType,
    TransitionContext, TransitionGoal, assess_transition,
)

ISO_MAP = {"ISO27001": CapabilityMappingEntry(
    capability_ids=["cap_incident_response", "cap_supplier_management"], confidence=Confidence.MEDIUM)}


def _profile():
    ctx = CompanyContext(
        company_id="kunde",
        certifications=[Certification(certification_id="ISO27001")],
        declarations=[Declaration(capability_id="cap_asset_management")],
        evidence=[ExistingEvidence(evidence_id="patch.pdf", evidence_type="policy", proves_capability_id="cap_patch_management")],
    )
    return build_company_profile(ctx, ISO_MAP)


def _ctx():
    return TransitionContext(company_id="kunde", known_certifications=["ISO27001"],
                             target=TransitionGoal(target_id="CRA", target_type=TargetType.REGULATION))


# CRA-Required (injiziert; in echt: Obligation->Control->Required Capability, Execution)
def _reqs():
    return [
        TargetRequirement(capability_id="cap_patch_management", expected_evidence=["policy"]),       # confirmed
        TargetRequirement(capability_id="cap_incident_response"),                                     # inferred (ISO)
        TargetRequirement(capability_id="cap_asset_management"),                                      # declared
        TargetRequirement(capability_id="cap_sbom", question_intent="verify_existence", expected_evidence=["sbom"]),  # missing
        TargetRequirement(capability_id="cap_vuln_handling", supports_obligations=["CRA.1", "CRA.2"]),  # missing, 2 obligations
        TargetRequirement(capability_id="cap_wastewater", unsupported=True),                          # not in corpus
    ]


def _req_ids(a):
    return [r.capability_id for r in a.question_requests]


def _cov(a, cap):
    return [c for c in a.coverage if c.capability_id == cap][0]


# The engine emits REQUESTS (information gaps), not rendered questions.
def test_emits_requests_not_questions():
    a = assess_transition(_ctx(), _reqs(), _profile())
    r = a.question_requests[0]
    assert r.capability_id and r.question_intent and r.priority
    # NO rendered question text anywhere — rendering is RS-005.1, not this engine
    assert not hasattr(r, "question")
    assert "question" not in type(r).model_fields and "rendered_text" not in type(r).model_fields


# Confirmed capability -> already_covered, NO request.
def test_confirmed_already_covered_no_request():
    a = assess_transition(_ctx(), _reqs(), _profile())
    assert _cov(a, "cap_patch_management").status == CoverageStatus.ALREADY_COVERED
    assert "cap_patch_management" not in _req_ids(a)


# A certification-inferred capability is PROBABLY_COVERED (Welt 1), not already_covered;
# it produces a confirmation request, never a verdict.
def test_certification_inferred_is_probable_with_confirm_request():
    a = assess_transition(_ctx(), _reqs(), _profile())
    c = _cov(a, "cap_incident_response")
    assert c.status == CoverageStatus.PROBABLY_COVERED
    assert c.status != CoverageStatus.ALREADY_COVERED  # cert alone never „erfüllt"
    req = [r for r in a.question_requests if r.capability_id == "cap_incident_response"][0]
    assert req.priority == RequestPriority.MEDIUM


# A missing required capability -> high-priority request.
def test_missing_high_priority_request():
    a = assess_transition(_ctx(), _reqs(), _profile())
    assert _cov(a, "cap_sbom").status == CoverageStatus.MISSING
    req = [r for r in a.question_requests if r.capability_id == "cap_sbom"][0]
    assert req.priority == RequestPriority.HIGH and req.information_gain == InformationGain.HIGH


# An unsupported domain -> no request (future corpus, honest).
def test_unsupported_no_request():
    a = assess_transition(_ctx(), _reqs(), _profile())
    assert _cov(a, "cap_wastewater").status == CoverageStatus.UNSUPPORTED
    assert "cap_wastewater" not in _req_ids(a)


# Requests are ranked: HIGH (missing) before MEDIUM (probable/declared).
def test_requests_ranked_high_before_medium():
    a = assess_transition(_ctx(), _reqs(), _profile())
    prios = [r.priority for r in a.question_requests]
    assert prios == sorted(prios, key=lambda p: {RequestPriority.HIGH: 0, RequestPriority.MEDIUM: 1, RequestPriority.LOW: 2}[p])
    # the two missing caps come first
    assert set(_req_ids(a)[:2]) == {"cap_sbom", "cap_vuln_handling"}


# The funnel: certs reduce the open questions (only 4 of 6 requirements need clarifying).
def test_funnel_reduces_open_questions():
    a = assess_transition(_ctx(), _reqs(), _profile())
    # already_covered (patch) + unsupported (wastewater) drop out -> 4 requests
    assert len(a.question_requests) == 4
    assert "%" not in a.summary.headline


# Deterministic + activates 2A: same inputs -> same result.
def test_deterministic():
    p = _profile()
    a1 = assess_transition(_ctx(), _reqs(), p)
    a2 = assess_transition(_ctx(), _reqs(), p)
    assert _req_ids(a1) == _req_ids(a2) and a1.summary.headline == a2.summary.headline


# No requirements / no profile -> empty assessment (no Execution data in product code).
def test_empty():
    a = assess_transition(_ctx())
    assert a.question_requests == [] and a.coverage == []