85 lines
3.3 KiB
Python
85 lines
3.3 KiB
Python
from inspect import getsource
|
|
|
|
import app.email_identity_extraction_service as extraction_service
|
|
import app.fiscal_enrichment_service as service
|
|
from app.admin_ui.pages import opportunities as opportunity_page
|
|
|
|
|
|
OPPORTUNITY = {
|
|
"id": "11111111-1111-1111-1111-111111111111",
|
|
"local_customer_id": None,
|
|
"linked_customer_id": None,
|
|
"linked_customer_name": None,
|
|
"linked_customer_tax_id": None,
|
|
}
|
|
|
|
|
|
def test_read_review_uses_latest_only_and_never_upserts(monkeypatch):
|
|
calls = {"latest": 0, "extract": 0, "upsert": 0}
|
|
|
|
def latest(opportunity_id):
|
|
calls["latest"] += 1
|
|
return {"company_mentions": ["Example Lda"], "confidence": 0.9}
|
|
|
|
monkeypatch.setattr(service, "latest_identity_for_opportunity", latest)
|
|
monkeypatch.setattr(service, "extract_identity_for_opportunity", lambda *a, **k: calls.__setitem__("extract", calls["extract"] + 1))
|
|
monkeypatch.setattr(extraction_service, "upsert_identity_extraction_for_opportunity", lambda *a, **k: calls.__setitem__("upsert", calls["upsert"] + 1))
|
|
monkeypatch.setattr(service, "_find_internal_customer_by_identity", lambda *a, **k: None)
|
|
|
|
result = service.email_identity_review_for_opportunity(
|
|
OPPORTUNITY["id"], refresh=False, opportunity=dict(OPPORTUNITY),
|
|
)
|
|
|
|
assert result["ok"] is True
|
|
assert result["identity"]["company_mentions"] == ["Example Lda"]
|
|
assert calls == {"latest": 1, "extract": 0, "upsert": 0}
|
|
|
|
|
|
def test_read_review_without_stored_identity_returns_identity_none(monkeypatch):
|
|
monkeypatch.setattr(service, "latest_identity_for_opportunity", lambda opportunity_id: None)
|
|
monkeypatch.setattr(
|
|
service, "extract_identity_for_opportunity",
|
|
lambda *a, **k: (_ for _ in ()).throw(AssertionError("read path extracted identity")),
|
|
)
|
|
|
|
result = service.email_identity_review_for_opportunity(
|
|
OPPORTUNITY["id"], refresh=False, opportunity=dict(OPPORTUNITY),
|
|
)
|
|
|
|
assert result["ok"] is True
|
|
assert result["identity"] is None
|
|
assert result["valid_company_mentions"] == []
|
|
assert result["conflict"] is False
|
|
|
|
|
|
def test_refresh_review_still_performs_extraction(monkeypatch):
|
|
calls = []
|
|
|
|
def extract(opportunity_id, **kwargs):
|
|
calls.append((opportunity_id, kwargs))
|
|
return {"company_mentions": [], "confidence": 0.5}
|
|
|
|
monkeypatch.setattr(service, "extract_identity_for_opportunity", extract)
|
|
monkeypatch.setattr(
|
|
service, "latest_identity_for_opportunity",
|
|
lambda *a, **k: (_ for _ in ()).throw(AssertionError("refresh used latest-only path")),
|
|
)
|
|
|
|
result = service.email_identity_review_for_opportunity(
|
|
OPPORTUNITY["id"], refresh=True, opportunity=dict(OPPORTUNITY),
|
|
)
|
|
|
|
assert result["ok"] is True
|
|
assert calls and calls[0][1]["refresh"] is True
|
|
|
|
|
|
def test_opportunity_detail_get_has_no_identity_write_or_schema_initialization():
|
|
review_source = getsource(service.email_identity_review_for_opportunity)
|
|
detail_source = getsource(opportunity_page.opportunity_detail_page)
|
|
|
|
assert "latest_identity_for_opportunity(opportunity_id)" in review_source
|
|
assert "ensure_email_identity_schema" not in review_source
|
|
assert "upsert_identity_extraction_for_opportunity" not in review_source
|
|
assert "ensure_pending_task_for_next_action" not in detail_source
|
|
assert "materialized_task" not in detail_source
|