GovOPlaN/govoplan-risk-compliance
2
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: reconcile sanctions screening freshness

Browse Source
This commit is contained in:
zemion
2026-07-30 01:55:03 +02:00
parent 11404ac42f
commit da1baede36
10 changed files with 926 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_manifest.py
+15
View File
@@ -2,6 +2,10 @@ from __future__ import annotations
import unittest
from govoplan_core.core.sanctions import (
CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING,
SanctionsScreeningProvider,
)
from govoplan_risk_compliance.backend.manifest import (
ADMIN_SCOPE,
READ_SCOPE,
@@ -49,6 +53,17 @@ class ManifestTests(unittest.TestCase):
manifest.requires_interfaces[0].name,
)
self.assertTrue(manifest.requires_interfaces[0].optional)
self.assertEqual(
{"risk_compliance.sanctions_screening"},
{
item.name
for item in manifest.provides_interfaces
},
)
capability = manifest.capability_factories[
CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING
](None)
self.assertIsInstance(capability, SanctionsScreeningProvider)
if __name__ == "__main__":
tests/test_sanctions_screening.py
+123
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
from dataclasses import replace
from datetime import timedelta
from types import SimpleNamespace
import hashlib
@@ -12,6 +13,10 @@ from govoplan_core.auth import ApiPrincipal
from govoplan_core.core.access import PrincipalRef
from govoplan_core.core.sanctions import (
CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS,
SanctionsScreeningFreshnessRequest,
SanctionsScreeningPolicy,
SanctionsScreeningRequest,
SanctionsScreeningSubject,
SanctionsSnapshotPayload,
SanctionsSnapshotReference,
)
@@ -29,6 +34,9 @@ from govoplan_risk_compliance.backend.db.models import (
RiskScreeningRun,
RiskScreeningSubjectSnapshot,
)
from govoplan_risk_compliance.backend.capabilities import (
RiskComplianceSanctionsScreeningProvider,
)
from govoplan_risk_compliance.backend.permissions import (
SANCTIONS_ADMIN_SCOPE,
SANCTIONS_READ_SCOPE,
@@ -48,7 +56,10 @@ from govoplan_risk_compliance.backend.screening import (
MATCHER_VERSION,
ScreeningPolicy,
ScreeningSubject,
assess_screening_freshness,
list_rescreening_requirements,
run_screening,
screening_evidence_ref,
)
@@ -306,6 +317,108 @@ class SanctionsScreeningTests(unittest.TestCase):
self.assertEqual("stale", item.outcome)
self.assertEqual(0, item.candidate_count)
def test_freshness_reconciles_source_subject_matcher_and_policy(self) -> None:
item, _ = run_screening(
self.session,
principal(),
list_snapshot_id=self.list_snapshot.id,
idempotency_key="freshness-1",
subject=ScreeningSubject(
subject_type="person",
primary_name="No Match",
subject_ref="person-1",
),
)
item.matcher_version = "sanctions-matcher-legacy"
self.provider.snapshot = replace(
self.provider.snapshot,
ref="sanctions-snapshot:fixture-v2",
source_version="fixture-v2",
acquired_at=utcnow(),
connector_run_id="run-fixture-v2",
)
current_snapshot, created = import_connector_snapshot(
self.session,
principal(),
registry=self.registry,
connector_snapshot_ref=self.provider.snapshot.ref,
)
self.assertTrue(created)
assessment = assess_screening_freshness(
self.session,
principal(),
evidence_ref=screening_evidence_ref(item.id),
current_subject=ScreeningSubject(
subject_type="person",
primary_name="Changed Subject",
subject_ref="person-1",
),
policy=ScreeningPolicy(fuzzy_threshold=0.9),
failure_policy="degraded",
)
self.assertFalse(assessment.fresh)
self.assertEqual(current_snapshot.id, assessment.current_list_snapshot_id)
self.assertEqual("degraded", assessment.gate_decision)
self.assertIn("source_snapshot_changed", assessment.reasons)
self.assertIn("subject_changed", assessment.reasons)
self.assertIn("matcher_version_changed", assessment.reasons)
self.assertIn("policy_changed", assessment.reasons)
requirements = list_rescreening_requirements(
self.session,
principal(),
)
self.assertEqual([item.id], [value.run.id for value in requirements])
def test_versioned_capability_returns_stable_gate_evidence(self) -> None:
provider = RiskComplianceSanctionsScreeningProvider()
request = SanctionsScreeningRequest(
list_snapshot_id=self.list_snapshot.id,
idempotency_key="capability-1",
subject=SanctionsScreeningSubject(
subject_type="entity",
primary_name="No Match Company",
subject_ref="entity-1",
),
policy=SanctionsScreeningPolicy(
failure_policy="review",
),
)
result = provider.request_screening(
self.session,
principal(),
request,
)
replay = provider.request_screening(
self.session,
principal(),
request,
)
changed = provider.check_freshness(
self.session,
principal(),
SanctionsScreeningFreshnessRequest(
evidence_ref=result.evidence.ref,
current_subject=SanctionsScreeningSubject(
subject_type="entity",
primary_name="Changed Company",
subject_ref="entity-1",
),
expected_list_snapshot_id=self.list_snapshot.id,
policy=request.policy,
),
)
self.assertTrue(result.created)
self.assertFalse(replay.created)
self.assertEqual(result.evidence.ref, replay.evidence.ref)
self.assertEqual("allow", result.freshness.gate_decision)
self.assertFalse(changed.fresh)
self.assertEqual("review", changed.gate_decision)
self.assertIn("subject_changed", changed.reasons)
def test_review_is_separated_and_exception_requires_review(self) -> None:
submitter = principal(
scopes=(
@@ -358,6 +471,16 @@ class SanctionsScreeningTests(unittest.TestCase):
1,
self.session.query(RiskScreeningException).count(),
)
expired = assess_screening_freshness(
self.session,
principal(),
evidence_ref=screening_evidence_ref(item.id),
expected_list_snapshot_id=self.list_snapshot.id,
failure_policy="review",
now=disposition.expires_at + timedelta(seconds=1),
)
self.assertIn("disposition_expired", expired.reasons)
self.assertEqual("review", expired.gate_decision)
repeated, _ = run_screening(
self.session,