101 lines
4.1 KiB
Python
101 lines
4.1 KiB
Python
from inspect import getsource
|
|
from pathlib import Path
|
|
|
|
import app.document_reconciliation_service as service
|
|
import app.reconciliation_service as reconciliation_service
|
|
|
|
|
|
class _Result:
|
|
def __init__(self, rowcount=1):
|
|
self.rowcount = rowcount
|
|
|
|
|
|
class _Connection:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def execute(self, statement, params=None):
|
|
self.calls.append((str(statement), params or {}))
|
|
return _Result()
|
|
|
|
|
|
def test_primary_projects_exact_document_identity_to_linked():
|
|
connection = _Connection()
|
|
assert service._sync_reconciliation_item_for_active_link(
|
|
connection, "doc-primary", "opp-primary", "PRIMARY",
|
|
) == 1
|
|
sql, params = connection.calls[0]
|
|
assert "status = 'linked'" in sql
|
|
assert "resolved_at = COALESCE(ri.resolved_at, now())" in sql
|
|
assert "ri.source_system = d.system" in sql
|
|
assert "ri.document_number = d.document_number" in sql
|
|
assert "ri.external_id = d.external_id" in sql
|
|
assert "payload" not in sql.lower()
|
|
assert "customer" not in sql.lower()
|
|
assert params == {"document_id": "doc-primary", "opportunity_id": "opp-primary"}
|
|
|
|
|
|
def test_secondary_and_reassign_project_destination():
|
|
connection = _Connection()
|
|
service._sync_reconciliation_item_for_active_link(
|
|
connection, "doc-secondary", "opp-destination", "SECONDARY",
|
|
)
|
|
assert connection.calls[0][1]["opportunity_id"] == "opp-destination"
|
|
reassign_source = getsource(service.reassign_document)
|
|
assert "destination_opportunity_id, \"SECONDARY\"" in reassign_source
|
|
assert "_sync_reconciliation_item_for_active_link" in reassign_source
|
|
|
|
|
|
def test_source_system_is_mandatory_and_cannot_cross_match():
|
|
connection = _Connection()
|
|
service._sync_reconciliation_item_for_active_link(connection, "doc", "opp", "PRIMARY")
|
|
sql = connection.calls[0][0]
|
|
assert "ri.source_system = d.system" in sql
|
|
assert "source_system" in sql
|
|
|
|
|
|
def test_ignored_historical_and_removed_do_not_change_items():
|
|
for relationship in ("IGNORED", "HISTORICAL", "REMOVED", "REVIEW_REQUIRED", "REASSIGNED"):
|
|
connection = _Connection()
|
|
assert service._sync_reconciliation_item_for_active_link(
|
|
connection, "doc", "opp", relationship,
|
|
) == 0
|
|
assert connection.calls == []
|
|
|
|
|
|
def test_set_relationship_and_auto_classification_use_central_projection():
|
|
set_source = getsource(service.set_document_relationship)
|
|
classify_source = getsource(service.classify_sync_document)
|
|
assert set_source.count("_sync_reconciliation_item_for_active_link") >= 2
|
|
assert classify_source.count("_sync_reconciliation_item_for_active_link") >= 2
|
|
assert "if old == relationship" in set_source
|
|
|
|
|
|
def test_read_safeguard_hides_stale_open_items_with_active_links():
|
|
predicate = service.active_document_link_exclusion_sql("ri")
|
|
assert "ended_at IS NULL" in predicate
|
|
assert "relationship IN ('PRIMARY','SECONDARY')" in predicate
|
|
assert "linked_cd.system = ri.source_system" in predicate
|
|
list_source = getsource(reconciliation_service.list_reconciliation_items)
|
|
by_ids_source = getsource(reconciliation_service._get_reconciliation_items_by_ids)
|
|
assert "active_document_link_exclusion_sql" in list_source
|
|
assert "active_document_link_exclusion_sql" in by_ids_source
|
|
|
|
|
|
def test_unlink_reopen_policy_is_preserved():
|
|
source = Path("app/admin_ui/pages/opportunities.py").read_text()
|
|
assert "status = CASE WHEN status IN ('resolved','linked','applied','open','needs_review','conflict') THEN 'needs_review' ELSE status END" in source
|
|
assert "resolved_at = NULL" in source
|
|
assert "reconciliation_items_unlinked" in source
|
|
|
|
|
|
def test_repair_is_dry_run_safe_exact_and_unambiguous():
|
|
source = Path("scripts/repair_document_reconciliation_item_links.py").read_text()
|
|
assert "apply: bool = False" in source
|
|
assert "--apply" in source
|
|
assert "COUNT(DISTINCT opportunity_id) = 1" in source
|
|
assert "cd.system = ri.source_system" in source
|
|
assert "payload::text" not in source
|
|
assert "customer_id" not in source
|
|
assert "ri.status NOT IN ('ignored','historical')" in source
|