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

"""Tests for Regulatory Change Intelligence (RCI).

Acceptance: a simulated RegulatoryChange against a stored ComplianceBaseline can
answer: (1) does it affect this product? (2) which obligations are new/changed?
(3) which are likely already covered by existing evidence? (4) what must a human
review? (5) what is not relevant?
"""

from __future__ import annotations

from compliance.profile.canonical import (
    CanonicalLifecyclePhase,
    CanonicalProductRegulatoryProfile,
    CanonicalProductType,
    EconomicOperatorRole,
)
from compliance.rci import (
    ChangeType,
    DeltaType,
    RegulatoryChange,
    assess_change,
    create_baseline,
)

PROFILE = CanonicalProductRegulatoryProfile(
    name="Industriespülmaschine",
    product_type=CanonicalProductType.MACHINERY,
    markets=["EU", "DE"],
    economic_operator_role=EconomicOperatorRole.MANUFACTURER,
    lifecycle_phase=CanonicalLifecyclePhase.PLACING_ON_MARKET,
    is_machine=True,
    is_component=False,
    has_software_updates=True,
    has_embedded_software=True,
    has_remote_access=True,
    technologies=["cloud", "ota_updates"],
)

# Evidence the customer already has, per obligation.
EVIDENCE = {"provide_security_updates": ["policy", "ticket"], "sbom_creation": ["sbom"]}

BASELINE = create_baseline(PROFILE, EVIDENCE, baseline_id="b1")


def _change(obs, regs=("CRA",), ctype=ChangeType.AMENDMENT, cid="c"):
    return RegulatoryChange(
        change_id=cid, affected_regulations=list(regs), affected_obligations=list(obs), change_type=ctype
    )


def _by_id(assessment):
    return {d.obligation_id: d for d in assessment.deltas}


# Baseline snapshots the registry-linked obligations from the frozen map.
def test_baseline_snapshots_registry_obligations():
    assert "sbom_creation" in BASELINE.applicable_obligations
    assert "provide_security_updates" in BASELINE.applicable_obligations
    assert BASELINE.regulatory_map_snapshot.scope_resolved is True


# 1 + 2. affects the product + flags a NEW obligation.
def test_affects_product_and_new_obligation():
    a = assess_change(BASELINE, _change(["cra_new_requirement_xyz"], cid="c1"))
    assert a.affects_product is True
    assert _by_id(a)["cra_new_requirement_xyz"].delta_type == DeltaType.NEW


# 2. an existing obligation amended -> CHANGED.
def test_existing_obligation_changed():
    a = assess_change(BASELINE, _change(["sbom_creation"], cid="c2"))
    assert _by_id(a)["sbom_creation"].delta_type == DeltaType.CHANGED


# 3. existing obligation with evidence + guidance update -> ALREADY_COVERED.
def test_already_covered_by_evidence():
    a = assess_change(BASELINE, _change(["provide_security_updates"], ctype=ChangeType.GUIDANCE_UPDATE, cid="c3"))
    assert _by_id(a)["provide_security_updates"].delta_type == DeltaType.ALREADY_COVERED
    assert a.summary.already_covered == ["provide_security_updates"]


# 4. what a human must review (existing obligation without evidence).
def test_needs_review():
    a = assess_change(BASELINE, _change(["vuln_handling_process"], ctype=ChangeType.GUIDANCE_UPDATE, cid="c4"))
    assert _by_id(a)["vuln_handling_process"].delta_type == DeltaType.NEEDS_REVIEW
    assert "vuln_handling_process" in a.summary.needs_review
    assert "vuln_handling_process" in a.summary.what_matters_for_this_product


# 5. a change to a regulation NOT in the map -> not relevant.
def test_not_relevant_offmap_regulation():
    a = assess_change(BASELINE, _change(["psd2_strong_customer_auth"], regs=["PSD2"], ctype=ChangeType.NEW_REGULATION, cid="c5"))
    assert a.affects_product is False
    assert _by_id(a)["psd2_strong_customer_auth"].delta_type == DeltaType.NOT_APPLICABLE
    assert a.summary.not_relevant == ["psd2_strong_customer_auth"]


# repeal removes an existing obligation.
def test_repeal_removes_existing():
    a = assess_change(BASELINE, _change(["sbom_creation"], ctype=ChangeType.REPEAL, cid="c6"))
    assert _by_id(a)["sbom_creation"].delta_type == DeltaType.REMOVED


# missing evidence is computed against the obligation's required evidence.
def test_missing_evidence_on_changed():
    a = assess_change(BASELINE, _change(["sbom_creation"], cid="c7"))  # requires sbom+repo_scan, has sbom
    d = _by_id(a)["sbom_creation"]
    assert "sbom" in d.affected_evidence
    assert "repo_scan" in d.missing_evidence


# a change to an UNCERTAIN regulation -> needs review (resolve applicability first).
def test_uncertain_regulation_needs_review():
    a = assess_change(BASELINE, _change(["red_cyber_req"], regs=["RED"], cid="c8"))
    assert a.affects_product is True  # RED is in the map's uncertain bucket
    assert _by_id(a)["red_cyber_req"].delta_type == DeltaType.NEEDS_REVIEW


# RCI answers "vs my map", not "what does the law say" — and works only on the snapshot.
def test_works_against_stored_map_no_reevaluation():
    # a change with no affected_obligations still resolves affects_product from the map
    a = assess_change(BASELINE, RegulatoryChange(change_id="c9", affected_regulations=["CRA"], affected_obligations=[], change_type=ChangeType.AMENDMENT))
    assert a.affects_product is True
    assert a.deltas == []


# delta_type is a THIRD vocabulary, disjoint from ClaimCoverage (Welt 1).
def test_delta_vocabulary_distinct_from_claimcoverage():
    from compliance.reasoning.enums import ClaimCoverage

    assert {d.value for d in DeltaType}.isdisjoint({c.value for c in ClaimCoverage})


# the management summary aggregates the five buckets coherently.
def test_summary_buckets():
    a = assess_change(
        BASELINE,
        RegulatoryChange(
            change_id="c10",
            affected_regulations=["CRA"],
            affected_obligations=["cra_new_one", "sbom_creation", "provide_security_updates"],
            change_type=ChangeType.AMENDMENT,
        ),
    )
    s = a.summary
    assert "cra_new_one" in s.what_matters_for_this_product  # NEW
    assert "sbom_creation" in s.needs_review  # CHANGED -> review
    assert s.what_changed  # non-empty management line