perf: add opportunity detail read context

tests/test_email_identity_review_read_semantics.py

@@ -0,0 +1,84 @@
+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

tests/test_opportunity_detail_context.py

@@ -0,0 +1,100 @@
+from inspect import getsource
+
+import app.admin_dashboard as dashboard
+import app.admin_ui.pages.opportunities as page
+import app.commercial_service as commercial
+import app.integration_outbox_service as outbox
+import app.jasmin_backfill_service as backfill
+import app.operation_service as operation
+
+
+def test_detail_context_loads_reusable_groups_once(monkeypatch):
+    calls = {}
+
+    def once(name, value):
+        def load(*args, **kwargs):
+            calls[name] = calls.get(name, 0) + 1
+            return value
+        return load
+
+    opportunity = {"id": "opp", "local_customer_id": "customer", "status": "open", "stage": "NEW_LEAD"}
+    monkeypatch.setattr(page, "get_opportunity", once("opportunity", opportunity))
+    monkeypatch.setattr(page, "list_opportunity_tasks", once("tasks", []))
+    monkeypatch.setattr(page, "list_opportunity_events", once("events", []))
+    monkeypatch.setattr(page, "list_opportunity_items", once("items", []))
+    monkeypatch.setattr(page, "list_products", once("products", []))
+    monkeypatch.setattr(page, "list_communications_for_opportunity", once("communications", []))
+    monkeypatch.setattr(page, "_opportunity_odoo_rows", once("odoo_candidates", ([], [])))
+    monkeypatch.setattr(commercial, "get_customer_for_opportunity", once("customer", {"id": "customer"}))
+    monkeypatch.setattr(commercial, "list_commercial_documents", once("documents", []))
+    monkeypatch.setattr(operation, "get_operation_links", once("operation_links", []))
+    monkeypatch.setattr(operation, "get_operation_snapshot", once("operation_snapshot", {"cards": [], "links": []}))
+    monkeypatch.setattr(outbox, "list_outbox", once("outbox", []))
+    monkeypatch.setattr(backfill, "find_jasmin_document_candidates_for_opportunity", once("jasmin_candidates", []))
+
+    context = page.load_opportunity_detail_context("opp")
+
+    assert context is not None
+    assert calls == {
+        "opportunity": 1, "tasks": 1, "events": 1, "items": 1, "products": 1,
+        "communications": 1, "odoo_candidates": 1, "customer": 1, "documents": 1,
+        "operation_links": 1, "operation_snapshot": 1, "outbox": 1,
+        "jasmin_candidates": 1,
+    }
+
+
+def test_products_panel_preloaded_render_matches_standalone(monkeypatch):
+    items = [{"id": "i1", "product_name": "Produto", "quantity": 1, "unit_price": 10,
+              "total_price": 10, "status": "ACTIVE", "jasmin_sales_item": "A1", "metadata": {}}]
+    products = [{"id": "p1", "sku": "SKU", "jasmin_sales_item": "A1", "name": "Produto"}]
+    monkeypatch.setattr(dashboard, "list_opportunity_items", lambda opportunity_id: items)
+    monkeypatch.setattr(dashboard, "list_products", lambda **kwargs: products)
+
+    standalone = dashboard.opportunity_products_panel_html("opp")
+    preloaded = dashboard.opportunity_products_panel_html(
+        "opp", preloaded_items=items, preloaded_products=products,
+    )
+    assert preloaded == standalone
+
+
+def test_detail_route_threads_context_into_repeated_panels():
+    source = getsource(page.opportunity_detail_page)
+    assert "load_opportunity_detail_context(opportunity_id)" in source
+    assert "preloaded_documents=context.resolved_documents" in source
+    assert "preloaded_items=context.opportunity_items" in source
+    assert "preloaded_products=context.active_products" in source
+    assert "preloaded_candidates=context.odoo_candidates" in source
+    assert "preloaded_items=context.outbox_items" in source
+    assert "preloaded=context.next_action_preloaded()" in source
+
+
+def test_detail_read_path_contains_no_schema_initialization_or_ddl():
+    readers = (
+        page.load_opportunity_detail_context,
+        commercial.get_customer_for_opportunity,
+        commercial.list_commercial_documents,
+        commercial.list_customers,
+        operation.get_operation_links,
+    )
+    forbidden = ("ensure_", "CREATE TABLE", "CREATE INDEX", "ALTER TABLE")
+    for reader in readers:
+        source = getsource(reader)
+        for token in forbidden:
+            assert token not in source, (reader.__name__, token)
+
+    route_source = getsource(page.opportunity_detail_page)
+    assert "ensure_pending_task_for_next_action" not in route_source
+    assert "materialized_task" not in route_source
+
+
+def test_context_next_action_customer_aliases_preserve_evidence_contract():
+    context = page.OpportunityDetailContext(
+        opportunity_id="opp", opportunity={"id": "opp", "local_customer_id": "customer"},
+        customer={"id": "customer", "email": "a@b.pt", "street_name": "Rua",
+                  "postal_zone": "1000-001", "city_name": "Lisboa"},
+    )
+    customer = context.next_action_preloaded()["customer"]
+    assert customer["billing_email"] == "a@b.pt"
+    assert customer["address"] == "Rua"
+    assert customer["postal_code"] == "1000-001"
+    assert customer["city"] == "Lisboa"

tests/test_v4928_1_5_37_odoo_sync_panel_static.py

@@ -6,7 +6,9 @@ ROOT = Path(__file__).resolve().parents[1]
 def test_odoo_status_panel_added_to_opportunity_detail():
     page = (ROOT / "app/admin_ui/pages/opportunities.py").read_text()
     assert '<a href="#odoo">Odoo</a>' in page
-    assert '<div id="odoo">{odoo_status_panel_html(opportunity_id)}</div>' in page
+    assert '<div id="odoo">{odoo_status_panel_html(' in page
+    assert 'preloaded_links=' in page
+    assert 'preloaded_candidates=context.odoo_candidates' in page
     assert '@router.get("/opportunities/{opportunity_id}/partials/odoo-status"' in page
     assert '@router.post("/opportunities/{opportunity_id}/odoo/sync-status")' in page
     assert '@router.post("/opportunities/{opportunity_id}/odoo/link-candidate/{item_id}")' in page

tests/test_v4928_1_5_96_materialize_send_invoice_task_static.py

@@ -9,11 +9,12 @@ def test_materializer_limits_to_send_invoice_and_is_idempotent():
     assert 'ON CONFLICT (idempotency_key) DO NOTHING' in text
 
 
-def test_opportunity_page_materializes_next_action_task_and_refreshes_tasks():
+def test_opportunity_page_is_read_only_and_does_not_materialize_tasks():
     text = Path('app/admin_ui/pages/opportunities.py').read_text()
-    assert 'ensure_pending_task_for_next_action' in text
-    assert 'materialized_task.get("created")' in text
-    assert 'tasks = list_opportunity_tasks(opportunity_id, limit=100)' in text
+    detail = text.split('async def opportunity_detail_page', 1)[1].split('@router.', 1)[0]
+    assert 'ensure_pending_task_for_next_action' not in detail
+    assert 'materialized_task' not in detail
+    assert 'GET detail is strictly read-only' in detail
 
 
 def test_repair_script_uses_central_next_action():