384 lines
12 KiB
Python
384 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import timedelta
|
|
from types import SimpleNamespace
|
|
import hashlib
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.sanctions import (
|
|
CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS,
|
|
SanctionsSnapshotPayload,
|
|
SanctionsSnapshotReference,
|
|
)
|
|
from govoplan_core.db.base import Base, utcnow
|
|
from govoplan_risk_compliance.backend.db.models import (
|
|
RiskSanctionsAddress,
|
|
RiskSanctionsAlias,
|
|
RiskSanctionsDate,
|
|
RiskSanctionsEntry,
|
|
RiskSanctionsIdentifier,
|
|
RiskSanctionsListSnapshot,
|
|
RiskScreeningCandidate,
|
|
RiskScreeningDisposition,
|
|
RiskScreeningException,
|
|
RiskScreeningRun,
|
|
RiskScreeningSubjectSnapshot,
|
|
)
|
|
from govoplan_risk_compliance.backend.permissions import (
|
|
SANCTIONS_ADMIN_SCOPE,
|
|
SANCTIONS_READ_SCOPE,
|
|
SANCTIONS_REVIEW_SCOPE,
|
|
SANCTIONS_SCREEN_SCOPE,
|
|
)
|
|
from govoplan_risk_compliance.backend.review import (
|
|
DispositionInput,
|
|
record_disposition,
|
|
)
|
|
from govoplan_risk_compliance.backend.sanctions_catalog import (
|
|
RiskSanctionsAccessError,
|
|
RiskSanctionsConflictError,
|
|
import_connector_snapshot,
|
|
)
|
|
from govoplan_risk_compliance.backend.screening import (
|
|
MATCHER_VERSION,
|
|
ScreeningPolicy,
|
|
ScreeningSubject,
|
|
run_screening,
|
|
)
|
|
|
|
|
|
UN_XML = b"""<?xml version="1.0" encoding="UTF-8"?>
|
|
<CONSOLIDATED_LIST dateGenerated="2026-07-29T00:00:00Z">
|
|
<INDIVIDUALS>
|
|
<INDIVIDUAL>
|
|
<DATAID>42</DATAID>
|
|
<VERSIONNUM>1</VERSIONNUM>
|
|
<FIRST_NAME>Example</FIRST_NAME>
|
|
<SECOND_NAME>Person</SECOND_NAME>
|
|
<UN_LIST_TYPE>Example programme</UN_LIST_TYPE>
|
|
<REFERENCE_NUMBER>QI.42.26</REFERENCE_NUMBER>
|
|
<LISTED_ON>2026-01-02</LISTED_ON>
|
|
<INDIVIDUAL_ALIAS>
|
|
<QUALITY>Good</QUALITY>
|
|
<ALIAS_NAME>Example Alias</ALIAS_NAME>
|
|
</INDIVIDUAL_ALIAS>
|
|
<INDIVIDUAL_DOCUMENT>
|
|
<TYPE_OF_DOCUMENT>Passport</TYPE_OF_DOCUMENT>
|
|
<NUMBER>P-123 456</NUMBER>
|
|
</INDIVIDUAL_DOCUMENT>
|
|
<INDIVIDUAL_DATE_OF_BIRTH>
|
|
<DATE>1980-01-02</DATE>
|
|
</INDIVIDUAL_DATE_OF_BIRTH>
|
|
<INDIVIDUAL_ADDRESS>
|
|
<CITY>Example City</CITY>
|
|
<COUNTRY>Example Country</COUNTRY>
|
|
</INDIVIDUAL_ADDRESS>
|
|
</INDIVIDUAL>
|
|
</INDIVIDUALS>
|
|
<ENTITIES>
|
|
<ENTITY>
|
|
<DATAID>84</DATAID>
|
|
<VERSIONNUM>1</VERSIONNUM>
|
|
<FIRST_NAME>Example Trading Company</FIRST_NAME>
|
|
<REFERENCE_NUMBER>QE.84.26</REFERENCE_NUMBER>
|
|
<LISTED_ON>2026-01-03</LISTED_ON>
|
|
</ENTITY>
|
|
</ENTITIES>
|
|
</CONSOLIDATED_LIST>
|
|
"""
|
|
|
|
TABLES = (
|
|
RiskSanctionsListSnapshot.__table__,
|
|
RiskSanctionsEntry.__table__,
|
|
RiskSanctionsAlias.__table__,
|
|
RiskSanctionsIdentifier.__table__,
|
|
RiskSanctionsDate.__table__,
|
|
RiskSanctionsAddress.__table__,
|
|
RiskScreeningSubjectSnapshot.__table__,
|
|
RiskScreeningRun.__table__,
|
|
RiskScreeningCandidate.__table__,
|
|
RiskScreeningDisposition.__table__,
|
|
RiskScreeningException.__table__,
|
|
)
|
|
|
|
|
|
def principal(
|
|
account_id: str = "account-1",
|
|
*,
|
|
scopes: tuple[str, ...] = (
|
|
SANCTIONS_READ_SCOPE,
|
|
SANCTIONS_SCREEN_SCOPE,
|
|
SANCTIONS_REVIEW_SCOPE,
|
|
SANCTIONS_ADMIN_SCOPE,
|
|
),
|
|
) -> ApiPrincipal:
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id=account_id,
|
|
membership_id=f"membership-{account_id}",
|
|
tenant_id="tenant-1",
|
|
scopes=frozenset(scopes),
|
|
),
|
|
account=SimpleNamespace(id=account_id),
|
|
user=SimpleNamespace(id=account_id),
|
|
)
|
|
|
|
|
|
class _SnapshotProvider:
|
|
def __init__(self) -> None:
|
|
digest = hashlib.sha256(UN_XML).hexdigest()
|
|
self.snapshot = SanctionsSnapshotReference(
|
|
ref="sanctions-snapshot:fixture",
|
|
tenant_id="tenant-1",
|
|
provider_id="synthetic.un_fixture",
|
|
publisher="United Nations Security Council fixture",
|
|
jurisdiction="UN",
|
|
list_type="consolidated_sanctions",
|
|
source_id="synthetic-un-v1",
|
|
source_version="fixture-v1",
|
|
publication_at=utcnow(),
|
|
effective_at=utcnow(),
|
|
acquired_at=utcnow(),
|
|
content_type="application/xml",
|
|
byte_count=len(UN_XML),
|
|
sha256=digest,
|
|
parser_version="unsc-xml-v1",
|
|
raw_evidence_ref="connector-evidence:fixture",
|
|
connector_run_id="run-fixture",
|
|
)
|
|
|
|
def list_snapshots(self, session, principal, *, limit=100):
|
|
del session, principal, limit
|
|
return (self.snapshot,)
|
|
|
|
def get_snapshot(self, session, principal, *, snapshot_ref):
|
|
del session, principal
|
|
return self.snapshot if snapshot_ref == self.snapshot.ref else None
|
|
|
|
def read_snapshot(self, session, principal, *, snapshot_ref):
|
|
del session, principal
|
|
if snapshot_ref != self.snapshot.ref:
|
|
raise ValueError("not found")
|
|
return SanctionsSnapshotPayload(
|
|
snapshot=self.snapshot,
|
|
content=UN_XML,
|
|
)
|
|
|
|
|
|
class _Registry:
|
|
def __init__(self, provider) -> None:
|
|
self.provider = provider
|
|
|
|
def has_capability(self, name: str) -> bool:
|
|
return name == CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS
|
|
|
|
def capability(self, name: str):
|
|
return self.provider if self.has_capability(name) else None
|
|
|
|
|
|
class SanctionsScreeningTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(self.engine, tables=TABLES)
|
|
self.session = Session(self.engine)
|
|
self.provider = _SnapshotProvider()
|
|
self.registry = _Registry(self.provider)
|
|
self.list_snapshot, _ = import_connector_snapshot(
|
|
self.session,
|
|
principal(),
|
|
registry=self.registry,
|
|
connector_snapshot_ref=self.provider.snapshot.ref,
|
|
)
|
|
self.session.commit()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_import_is_immutable_idempotent_and_normalized(self) -> None:
|
|
second, created = import_connector_snapshot(
|
|
self.session,
|
|
principal(),
|
|
registry=self.registry,
|
|
connector_snapshot_ref=self.provider.snapshot.ref,
|
|
)
|
|
|
|
self.assertFalse(created)
|
|
self.assertEqual(self.list_snapshot.id, second.id)
|
|
self.assertEqual(2, second.entry_count)
|
|
person = self.session.query(RiskSanctionsEntry).filter_by(
|
|
source_entry_id="QI.42.26"
|
|
).one()
|
|
self.assertEqual("example person", person.normalized_name)
|
|
self.assertEqual("example alias", person.aliases[0].normalized_name)
|
|
self.assertEqual("p123456", person.identifiers[0].normalized_value)
|
|
self.assertEqual(
|
|
"INDIVIDUALS/INDIVIDUAL[1]",
|
|
person.raw_evidence_locator,
|
|
)
|
|
|
|
def test_screening_is_versioned_explainable_and_idempotent(self) -> None:
|
|
item, created = run_screening(
|
|
self.session,
|
|
principal(),
|
|
list_snapshot_id=self.list_snapshot.id,
|
|
idempotency_key="screening-1",
|
|
subject=ScreeningSubject(
|
|
subject_type="person",
|
|
primary_name="Exampel Person",
|
|
),
|
|
)
|
|
|
|
self.assertTrue(created)
|
|
self.assertEqual("potential", item.outcome)
|
|
self.assertEqual(MATCHER_VERSION, item.matcher_version)
|
|
self.assertEqual(1, item.candidate_count)
|
|
self.assertEqual("name_fuzzy", item.candidates[0].match_kind)
|
|
self.assertEqual("pending", item.candidates[0].review_status)
|
|
self.assertFalse(
|
|
item.candidates[0].evidence[0]["auto_confirmed"]
|
|
)
|
|
|
|
replay, replay_created = run_screening(
|
|
self.session,
|
|
principal(),
|
|
list_snapshot_id=self.list_snapshot.id,
|
|
idempotency_key="screening-1",
|
|
subject=ScreeningSubject(
|
|
subject_type="person",
|
|
primary_name="Exampel Person",
|
|
),
|
|
)
|
|
self.assertFalse(replay_created)
|
|
self.assertEqual(item.id, replay.id)
|
|
|
|
with self.assertRaises(RiskSanctionsConflictError):
|
|
run_screening(
|
|
self.session,
|
|
principal(),
|
|
list_snapshot_id=self.list_snapshot.id,
|
|
idempotency_key="screening-1",
|
|
subject=ScreeningSubject(
|
|
subject_type="person",
|
|
primary_name="Different Person",
|
|
),
|
|
)
|
|
|
|
def test_identifier_match_is_exact(self) -> None:
|
|
item, _ = run_screening(
|
|
self.session,
|
|
principal(),
|
|
list_snapshot_id=self.list_snapshot.id,
|
|
idempotency_key="identifier-1",
|
|
subject=ScreeningSubject(
|
|
subject_type="person",
|
|
identifiers=(
|
|
{
|
|
"type": "passport",
|
|
"value": "P123-456",
|
|
},
|
|
),
|
|
),
|
|
)
|
|
|
|
self.assertEqual("identifier_exact", item.candidates[0].match_kind)
|
|
self.assertEqual(100, item.candidates[0].score)
|
|
|
|
def test_stale_list_has_explicit_outcome(self) -> None:
|
|
self.list_snapshot.acquired_at = utcnow() - timedelta(days=30)
|
|
item, _ = run_screening(
|
|
self.session,
|
|
principal(),
|
|
list_snapshot_id=self.list_snapshot.id,
|
|
idempotency_key="stale-1",
|
|
subject=ScreeningSubject(
|
|
subject_type="person",
|
|
primary_name="Example Person",
|
|
),
|
|
policy=ScreeningPolicy(max_snapshot_age_days=7),
|
|
)
|
|
|
|
self.assertEqual("stale", item.outcome)
|
|
self.assertEqual(0, item.candidate_count)
|
|
|
|
def test_review_is_separated_and_exception_requires_review(self) -> None:
|
|
submitter = principal(
|
|
scopes=(
|
|
SANCTIONS_READ_SCOPE,
|
|
SANCTIONS_SCREEN_SCOPE,
|
|
)
|
|
)
|
|
item, _ = run_screening(
|
|
self.session,
|
|
submitter,
|
|
list_snapshot_id=self.list_snapshot.id,
|
|
idempotency_key="review-1",
|
|
subject=ScreeningSubject(
|
|
subject_type="person",
|
|
primary_name="Example Person",
|
|
),
|
|
)
|
|
candidate = item.candidates[0]
|
|
with self.assertRaises(RiskSanctionsAccessError):
|
|
record_disposition(
|
|
self.session,
|
|
principal(
|
|
scopes=(SANCTIONS_REVIEW_SCOPE,)
|
|
),
|
|
candidate_id=candidate.id,
|
|
disposition=DispositionInput(
|
|
decision="false_positive",
|
|
reason="Independent evidence excludes this subject.",
|
|
),
|
|
)
|
|
|
|
reviewer = principal(
|
|
"account-2",
|
|
scopes=(SANCTIONS_REVIEW_SCOPE,),
|
|
)
|
|
reviewed, disposition = record_disposition(
|
|
self.session,
|
|
reviewer,
|
|
candidate_id=candidate.id,
|
|
disposition=DispositionInput(
|
|
decision="false_positive",
|
|
reason="Independent evidence excludes this subject.",
|
|
exception_scope="subject_entry",
|
|
expires_at=utcnow() + timedelta(days=30),
|
|
),
|
|
)
|
|
self.assertEqual("false_positive", reviewed.review_status)
|
|
self.assertEqual("independent", disposition.separation_status)
|
|
self.assertEqual(
|
|
1,
|
|
self.session.query(RiskScreeningException).count(),
|
|
)
|
|
|
|
repeated, _ = run_screening(
|
|
self.session,
|
|
submitter,
|
|
list_snapshot_id=self.list_snapshot.id,
|
|
idempotency_key="review-2",
|
|
subject=ScreeningSubject(
|
|
subject_type="person",
|
|
primary_name="Example Person",
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
"exception_review",
|
|
repeated.candidates[0].review_status,
|
|
)
|
|
self.assertEqual(
|
|
"prior_exception",
|
|
repeated.candidates[0].evidence[-1]["kind"],
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|