tests/test_api.py

"""API contract — FastAPI TestClient with the pipeline mocked (no model, no network).

conftest sets ALLOWED_API_KEYS / WARM_MODEL_ON_STARTUP / rate limits before this
module imports `api`. `run` is monkeypatched to a canned result dict so we exercise
auth, the response schema, error mapping, and the batch job flow only.
"""

from __future__ import annotations

import time

import pytest
from fastapi.testclient import TestClient

from lawn_estimator import api
from lawn_estimator.db import (
    ApiKeyStore,
    AuditLog,
    ConfigStore,
    LeadStore,
    MeasurementEditStore,
    RunLedger,
    TenantStore,
    UserStore,
    connect,
)
from lawn_estimator.pricing import PricingConfig, Service
from lawn_estimator.tenants import ServiceArea, Tenant

KEY = {"X-API-Key": "test-key"}

CANNED_RESULT = {
    "rgb_veg_sqft": 1200.0,
    "lidar_lawn_sqft": 4669.8,
    "lawn_sqft": 4669.8,
    "parcel_area_sqft": 6504.6,
    "estimation_area_sqft": 7104.8,
    "ground_sampled_sqft": 5346.3,
    "row_buffer_ft": 12.0,
    "formatted_address": "5534 Mayberry St, Omaha, NE 68106",
    "zip_code": "68106",
    "visualization_path": "/does/not/exist.png",
    "imagery_source": "naip",
    "region": "Douglas County, NE",
    "method": "lidar+rgb",
    "confidence": "high",
    "extensions": [{"street": "Mayberry St", "area_sqft": 600.2,
                    "centroid": [-96.0, 41.26], "street_facing": True}],
    "extension_total_sqft": 600.2,
    "warning": None,
}


@pytest.fixture(scope="module")
def client():
    with TestClient(api.app) as c:
        yield c


class _FakeGeocode:
    """Minimal stand-in for GeocodeResult — the API only reads .zip_code for gating."""

    def __init__(self, zip_code="68106"):
        self.zip_code = zip_code
        self.formatted_address = "5534 Mayberry St, Omaha, NE 68106"
        self.latitude, self.longitude = 41.26, -96.0


def _patch_resolve(monkeypatch, zip_code="68106"):
    monkeypatch.setattr(api, "resolve", lambda address: (object(), _FakeGeocode(zip_code)))


@pytest.fixture
def mock_run(monkeypatch):
    def _fake_run(address, imagery="auto", **kwargs):
        return dict(CANNED_RESULT)

    monkeypatch.setattr(api, "run", _fake_run)
    _patch_resolve(monkeypatch, "68106")  # geocode-only pre-pipeline gate
    return _fake_run


@pytest.fixture(autouse=True)
def _default_resolve(monkeypatch):
    # _run_one now geocodes (api.resolve) before the pipeline; keep every test off the
    # network with an in-Coverage default. Tests needing a specific ZIP re-patch it.
    _patch_resolve(monkeypatch, "68106")


@pytest.fixture(autouse=True)
def _reset_email_throttle():
    # The per-recipient send throttle is a module-level dict; clear it between
    # tests so the email cases don't consume each other's quota.
    api._EMAIL_SEND_LOG.clear()
    yield


# ---------------------------------------------------------------------------
# Auth + health
# ---------------------------------------------------------------------------
def test_health_needs_no_auth(client):
    r = client.get("/health")
    assert r.status_code == 200
    assert r.json() == {"status": "ok", "service": "lawn-estimator"}


# ---------------------------------------------------------------------------
# Dashboard auth (P2a) — Clerk identity + our tenant-scoped authorization
# ---------------------------------------------------------------------------
def test_dashboard_me_gate(client, monkeypatch):
    monkeypatch.setattr(api, "USERS", UserStore(connect(":memory:")))
    api.USERS.create("acme", "clerk_owner", "o@acme.com", "owner", status="active")
    api.USERS.create("acme", "clerk_invited", "p@acme.com", "staff", status="invited")

    # No Clerk session (fail-closed: CLERK_SECRET_KEY unset) → 401.
    assert client.get("/dashboard/me").status_code == 401

    # A verified Clerk session resolves to OUR user (tenant + role).
    monkeypatch.setattr(api, "verify_clerk_session", lambda t: "clerk_owner")
    body = client.get("/dashboard/me", headers={"Authorization": "Bearer x"}).json()
    assert body["tenant_id"] == "acme" and body["role"] == "owner" and body["email"] == "o@acme.com"

    # Verified Clerk user with NO provisioned row → 403 (invite-only, no open signup).
    monkeypatch.setattr(api, "verify_clerk_session", lambda t: "clerk_ghost")
    assert client.get("/dashboard/me", headers={"Authorization": "Bearer x"}).status_code == 403

    # Provisioned but not yet active (invited) → 403.
    monkeypatch.setattr(api, "verify_clerk_session", lambda t: "clerk_invited")
    assert client.get("/dashboard/me", headers={"Authorization": "Bearer x"}).status_code == 403


def test_require_owner_gates_staff():
    import pytest
    from fastapi import HTTPException

    owner = api.DashboardUser(id=1, tenant_id="acme", role="owner", email="o@x.com")
    assert api.require_owner(user=owner) is owner
    staff = api.DashboardUser(id=2, tenant_id="acme", role="staff", email="s@x.com")
    with pytest.raises(HTTPException) as e:
        api.require_owner(user=staff)
    assert e.value.status_code == 403


def test_dashboard_dev_token_shim(monkeypatch):
    monkeypatch.setattr(api, "CLERK_SECRET_KEY", None)          # pre-Clerk
    monkeypatch.setattr(api, "DASHBOARD_DEV_TOKEN", "devtok")
    monkeypatch.setattr(api, "DASHBOARD_DEV_CLERK_ID", "dev_user")
    assert api.verify_clerk_session("devtok") == "dev_user"
    assert api.verify_clerk_session("wrong") is None
    monkeypatch.setattr(api, "CLERK_SECRET_KEY", "sk_live")     # once Clerk is on, dev token is dead
    assert api.verify_clerk_session("devtok") is None


def test_dashboard_config_editor_flow(client, monkeypatch):
    db = connect(":memory:")   # one DB for all the dashboard stores
    monkeypatch.setattr(api, "USERS", UserStore(db))
    monkeypatch.setattr(api, "TENANT_STORE", TenantStore(db))
    monkeypatch.setattr(api, "CONFIG_STORE", ConfigStore(db))
    monkeypatch.setattr(api, "AUDIT", AuditLog(db))
    monkeypatch.setattr(api, "TENANTS", {})
    api.USERS.create("acme", "clerk_owner", "o@acme.com", "owner", status="active")
    api.USERS.create("acme", "clerk_staff", "s@acme.com", "staff", status="active")
    hdr = {"Authorization": "Bearer x"}
    valid = {"company": "Acme", "currency": "USD", "presentation": "firm",
             "services": [{"id": "mowing", "label": "Mowing", "rate_per_1000_sqft": 5, "min_charge": 45}]}

    # --- as OWNER ---
    monkeypatch.setattr(api, "verify_clerk_session", lambda t: "clerk_owner")
    got = client.get("/dashboard/config", headers=hdr).json()
    assert got["live"] is None and got["draft"] is None
    assert client.put("/dashboard/config/draft", headers=hdr, json={"config": valid}).status_code == 200
    assert client.get("/dashboard/config", headers=hdr).json()["draft"]["company"] == "Acme"

    pub = client.post("/dashboard/config/publish", headers=hdr, json={"config": valid}).json()
    assert pub["ok"] is True
    assert api.TENANT_STORE.get("acme")["company"] == "Acme"     # live in the DB
    assert "acme" in api.TENANTS                                  # widget hot-reloaded, no redeploy

    bad = {**valid, "branding": {"cta_url": "javascript:alert(1)"}}   # XSS attempt
    pubbad = client.post("/dashboard/config/publish", headers=hdr, json={"config": bad}).json()
    assert pubbad["ok"] is False and any("cta_url" in e for e in pubbad["errors"])

    actions = {e["action"] for e in client.get("/dashboard/audit", headers=hdr).json()["entries"]}
    assert {"config.publish", "config.draft.save"} <= actions

    # --- as STAFF: may view, may not edit (owner-only) ---
    monkeypatch.setattr(api, "verify_clerk_session", lambda t: "clerk_staff")
    assert client.get("/dashboard/config", headers=hdr).status_code == 200
    assert client.put("/dashboard/config/draft", headers=hdr, json={"config": valid}).status_code == 403
    assert client.post("/dashboard/config/publish", headers=hdr, json={"config": valid}).status_code == 403


def test_dashboard_page_is_served(client):
    r = client.get("/dashboard")
    assert r.status_code == 200 and "Lawn Dashboard" in r.text


def test_dashboard_batch_runs_and_is_tenant_scoped(client, mock_run, monkeypatch):
    monkeypatch.setattr(api, "USERS", UserStore(connect(":memory:")))
    api.USERS.create("acme", "clerk_acme", "a@acme.com", "staff", status="active")
    api.USERS.create("beta", "clerk_beta", "b@beta.com", "owner", status="active")
    hdr = {"Authorization": "Bearer x"}

    monkeypatch.setattr(api, "verify_clerk_session", lambda t: "clerk_acme")
    r = client.post("/dashboard/batch", headers=hdr,
                    json={"addresses": ["5534 Mayberry St, Omaha, NE 68106", "5819 Hickory St"]})
    assert r.status_code == 200 and r.json()["total"] == 2
    job_id = r.json()["job_id"]

    status = {}
    deadline = time.time() + 5
    while time.time() < deadline:
        status = client.get(f"/dashboard/jobs/{job_id}", headers=hdr).json()
        if status["status"] == "done":
            break
        time.sleep(0.05)
    assert status["status"] == "done" and status["completed"] == 2
    assert all(res["status"] == "ok" for res in status["results"])

    # Another tenant can't see the job (404, not 403 — no existence leak across tenants).
    monkeypatch.setattr(api, "verify_clerk_session", lambda t: "clerk_beta")
    assert client.get(f"/dashboard/jobs/{job_id}", headers=hdr).status_code == 404


def test_admin_page_is_served(client):
    r = client.get("/admin")
    assert r.status_code == 200 and "Platform Admin" in r.text


def test_admin_overview_gate_and_cross_tenant(client, monkeypatch):
    monkeypatch.setattr(api, "METER", RunLedger(connect(":memory:")))
    api.METER.record("acme", "1", api.BATCH, duration_s=20.0)
    api.METER.record_failure("acme", "x", api.SINGLE, error_class="ValueError", outcome="error")
    api.METER.record("beta", "9", api.BATCH, duration_s=10.0)
    monkeypatch.setattr(api, "CLERK_SECRET_KEY", None)          # pre-Clerk dev path
    monkeypatch.setattr(api, "ADMIN_DEV_TOKEN", "admtok")
    monkeypatch.setattr(api, "PLATFORM_ADMIN_EMAILS", {"me@platform.com"})

    assert client.get("/admin/overview").status_code == 403                                   # no token
    assert client.get("/admin/overview", headers={"Authorization": "Bearer nope"}).status_code == 403
    d = client.get("/admin/overview", headers={"Authorization": "Bearer admtok"}).json()      # dev admin
    assert d["totals"]["tenants"] == 2 and d["totals"]["total"] == 3 and d["totals"]["errors"] == 1
    assert d["tenants"][0]["tenant"] == "acme"                  # most active first, cross-tenant


def test_admin_provision_tenant_end_to_end(client, monkeypatch, mock_run):
    db = connect(":memory:")
    monkeypatch.setattr(api, "CLERK_SECRET_KEY", None)
    monkeypatch.setattr(api, "ADMIN_DEV_TOKEN", "admtok")
    monkeypatch.setattr(api, "CLERK", None)
    monkeypatch.setattr(api, "USERS", UserStore(db))
    monkeypatch.setattr(api, "AUDIT", AuditLog(db))
    monkeypatch.setattr(api, "KEYS", ApiKeyStore(db))
    monkeypatch.setattr(api, "TENANT_STORE", TenantStore(db))
    monkeypatch.setattr(api, "TENANTS", {})
    monkeypatch.setattr(api, "WIDGET_KEYS", {})
    monkeypatch.setattr(api, "ALLOWED_API_KEYS", {})
    hdr = {"Authorization": "Bearer admtok"}

    r = client.post("/admin/tenants", headers=hdr, json={
        "tenant_id": "acme-lawn", "company": "Acme Lawn", "owner_email": "o@acme.com",
        "origin": "https://acme.com"})
    assert r.status_code == 200
    body = r.json()
    key = body["publishable_key"]
    assert body["tenant_id"] == "acme-lawn" and body["invited"] == "o@acme.com" and key

    # Registered live — no restart: in the registry, key map, DB key store, and config store.
    assert "acme-lawn" in api.TENANTS and api.WIDGET_KEYS[key] == "acme-lawn"
    assert api.KEYS.key_for_tenant("acme-lawn") == key
    assert api.TENANT_STORE.get("acme-lawn")["company"] == "Acme Lawn"
    assert any(m["email"] == "o@acme.com" and m["role"] == "owner" for m in api.USERS.by_tenant("acme-lawn"))

    # Appears in the admin clients list.
    lst = client.get("/admin/tenants", headers=hdr).json()["tenants"]
    assert any(t["tenant_id"] == "acme-lawn" and t["has_key"] and t["members"] >= 1 for t in lst)

    # The minted key authenticates a widget /quote — domain-locked to the tenant's origin.
    ok = client.post("/quote", json={"address": "x"}, headers={"X-API-Key": key, "Origin": "https://acme.com"})
    assert ok.status_code == 200
    bad = client.post("/quote", json={"address": "x"}, headers={"X-API-Key": key, "Origin": "https://evil.com"})
    assert bad.status_code == 403

    # Duplicate id → 409; bad slug / missing fields → 422.
    assert client.post("/admin/tenants", headers=hdr, json={
        "tenant_id": "acme-lawn", "company": "X", "owner_email": "a@b.com"}).status_code == 409
    assert client.post("/admin/tenants", headers=hdr, json={
        "tenant_id": "Bad Slug!", "company": "X", "owner_email": "a@b.com"}).status_code == 422
    assert client.post("/admin/tenants", headers=hdr, json={
        "tenant_id": "ok-slug", "company": "X", "owner_email": "not-an-email"}).status_code == 422


def test_admin_is_a_separate_realm_from_a_tenant_session(client, monkeypatch):
    # A valid TENANT (operator) session must NOT reach /admin — the realms are distinct.
    monkeypatch.setattr(api, "USERS", UserStore(connect(":memory:")))
    api.USERS.create("acme", "clerk_owner", "o@acme.com", "owner", status="active")
    monkeypatch.setattr(api, "verify_clerk_session", lambda t: "clerk_owner")   # valid tenant session
    monkeypatch.setattr(api, "CLERK_SECRET_KEY", None)
    monkeypatch.setattr(api, "ADMIN_DEV_TOKEN", None)          # no admin credential
    assert client.get("/admin/overview", headers={"Authorization": "Bearer x"}).status_code == 403


def test_quote_failure_records_a_rejected_run(client, monkeypatch):
    _register_tenant(monkeypatch, ACME, key="acmekey")
    monkeypatch.setattr(api, "METER", RunLedger(connect(":memory:")))

    def boom(_address):
        raise ValueError("outside our service area")

    monkeypatch.setattr(api, "resolve", boom)
    r = client.post("/quote", json={"address": "x"}, headers={"X-API-Key": "acmekey"})
    assert r.status_code == 422
    s = api.METER.stats("acme")
    assert s["rejected"] == 1 and s["errors"] == 0 and s["ok"] == 0   # failure logged, not billable


def test_dashboard_measurement_edit_batch_only(client, monkeypatch):
    db = connect(":memory:")
    monkeypatch.setattr(api, "USERS", UserStore(db))
    monkeypatch.setattr(api, "METER", RunLedger(db))
    monkeypatch.setattr(api, "MEDITS", MeasurementEditStore(db))
    monkeypatch.setattr(api, "AUDIT", AuditLog(db))
    api.USERS.create("acme", "clerk_acme", "a@acme.com", "owner", status="active")
    api.METER.record("acme", "1 Batch St", api.BATCH, lawn_sqft=4000.0, result_id="rbatch", zip="68106")
    api.METER.record("acme", "2 Cust St", api.SINGLE, lawn_sqft=3000.0, result_id="rcust", zip="68106")
    monkeypatch.setattr(api, "verify_clerk_session", lambda t: "clerk_acme")
    hdr = {"Authorization": "Bearer x"}

    # A batch measurement can be corrected → stored (original vs edited) + audited + re-priced.
    r = client.post("/dashboard/measurement/edit", headers=hdr,
                    json={"result_id": "rbatch", "edited_sqft": 4500})
    assert r.status_code == 200 and r.json()["edited_sqft"] == 4500
    assert api.MEDITS.recent("acme")[0]["original_sqft"] == 4000.0
    assert any(e["action"] == "measurement.edit" for e in api.AUDIT.recent("acme"))

    # A customer/widget quote is LOCKED — never editable (it's what the homeowner agreed to).
    assert client.post("/dashboard/measurement/edit", headers=hdr,
                       json={"result_id": "rcust", "edited_sqft": 3500}).status_code == 403
    # Unknown result id → 404; non-positive area → 422.
    assert client.post("/dashboard/measurement/edit", headers=hdr,
                       json={"result_id": "nope", "edited_sqft": 100}).status_code == 404
    assert client.post("/dashboard/measurement/edit", headers=hdr,
                       json={"result_id": "rbatch", "edited_sqft": 0}).status_code == 422


def test_dashboard_snippet_returns_the_tenants_widget_key(client, monkeypatch):
    monkeypatch.setattr(api, "USERS", UserStore(connect(":memory:")))
    api.USERS.create("acme", "clerk_acme", "a@acme.com", "owner", status="active")
    api.USERS.create("beta", "clerk_beta", "b@beta.com", "owner", status="active")
    monkeypatch.setattr(api, "WIDGET_KEYS", {"pub-acme": "acme"})   # acme issued a key; beta not
    hdr = {"Authorization": "Bearer x"}
    monkeypatch.setattr(api, "verify_clerk_session", lambda t: "clerk_acme")
    body = client.get("/dashboard/snippet", headers=hdr).json()
    assert body["widget_key"] == "pub-acme" and body["host"]
    # Behind a TLS proxy the embed host must be https (else mixed-content-blocked on the tenant's
    # https page): X-Forwarded-Proto: https upgrades the http base_url TestClient reports.
    fwd = client.get("/dashboard/snippet", headers={**hdr, "X-Forwarded-Proto": "https"}).json()
    assert fwd["host"].startswith("https://")
    monkeypatch.setattr(api, "verify_clerk_session", lambda t: "clerk_beta")   # scoped: not acme's key
    assert client.get("/dashboard/snippet", headers=hdr).json()["widget_key"] is None


def test_dashboard_usage_and_leads_are_session_scoped(client, monkeypatch):
    db = connect(":memory:")
    monkeypatch.setattr(api, "USERS", UserStore(db))
    monkeypatch.setattr(api, "METER", RunLedger(db))
    monkeypatch.setattr(api, "LEADS", LeadStore(db))
    api.USERS.create("acme", "clerk_owner", "o@acme.com", "owner", status="active")
    api.LEADS.record("acme", "widget", "1 A St", email="a@x.com")
    api.LEADS.record("beta", "widget", "9 Z St", email="z@x.com")   # other tenant
    monkeypatch.setattr(api, "verify_clerk_session", lambda t: "clerk_owner")
    hdr = {"Authorization": "Bearer x"}
    leads = client.get("/dashboard/leads", headers=hdr).json()["leads"]
    assert {lead["address"] for lead in leads} == {"1 A St"}        # never beta's (tenant-scoped)
    assert client.get("/dashboard/usage", headers=hdr).json()["tenant"] == "acme"


def test_widget_page_is_served(client):
    # The embeddable widget is a public, self-contained page (its /quote calls need a key).
    r = client.get("/widget")
    assert r.status_code == 200
    assert 'id="address"' in r.text and "Get my price" in r.text


def test_quote_missing_key_is_rejected(client):
    r = client.post("/quote", json={"address": "5534 Mayberry St, Omaha, NE 68106"})
    assert r.status_code == 422  # required X-API-Key header absent


def test_quote_bad_key_is_unauthorized(client):
    r = client.post("/quote", json={"address": "5534 Mayberry St, Omaha, NE 68106"},
                    headers={"X-API-Key": "not-the-key"})
    assert r.status_code == 401


def test_key_parsing_supports_labels_and_bare_keys(monkeypatch):
    monkeypatch.setenv("ALLOWED_API_KEYS", "website:G1Bvsecret, ownerbarekey ,, partner : p91dsecret ")
    keys = api._load_api_keys()
    assert keys["G1Bvsecret"] == "website"
    assert keys["ownerbarekey"] == "owne…"  # bare key -> prefix label
    assert keys["p91dsecret"] == "partner"  # whitespace around label/key is stripped


def test_labeled_key_authenticates_with_bare_key_and_logs_label(client, mock_run, monkeypatch, caplog):
    # clients send ONLY the key — the label is server-side, for the logs
    monkeypatch.setattr(api, "ALLOWED_API_KEYS", {"G1Bvsecret": "website"})
    with caplog.at_level("INFO"):
        r = client.post("/quote", json={"address": "5534 Mayberry St, Omaha, NE 68106"},
                        headers={"X-API-Key": "G1Bvsecret"})
    assert r.status_code == 200
    assert any("client=website" in rec.message for rec in caplog.records)

    r = client.post("/quote", json={"address": "5534 Mayberry St, Omaha, NE 68106"},
                    headers={"X-API-Key": "website:G1Bvsecret"})
    assert r.status_code == 401  # the label is not part of the credential


# ---------------------------------------------------------------------------
# /quote contract
# ---------------------------------------------------------------------------
def test_quote_success_returns_full_contract(client, mock_run):
    r = client.post("/quote", json={"address": "5534 Mayberry St, Omaha, NE 68106"}, headers=KEY)
    assert r.status_code == 200
    body = r.json()

    assert body["address"] == "5534 Mayberry St, Omaha, NE 68106"
    assert body["formatted_address"] == "5534 Mayberry St, Omaha, NE 68106"
    assert body["lawn_sqft"] == 4669.8
    assert body["parcel_sqft"] == 6504.6
    assert body["estimation_area_sqft"] == 7104.8
    assert body["ground_sampled_sqft"] == 5346.3
    assert body["row_buffer_ft"] == 12.0
    assert body["pct_of_estimation"] == pytest.approx(65.7)
    assert body["imagery_source"] == "naip"
    assert body["region"] == "Douglas County, NE"
    assert body["method"] == "lidar+rgb"
    assert body["confidence"] == "high"
    assert body["extension_total_sqft"] == 600.2
    assert body["extensions"][0]["street"] == "Mayberry St"
    # The viz is internal-only, addressed via result_id (/viz/{id}.png); the
    # explicit visualization_url field was dropped from the web-dev contract.
    assert "visualization_url" not in body
    assert body["result_id"]
    assert body["email_sent"] is None  # no email requested
    assert body["warning"] is None
    assert body["status"] == "ok"

    # Pricing is now computed from lawn_sqft (4669.8 → mowing hits its $45 minimum).
    pricing = body["pricing"]
    assert pricing is not None
    assert pricing["currency"] == "USD"
    assert pricing["presentation"] == "estimate"
    assert len(pricing["line_items"]) == 4
    mowing = next(i for i in pricing["line_items"] if i["service"] == "mowing")
    assert mowing["price"] == 45.0 and mowing["min_applied"] is True

    # Default tenant has no service area → serves everywhere, ZIP echoed.
    assert body["zip_code"] == "68106"


def _register_tenant(monkeypatch, tenant, key="acmekey"):
    """Point auth + registry at a single test tenant reachable by `key`."""
    monkeypatch.setattr(api, "ALLOWED_API_KEYS", {key: tenant.id})
    monkeypatch.setattr(api, "TENANTS", {tenant.id: tenant})


# A tenant that only serves ZIP 68022, with a doubled ($/sqft × travel factor) package there.
ACME = Tenant(
    id="acme",
    pricing=PricingConfig("Acme", "USD", "firm", (Service("mowing", "Mowing", 5, 45),)),
    areas=(
        ServiceArea(
            name="Outer",
            zips=frozenset({"68022"}),
            pricing=PricingConfig("Acme", "USD", "firm", (Service("mowing", "Mowing", 5, 0),)),
            travel_time_factor=2.0,
        ),
    ),
)


def test_quote_in_area_uses_per_area_pricing(client, monkeypatch, mock_run):
    _register_tenant(monkeypatch, ACME)
    _patch_resolve(monkeypatch, "68022")  # inside ACME's service area
    r = client.post("/quote", json={"address": "x, Elkhorn, NE 68022"}, headers={"X-API-Key": "acmekey"})
    assert r.status_code == 200
    mowing = next(i for i in r.json()["pricing"]["line_items"] if i["service"] == "mowing")
    # 4669.8 × $5/1k × 2.0 travel factor, min $0 → factor is visible.
    assert mowing["price"] == pytest.approx(round(4669.8 * 5 / 1000 * 2.0, 2))


def test_usage_counts_billable_quotes_and_dedups_customer_repeats(client, monkeypatch, mock_run):
    monkeypatch.setattr(api, "METER", RunLedger(connect(":memory:")))  # fresh, deterministic
    addr = {"address": "5534 Mayberry St, Omaha, NE 68106"}
    assert client.post("/quote", json=addr, headers=KEY).status_code == 200
    assert client.post("/quote", json=addr, headers=KEY).status_code == 200  # same address
    body = client.get("/usage", headers=KEY).json()
    assert body["billable_quotes"] == 1   # the customer duplicate isn't charged


def test_leads_endpoint_is_tenant_scoped_and_secret_only(client, monkeypatch):
    monkeypatch.setattr(api, "LEADS", LeadStore(connect(":memory:")))
    monkeypatch.setattr(api, "ALLOWED_API_KEYS", {"acme-secret": "acme", "beta-secret": "beta"})
    api.LEADS.record("acme", "widget", "1 A St", email="a@x.com")
    api.LEADS.record("acme", "quote", "2 B St", email="b@x.com", lawn_sqft=4200.0, price_total=90.0)
    api.LEADS.record("beta", "widget", "9 Z St", email="z@x.com")   # another tenant

    body = client.get("/leads", headers={"X-API-Key": "acme-secret"}).json()
    assert body["tenant"] == "acme" and body["count"] == 2
    assert body["leads"][0]["address"] == "2 B St"                  # newest first
    assert body["leads"][0]["lawn_sqft"] == 4200.0
    assert {lead["address"] for lead in body["leads"]} == {"1 A St", "2 B St"}   # never beta's
    # Business data: a publishable widget key (not in ALLOWED_API_KEYS) can't read it.
    assert client.get("/leads", headers={"X-API-Key": "pub-k"}).status_code == 401
    assert client.get("/leads").status_code == 422   # required key header absent


def test_quote_out_of_area_is_rejected_without_measuring(client, monkeypatch):
    _register_tenant(monkeypatch, ACME)
    _patch_resolve(monkeypatch, "68106")  # in Coverage, outside ACME's service area

    def _run_must_not_run(*a, **k):
        raise AssertionError("pipeline must not run for an out-of-area address")

    monkeypatch.setattr(api, "run", _run_must_not_run)
    r = client.post("/quote", json={"address": "5534 Mayberry St, Omaha, NE 68106"}, headers={"X-API-Key": "acmekey"})
    assert r.status_code == 422
    assert "don't currently serve" in r.json()["detail"]


_CENTRAL_POLY = {
    "type": "Polygon",
    "coordinates": [[[-96.02, 41.22], [-95.95, 41.22], [-95.95, 41.28], [-96.02, 41.28], [-96.02, 41.22]]],
}


def test_service_area_zips_endpoint(client, monkeypatch):
    db = connect(":memory:")
    monkeypatch.setattr(api, "USERS", UserStore(db))
    api.USERS.create("acme", "clerk_owner", "o@acme.com", "owner", status="active")
    monkeypatch.setattr(api, "verify_clerk_session", lambda t: "clerk_owner")
    r = client.post("/dashboard/service-area/zips", headers={"Authorization": "Bearer x"},
                    json={"polygon": _CENTRAL_POLY})
    assert r.status_code == 200
    zips = r.json()["zips"]
    assert "68106" in zips and "68022" not in zips     # Elkhorn is outside the drawn box


def test_quote_polygon_gate_rejects_outside(client, monkeypatch):
    t = Tenant(id="acme", pricing=PricingConfig("Acme", "USD", "firm", (Service("mowing", "Mowing", 5, 45),)),
               service_area_polygon=_CENTRAL_POLY)
    _register_tenant(monkeypatch, t)
    g = _FakeGeocode("68106")
    g.latitude, g.longitude = 41.00, -96.50    # inside Coverage, outside the drawn polygon
    monkeypatch.setattr(api, "resolve", lambda a: (object(), g))

    def _must_not_run(*a, **k):
        raise AssertionError("pipeline must not run for an out-of-polygon address")

    monkeypatch.setattr(api, "run", _must_not_run)
    r = client.post("/quote", json={"address": "x"}, headers={"X-API-Key": "acmekey"})
    assert r.status_code == 422 and "don't currently serve" in r.json()["detail"]


def test_quote_polygon_gate_allows_inside(client, monkeypatch, mock_run):
    # mock_run patches resolve to the default in-box point (41.26, -96.0) + a canned result.
    t = Tenant(id="acme", pricing=PricingConfig("Acme", "USD", "firm", (Service("mowing", "Mowing", 5, 45),)),
               service_area_polygon=_CENTRAL_POLY)
    _register_tenant(monkeypatch, t)
    r = client.post("/quote", json={"address": "x"}, headers={"X-API-Key": "acmekey"})
    assert r.status_code == 200


def test_geocode_failure_is_sanitized_for_the_homeowner(client, monkeypatch):
    # The raw geocoder error names county internals — must never reach the homeowner.
    def _raise(address):
        raise ValueError("Address not found in Sarpy County address points: 'asdf'")

    monkeypatch.setattr(api, "resolve", _raise)
    r = client.post("/quote", json={"address": "asdf nonsense"}, headers=KEY)
    assert r.status_code == 422
    detail = r.json()["detail"]
    assert "County" not in detail and "address points" not in detail
    assert "double-check" in detail.lower()


# --- Publishable widget key + domain-lock (W2) ------------------------------

WIDGET_TENANT = Tenant(
    id="wco",
    pricing=PricingConfig("WCo", "USD", "firm", (Service("mowing", "Mowing", 5, 45),)),
    allowed_origins=frozenset({"https://wco.com"}),
)


def _register_widget(monkeypatch):
    monkeypatch.setattr(api, "ALLOWED_API_KEYS", {"secret-k": "wco"})
    monkeypatch.setattr(api, "WIDGET_KEYS", {"pub-k": "wco"})
    monkeypatch.setattr(api, "TENANTS", {"wco": WIDGET_TENANT})


def test_publishable_key_quotes_from_allowed_origin(client, monkeypatch, mock_run):
    _register_widget(monkeypatch)
    r = client.post("/quote", json={"address": "5534 Mayberry St, Omaha, NE 68106"},
                    headers={"X-API-Key": "pub-k", "Origin": "https://wco.com"})
    assert r.status_code == 200


def test_publishable_key_rejected_from_other_origin(client, monkeypatch, mock_run):
    _register_widget(monkeypatch)
    r = client.post("/quote", json={"address": "5534 Mayberry St, Omaha, NE 68106"},
                    headers={"X-API-Key": "pub-k", "Origin": "https://evil.com"})
    assert r.status_code == 403


def test_publishable_key_rejected_without_origin(client, monkeypatch, mock_run):
    _register_widget(monkeypatch)
    r = client.post("/quote", json={"address": "5534 Mayberry St, Omaha, NE 68106"},
                    headers={"X-API-Key": "pub-k"})
    assert r.status_code == 403


def test_publishable_key_denied_on_batch(client, monkeypatch):
    _register_widget(monkeypatch)
    r = client.post("/quote/batch", json={"addresses": ["a"]},
                    headers={"X-API-Key": "pub-k", "Origin": "https://wco.com"})
    assert r.status_code == 401   # batch is secret-key only (require_api_key)


def test_publishable_key_denied_on_usage(client, monkeypatch):
    _register_widget(monkeypatch)
    r = client.get("/usage", headers={"X-API-Key": "pub-k"})
    assert r.status_code == 401


def test_secret_key_quotes_without_origin(client, monkeypatch, mock_run):
    _register_widget(monkeypatch)
    r = client.post("/quote", json={"address": "5534 Mayberry St, Omaha, NE 68106"},
                    headers={"X-API-Key": "secret-k"})   # secret key: no origin check
    assert r.status_code == 200


# --- W2b CAPTCHA + W3b config -----------------------------------------------

def test_widget_config_exposes_public_config(client, monkeypatch):
    monkeypatch.setattr(api, "TURNSTILE_SITE_KEY", "site-abc")
    monkeypatch.setattr(api, "GOOGLE_PLACES_SERVER_KEY", "server-key")  # server-side; NOT exposed
    body = client.get("/widget/config").json()  # no key → no branding
    # The Google key is NOT in the response — only whether autocomplete is available.
    assert body == {"turnstile_site_key": "site-abc", "autocomplete": True, "branding": None,
                    "questions": []}


def test_autocomplete_proxies_google_and_is_domain_locked(client, monkeypatch):
    _register_widget(monkeypatch)
    monkeypatch.setattr(api, "GOOGLE_PLACES_SERVER_KEY", "server-key")

    class FakeResp:
        def json(self):
            return {"status": "OK", "predictions": [
                {"description": "5534 Mayberry St, Omaha, NE, USA", "place_id": "p1"},
                {"description": "5534 Maple Ave, Omaha, NE, USA", "place_id": "p2"},
            ]}

    import requests
    monkeypatch.setattr(requests, "get", lambda *a, **k: FakeResp())

    ok = client.get("/autocomplete?q=5534 May",
                    headers={"X-API-Key": "pub-k", "Origin": "https://wco.com"})
    assert ok.status_code == 200
    preds = ok.json()["predictions"]
    assert preds[0] == {"description": "5534 Mayberry St, Omaha, NE, USA", "place_id": "p1"}
    # Domain-locked (no CAPTCHA, but origin still enforced), and missing key is rejected.
    assert client.get("/autocomplete?q=5534 May",
                      headers={"X-API-Key": "pub-k", "Origin": "https://evil.com"}).status_code == 403
    assert client.get("/autocomplete?q=5534 May").status_code == 422  # required key header absent


def test_autocomplete_degrades_to_empty(client, monkeypatch):
    _register_widget(monkeypatch)
    # Authorized, but no server key configured and a too-short query → graceful [].
    monkeypatch.setattr(api, "GOOGLE_PLACES_SERVER_KEY", None)
    r = client.get("/autocomplete?q=5534 May",
                   headers={"X-API-Key": "pub-k", "Origin": "https://wco.com"})
    assert r.status_code == 200 and r.json() == {"predictions": []}


def test_autocomplete_logs_google_denial_and_still_degrades(client, monkeypatch, caplog):
    _register_widget(monkeypatch)
    monkeypatch.setattr(api, "GOOGLE_PLACES_SERVER_KEY", "server-key")

    class DeniedResp:
        def json(self):
            return {"status": "REQUEST_DENIED", "error_message": "The provided API key is invalid."}

    import requests
    monkeypatch.setattr(requests, "get", lambda *a, **k: DeniedResp())

    with caplog.at_level("WARNING"):
        r = client.get("/autocomplete?q=5534 May",
                       headers={"X-API-Key": "pub-k", "Origin": "https://wco.com"})
    # Still fails safe for the user (empty list), but the WHY is now in the logs — and the
    # message carries Google's reason, never our key.
    assert r.status_code == 200 and r.json() == {"predictions": []}
    assert "REQUEST_DENIED" in caplog.text and "provided API key is invalid" in caplog.text
    assert "server-key" not in caplog.text


def test_widget_config_returns_tenant_branding_for_a_valid_widget_key(client, monkeypatch):
    branded = Tenant(id="wco", pricing=None, allowed_origins=frozenset({"https://wco.com"}),
                     branding={"name": "WCo Lawns", "accent": "#123456", "cta_url": "https://wco.com"})
    monkeypatch.setattr(api, "WIDGET_KEYS", {"pub-k": "wco"})
    monkeypatch.setattr(api, "TENANTS", {"wco": branded})
    ok = client.get("/widget/config", headers={"X-API-Key": "pub-k", "Origin": "https://wco.com"})
    assert ok.json()["branding"]["name"] == "WCo Lawns"
    bad_origin = client.get("/widget/config", headers={"X-API-Key": "pub-k", "Origin": "https://evil.com"})
    assert bad_origin.json()["branding"] is None    # domain-locked
    assert client.get("/widget/config").json()["branding"] is None   # no key


def test_captcha_enforced_for_widget_key_when_secret_set(client, monkeypatch, mock_run):
    _register_widget(monkeypatch)
    monkeypatch.setattr(api, "TURNSTILE_SECRET_KEY", "sekret")
    monkeypatch.setattr(api, "verify_turnstile", lambda token, ip=None: token == "good")
    hdr = {"X-API-Key": "pub-k", "Origin": "https://wco.com"}
    assert client.post("/quote", json={"address": ADDRESS}, headers=hdr).status_code == 403  # no token
    assert client.post("/quote", json={"address": ADDRESS},
                       headers={**hdr, "CF-Turnstile-Response": "nope"}).status_code == 403  # bad token
    r = client.post("/quote", json={"address": ADDRESS},
                    headers={**hdr, "CF-Turnstile-Response": "good"})                          # valid token
    assert r.status_code == 200


def test_captcha_skipped_for_secret_key(client, monkeypatch, mock_run):
    _register_widget(monkeypatch)
    monkeypatch.setattr(api, "TURNSTILE_SECRET_KEY", "sekret")
    r = client.post("/quote", json={"address": ADDRESS}, headers={"X-API-Key": "secret-k"})   # operator skips CAPTCHA
    assert r.status_code == 200


# --- Lead capture (W3) ------------------------------------------------------

def test_lead_captured_with_contact(client):
    r = client.post("/lead", headers=KEY, json={
        "address": "5534 Mayberry St, Omaha, NE 68106",
        "name": "Sam", "email": "sam@example.com", "phone": "402-555-1212",
    })
    assert r.status_code == 200
    assert r.json()["captured"] is True   # notified depends on an ESP being configured


def test_lead_requires_email_or_phone(client):
    r = client.post("/lead", headers=KEY, json={"address": "5534 Mayberry St"})
    assert r.status_code == 422


def test_lead_missing_key_is_rejected(client):
    r = client.post("/lead", json={"address": "x", "email": "a@b.com"})
    assert r.status_code == 422  # required X-API-Key header absent


def test_lead_publishable_key_is_domain_locked(client, monkeypatch):
    _register_widget(monkeypatch)
    ok = client.post("/lead", headers={"X-API-Key": "pub-k", "Origin": "https://wco.com"},
                     json={"address": "x", "email": "a@b.com"})
    assert ok.status_code == 200
    bad = client.post("/lead", headers={"X-API-Key": "pub-k", "Origin": "https://evil.com"},
                      json={"address": "x", "email": "a@b.com"})
    assert bad.status_code == 403


def test_lead_endpoint_persists_a_durable_row(client, monkeypatch):
    _register_widget(monkeypatch)
    monkeypatch.setattr(api, "LEADS", LeadStore(connect(":memory:")))
    r = client.post("/lead", headers={"X-API-Key": "pub-k", "Origin": "https://wco.com"},
                    json={"address": "5534 Mayberry St", "name": "Sam",
                          "email": "sam@example.com", "phone": "402-555-1212"})
    assert r.status_code == 200
    leads = api.LEADS.recent("wco")
    assert len(leads) == 1
    assert leads[0]["source"] == "widget"
    assert leads[0]["email"] == "sam@example.com"
    assert leads[0]["name"] == "Sam"
    assert leads[0]["lawn_sqft"] is None   # captured before measurement runs


def test_quote_with_email_persists_a_lead_with_the_measurement(client, mock_run, monkeypatch):
    _register_widget(monkeypatch)
    monkeypatch.setattr(api, "LEADS", LeadStore(connect(":memory:")))
    monkeypatch.setattr(api, "send_quote_emails", lambda *a, **k: True)
    r = client.post("/quote", headers={"X-API-Key": "pub-k", "Origin": "https://wco.com"},
                    json={"address": ADDRESS, "email": "cust@example.com", "name": "Sam"})
    assert r.status_code == 200
    leads = api.LEADS.recent("wco")
    assert len(leads) == 1
    assert leads[0]["source"] == "quote"
    assert leads[0]["email"] == "cust@example.com"
    assert leads[0]["lawn_sqft"] == 4669.8   # measurement filled in



def test_out_of_coverage_message_passes_through(client, monkeypatch):
    # The already-clean out-of-Coverage message is kept verbatim.
    def _raise(address):
        raise ValueError("'X' is outside our service area. We currently cover Douglas County and Sarpy County only.")

    monkeypatch.setattr(api, "resolve", _raise)
    r = client.post("/quote", json={"address": "1600 Pennsylvania Ave NW, Washington, DC 20500"}, headers=KEY)
    assert r.status_code == 422
    assert "outside our service area" in r.json()["detail"]


def test_quote_pipeline_error_maps_to_500_without_leaking_internals(client, monkeypatch):
    # A non-ValueError failure returns a generic message — internals (which can
    # embed request URLs carrying the Google key) must never reach the client.
    def _boom(address, imagery="auto", **kwargs):
        raise RuntimeError("parcel lookup failed at http://internal/secret?key=AIzaLEAK")

    monkeypatch.setattr(api, "run", _boom)
    r = client.post("/quote", json={"address": "bad"}, headers=KEY)
    assert r.status_code == 500
    detail = r.json()["detail"]
    assert detail == api.GENERIC_PIPELINE_ERROR
    assert "secret" not in detail and "AIzaLEAK" not in detail and "parcel lookup" not in detail


def test_quote_value_error_maps_to_422_with_message(client, monkeypatch):
    # Curated, address-facing failures keep their (safe) message and are the
    # caller's fault to fix → 422, not 500.
    def _reject(address, imagery="auto", **kwargs):
        raise ValueError("'X, Faketown' is outside our service area.")

    monkeypatch.setattr(api, "run", _reject)
    r = client.post("/quote", json={"address": "X, Faketown"}, headers=KEY)
    assert r.status_code == 422
    assert "outside our service area" in r.json()["detail"]


def test_quote_email_is_throttled_after_limit(client, mock_run, monkeypatch):
    monkeypatch.setattr(api, "send_quote_emails", lambda *a, **k: True)
    body = {"address": ADDRESS, "email": "spammy@example.com"}
    for _ in range(api._EMAIL_MAX_PER_HOUR):
        assert client.post("/quote", json=body, headers=KEY).json()["email_sent"] is True
    # Over the limit: still a full 200 quote, but no send.
    r = client.post("/quote", json=body, headers=KEY)
    assert r.status_code == 200
    assert r.json()["email_sent"] is False
    assert r.json()["lawn_sqft"] == 4669.8


# ---------------------------------------------------------------------------
# Customer email / lead capture on /quote
# ---------------------------------------------------------------------------
ADDRESS = "5534 Mayberry St, Omaha, NE 68106"


def test_quote_with_email_sends_and_still_returns_full_quote(client, mock_run, monkeypatch):
    calls = {}

    def fake_send(payload, email, name=None, company=None, **kwargs):
        calls.update(payload=payload, email=email, name=name, company=company, **kwargs)
        return True

    monkeypatch.setattr(api, "send_quote_emails", fake_send)
    r = client.post("/quote", json={"address": ADDRESS, "email": "cust@example.com", "name": "Sam"},
                    headers=KEY)
    assert r.status_code == 200
    body = r.json()
    # the website still gets everything it displays
    assert body["lawn_sqft"] == 4669.8
    assert body["pricing"] is not None
    assert body["email_sent"] is True
    # and the emailer got the same quote payload + the customer's contact
    assert calls["email"] == "cust@example.com"
    assert calls["name"] == "Sam"
    assert calls["company"] == "Partner Lawn Care"
    assert calls["payload"]["lawn_sqft"] == 4669.8


def test_quote_email_failure_never_fails_the_quote(client, mock_run, monkeypatch):
    monkeypatch.setattr(api, "send_quote_emails", lambda *a, **k: False)
    r = client.post("/quote", json={"address": ADDRESS, "email": "cust@example.com"}, headers=KEY)
    assert r.status_code == 200
    assert r.json()["email_sent"] is False
    assert r.json()["lawn_sqft"] == 4669.8


def test_quote_email_crash_never_fails_the_quote(client, mock_run, monkeypatch):
    def _boom(*a, **k):
        raise RuntimeError("renderer bug")

    monkeypatch.setattr(api, "send_quote_emails", _boom)
    r = client.post("/quote", json={"address": ADDRESS, "email": "cust@example.com"}, headers=KEY)
    assert r.status_code == 200
    assert r.json()["email_sent"] is False
    assert r.json()["lawn_sqft"] == 4669.8


def test_quote_without_email_never_touches_the_emailer(client, mock_run, monkeypatch):
    def _fail(*a, **k):
        raise AssertionError("emailer must not be called without an email")

    monkeypatch.setattr(api, "send_quote_emails", _fail)
    r = client.post("/quote", json={"address": ADDRESS}, headers=KEY)
    assert r.status_code == 200
    assert r.json()["email_sent"] is None


def test_quote_invalid_email_is_rejected_before_the_pipeline(client, monkeypatch):
    def _boom(address, imagery="auto", **kwargs):
        raise AssertionError("pipeline must not run for an invalid email")

    monkeypatch.setattr(api, "run", _boom)
    r = client.post("/quote", json={"address": ADDRESS, "email": "not-an-email"}, headers=KEY)
    assert r.status_code == 422


# ---------------------------------------------------------------------------
# /quote/batch job flow
# ---------------------------------------------------------------------------
def test_batch_rejects_empty_address_list(client):
    r = client.post("/quote/batch", json={"addresses": []}, headers=KEY)
    assert r.status_code == 422


def test_batch_submits_job_and_completes(client, mock_run):
    r = client.post("/quote/batch",
                    json={"addresses": ["5534 Mayberry St, Omaha, NE 68106",
                                        "5819 Hickory St, Omaha, NE 68106"]},
                    headers=KEY)
    assert r.status_code == 200
    job_id = r.json()["job_id"]
    assert r.json()["total"] == 2

    status = {}
    deadline = time.time() + 5
    while time.time() < deadline:
        status = client.get(f"/jobs/{job_id}", headers=KEY).json()
        if status["status"] == "done":
            break
        time.sleep(0.05)

    assert status["status"] == "done"
    assert status["completed"] == 2
    assert all(res["status"] == "ok" for res in status["results"])
    # batch results follow the same contract: viz via result_id only
    assert all(res["result_id"] and "visualization_url" not in res for res in status["results"])


# ---------------------------------------------------------------------------
# Backpressure (M-1): a single quote can't wait on the lock forever
# ---------------------------------------------------------------------------
def test_quote_returns_503_when_pipeline_is_busy(client, mock_run, monkeypatch):
    monkeypatch.setattr(api, "QUOTE_LOCK_TIMEOUT_S", 0.05)
    api._PIPELINE_LOCK.acquire()  # simulate a batch holding the pipeline
    try:
        r = client.post("/quote", json={"address": ADDRESS}, headers=KEY)
    finally:
        api._PIPELINE_LOCK.release()
    assert r.status_code == 503
    assert r.headers.get("Retry-After") == "60"


# ---------------------------------------------------------------------------
# Per-client ownership (M-3): jobs and visualizations aren't cross-readable
# ---------------------------------------------------------------------------
def test_job_not_readable_by_another_client(client, mock_run, monkeypatch):
    monkeypatch.setattr(api, "ALLOWED_API_KEYS", {"akey": "alpha", "bkey": "beta"})
    job_id = client.post("/quote/batch", json={"addresses": ["5534 Mayberry St, Omaha, NE 68106"]},
                         headers={"X-API-Key": "akey"}).json()["job_id"]
    # beta cannot see alpha's job (404, not 403 — existence must not leak)
    assert client.get(f"/jobs/{job_id}", headers={"X-API-Key": "bkey"}).status_code == 404
    # alpha can
    assert client.get(f"/jobs/{job_id}", headers={"X-API-Key": "akey"}).status_code == 200


def test_viz_requires_a_key(client):
    assert client.get("/viz/anything.png").status_code == 422  # X-API-Key header absent


def test_viz_unknown_id_is_404(client):
    assert client.get("/viz/nope123.png", headers=KEY).status_code == 404


def test_viz_served_to_owner_only(client, tmp_path, monkeypatch):
    monkeypatch.setattr(api, "ALLOWED_API_KEYS", {"akey": "alpha", "bkey": "beta"})
    png = tmp_path / "v.png"
    png.write_bytes(b"\x89PNG\r\n\x1a\n")
    overlay = tmp_path / "o.png"
    overlay.write_bytes(b"\x89PNG\r\n\x1a\n")
    monkeypatch.setitem(api.VIZ_PATHS, "rid123", (png, overlay, "alpha"))
    assert client.get("/viz/rid123.png", headers={"X-API-Key": "akey"}).status_code == 200
    assert client.get("/viz/rid123.png", headers={"X-API-Key": "bkey"}).status_code == 404


def test_overlay_served_to_owner_only(client, tmp_path, monkeypatch):
    # The homeowner overlay is widget-callable (secret OR domain-locked publishable key),
    # but still scoped to the owning client.
    monkeypatch.setattr(api, "ALLOWED_API_KEYS", {"akey": "alpha", "bkey": "beta"})
    png, overlay = tmp_path / "v.png", tmp_path / "o.png"
    png.write_bytes(b"\x89PNG\r\n\x1a\n")
    overlay.write_bytes(b"\x89PNG\r\n\x1a\n")
    monkeypatch.setitem(api.VIZ_PATHS, "rid123", (png, overlay, "alpha"))
    assert client.get("/overlay/rid123.png", headers={"X-API-Key": "akey"}).status_code == 200
    assert client.get("/overlay/rid123.png", headers={"X-API-Key": "bkey"}).status_code == 404


def test_overlay_404_when_absent(client, tmp_path, monkeypatch):
    monkeypatch.setattr(api, "ALLOWED_API_KEYS", {"akey": "alpha"})
    png = tmp_path / "v.png"
    png.write_bytes(b"\x89PNG\r\n\x1a\n")
    monkeypatch.setitem(api.VIZ_PATHS, "rid456", (png, None, "alpha"))  # viz but no overlay
    assert client.get("/overlay/rid456.png", headers={"X-API-Key": "akey"}).status_code == 404


# ---------------------------------------------------------------------------
# Job/viz lifecycle (M-9 + M-4): evict only finished jobs; clean up their viz
# ---------------------------------------------------------------------------
def test_evict_only_removes_done_jobs(monkeypatch):
    from lawn_estimator.db import JobStore
    store = JobStore(connect(":memory:"))
    monkeypatch.setattr(api, "JOBS_STORE", store)
    monkeypatch.setattr(api, "MAX_JOBS_KEPT", 1)
    # Explicit created_at so "oldest done first" is deterministic (no timestamp ties).
    for jid, status, ts in [("d1", "done", "2020-01-01"), ("q1", "queued", "2020-01-02"),
                            ("d2", "done", "2020-01-03")]:
        store._db.execute(
            "INSERT INTO jobs (id, client, imagery, addresses, status, total, completed, created_at) "
            "VALUES (?, 'c', 'auto', '[]', ?, 0, 0, ?)", (jid, status, ts))
    store._db.commit()
    api._evict_old_jobs()
    ids = {r["id"] for r in store._db.execute("SELECT id FROM jobs").fetchall()}
    assert "q1" in ids       # a queued job is never evicted (would 404 the poller)
    assert "d1" not in ids   # finished jobs go, oldest first


def test_evict_removes_viz_entry_and_unlinks_png(monkeypatch, tmp_path):
    from lawn_estimator.db import JobStore
    store = JobStore(connect(":memory:"))
    monkeypatch.setattr(api, "JOBS_STORE", store)
    monkeypatch.setattr(api, "OUTPUT_DIR", tmp_path)
    monkeypatch.setattr(api, "MAX_JOBS_KEPT", 0)
    api.VIZ_PATHS.clear()
    try:
        png = tmp_path / "e.png"
        png.write_bytes(b"x")
        api.VIZ_PATHS["r1"] = (png, None, "c")
        store.create("j1", client="c", imagery="auto", addresses=["a"])
        store.append_result("j1", 1, {"result_id": "r1", "status": "ok"})
        store.mark_done("j1")
        api._evict_old_jobs()
        assert "r1" not in api.VIZ_PATHS
        assert not png.exists()  # PNG unlinked, not orphaned on the ephemeral disk
    finally:
        api.VIZ_PATHS.clear()


# ---------------------------------------------------------------------------
# Clerk auth: /auth/config + DB-backed team invitations (Option B)
# ---------------------------------------------------------------------------
def test_pending_invite_activates_on_first_signin(client, monkeypatch):
    db = connect(":memory:")
    monkeypatch.setattr(api, "USERS", UserStore(db))
    monkeypatch.setattr(api, "AUDIT", AuditLog(db))
    monkeypatch.setattr(api, "CLERK_SECRET_KEY", "sk_test")
    api.USERS.invite("thelawnstandard", "op@x.com", "owner")   # a pending invite exists
    monkeypatch.setattr(api, "verify_clerk_session", lambda t: "clerk_op1")
    monkeypatch.setattr(api, "verify_clerk_email", lambda t: "op@x.com")
    body = client.get("/dashboard/me", headers={"Authorization": "Bearer x"}).json()
    assert body["tenant_id"] == "thelawnstandard" and body["role"] == "owner" and body["email"] == "op@x.com"
    assert api.USERS.by_clerk_id("clerk_op1")["status"] == "active"   # bound on first sign-in


def test_signin_without_an_invite_is_denied(client, monkeypatch):
    db = connect(":memory:")
    monkeypatch.setattr(api, "USERS", UserStore(db))
    monkeypatch.setattr(api, "CLERK_SECRET_KEY", "sk_test")
    monkeypatch.setattr(api, "verify_clerk_session", lambda t: "clerk_rando")
    monkeypatch.setattr(api, "verify_clerk_email", lambda t: "stranger@nope.com")
    r = client.get("/dashboard/me", headers={"Authorization": "Bearer x"})
    assert r.status_code == 403                                  # invite-only
    assert api.USERS.by_clerk_id("clerk_rando") is None


def test_auth_config_exposes_publishable_key_only(client, monkeypatch):
    monkeypatch.setattr(api, "CLERK_PUBLISHABLE_KEY",
                        "pk_test_c3VubnktZG9ua2V5LTU1LmNsZXJrLmFjY291bnRzLmRldiQ")
    body = client.get("/auth/config").json()
    assert body["clerk_publishable_key"].startswith("pk_test_")
    assert body["clerk_frontend_api"] == "sunny-donkey-55.clerk.accounts.dev"
    assert "secret" not in str(body).lower()          # no secret ever leaks here
    monkeypatch.setattr(api, "CLERK_PUBLISHABLE_KEY", None)
    assert client.get("/auth/config").json()["clerk_publishable_key"] is None   # -> dev-token mode


def _owner_session(monkeypatch, db, tenant="thelawnstandard", email="owner@x.com", clerk="clerk_owner"):
    monkeypatch.setattr(api, "USERS", UserStore(db))
    monkeypatch.setattr(api, "AUDIT", AuditLog(db))
    monkeypatch.setattr(api, "CLERK_SECRET_KEY", "sk_test")
    monkeypatch.setattr(api, "CLERK", None)   # skip the best-effort Clerk invite call in tests
    uid = api.USERS.create(tenant, clerk, email, "owner", status="active")
    monkeypatch.setattr(api, "verify_clerk_session", lambda t: clerk)
    monkeypatch.setattr(api, "verify_clerk_email", lambda t: email)
    return uid


def test_owner_invites_teammate_dedupes_and_caps(client, monkeypatch):
    db = connect(":memory:")
    _owner_session(monkeypatch, db)
    hdr = {"Authorization": "Bearer x"}
    assert client.post("/dashboard/members/invite", headers=hdr,
                       json={"email": "staff@x.com", "role": "staff"}).json()["invited"] == "staff@x.com"
    members = client.get("/dashboard/members", headers=hdr).json()["members"]
    assert any(m["email"] == "staff@x.com" and m["status"] == "invited" for m in members)
    assert client.post("/dashboard/members/invite", headers=hdr,
                       json={"email": "staff@x.com"}).status_code == 409          # dedupe
    for i in range(3):   # owner + staff = 2 seats; fill to 5, then reject the 6th
        assert client.post("/dashboard/members/invite", headers=hdr,
                           json={"email": f"m{i}@x.com"}).status_code == 200
    assert client.post("/dashboard/members/invite", headers=hdr,
                       json={"email": "over@x.com"}).status_code == 422           # cap of 5


def test_staff_cannot_invite(client, monkeypatch):
    db = connect(":memory:")
    monkeypatch.setattr(api, "USERS", UserStore(db))
    monkeypatch.setattr(api, "CLERK_SECRET_KEY", "sk_test")
    api.USERS.create("acme", "clerk_staff", "staff@x.com", "staff", status="active")
    monkeypatch.setattr(api, "verify_clerk_session", lambda t: "clerk_staff")
    monkeypatch.setattr(api, "verify_clerk_email", lambda t: "staff@x.com")
    assert client.post("/dashboard/members/invite", headers={"Authorization": "Bearer x"},
                       json={"email": "x@y.com"}).status_code == 403   # require_owner


def test_cannot_remove_or_demote_last_owner(client, monkeypatch):
    db = connect(":memory:")
    uid = _owner_session(monkeypatch, db)
    hdr = {"Authorization": "Bearer x"}
    assert client.post(f"/dashboard/members/{uid}/remove", headers=hdr).status_code == 422   # self
    assert client.post(f"/dashboard/members/{uid}/role", headers=hdr,
                       json={"role": "staff"}).status_code == 422                             # last owner


def test_admin_invite_owner_bootstraps_a_tenant(client, monkeypatch):
    db = connect(":memory:")
    monkeypatch.setattr(api, "USERS", UserStore(db))
    monkeypatch.setattr(api, "AUDIT", AuditLog(db))
    monkeypatch.setattr(api, "CLERK", None)
    monkeypatch.setattr(api, "CLERK_SECRET_KEY", None)          # dev-token admin path
    monkeypatch.setattr(api, "ADMIN_DEV_TOKEN", "admintok")
    monkeypatch.setattr(api, "PLATFORM_ADMIN_EMAILS", {"me@x.com"})
    r = client.post("/admin/invite-owner", headers={"Authorization": "Bearer admintok"},
                    json={"tenant_id": "newco", "email": "owner@newco.com"})
    assert r.status_code == 200 and r.json()["tenant"] == "newco"
    assert api.USERS.pending_by_email("owner@newco.com")["role"] == "owner"


# ---------------------------------------------------------------------------
# W5 — in-widget booking (/book)
# ---------------------------------------------------------------------------
def test_book_records_booking_lead_and_notifies(client, monkeypatch):
    _register_widget(monkeypatch)
    monkeypatch.setattr(api, "LEADS", LeadStore(connect(":memory:")))
    sent = {}
    monkeypatch.setattr(api, "notify_booking", lambda **kw: (sent.update(kw), True)[1])
    wco = {"X-API-Key": "pub-k", "Origin": "https://wco.com"}

    ok = client.post("/book", headers=wco, json={
        "address": "1 A St", "email": "a@x.com", "result_id": "r1",
        "services": ["Mowing", "Fertilizer"], "total": 105.0})
    assert ok.status_code == 200 and ok.json()["booked"] is True
    row = api.LEADS.recent("wco")[0]
    assert row["source"] == "booking" and row["price_total"] == 105.0 and row["result_id"] == "r1"
    assert sent["services"] == ["Mowing", "Fertilizer"] and sent["total"] == 105.0

    # Validation: no contact, or no service selected -> 422.
    assert client.post("/book", headers=wco,
                       json={"address": "x", "services": ["Mowing"]}).status_code == 422
    assert client.post("/book", headers=wco,
                       json={"address": "x", "email": "a@x.com", "services": []}).status_code == 422
    # Domain-locked like /quote and /lead.
    assert client.post("/book", headers={"X-API-Key": "pub-k", "Origin": "https://evil.com"},
                       json={"address": "x", "email": "a@x.com", "services": ["Mowing"]}).status_code == 403


def test_book_captures_bundle(client, monkeypatch):
    _register_widget(monkeypatch)
    monkeypatch.setattr(api, "LEADS", LeadStore(connect(":memory:")))
    monkeypatch.setattr(api, "notify_booking", lambda **kw: True)
    wco = {"X-API-Key": "pub-k", "Origin": "https://wco.com"}
    r = client.post("/book", headers=wco, json={
        "address": "1 A St", "email": "a@x.com", "result_id": "r1", "total": 120.0,
        "services": ["Mowing", "Aeration"], "bundle_id": "spring", "bundle_name": "Spring Package",
        "cadence": "seasonal"})
    assert r.status_code == 200
    import json as _json
    detail = _json.loads(api.LEADS.recent("wco")[0]["detail"])
    assert detail["bundle"] == {"id": "spring", "name": "Spring Package", "cadence": "seasonal"}


def test_widget_config_includes_qualifying_questions(client, monkeypatch):
    q = [{"label": "Locked gate?", "type": "boolean", "surcharge": 10}]
    t = Tenant(id="wco", pricing=None, allowed_origins=frozenset({"https://wco.com"}), questions=tuple(q))
    monkeypatch.setattr(api, "WIDGET_KEYS", {"pub-k": "wco"})
    monkeypatch.setattr(api, "TENANTS", {"wco": t})
    body = client.get("/widget/config", headers={"X-API-Key": "pub-k", "Origin": "https://wco.com"}).json()
    assert body["questions"] == q


def test_book_captures_answers(client, monkeypatch):
    _register_widget(monkeypatch)
    monkeypatch.setattr(api, "LEADS", LeadStore(connect(":memory:")))
    sent = {}
    monkeypatch.setattr(api, "notify_booking", lambda **kw: (sent.update(kw), True)[1])
    r = client.post("/book", headers={"X-API-Key": "pub-k", "Origin": "https://wco.com"}, json={
        "address": "1 A St", "email": "a@x.com", "services": ["Mowing"], "total": 55.0,
        "answers": {"Locked gate?": "Yes", "Frequency": "One-time"}})
    assert r.status_code == 200
    assert sent["answers"] == {"Locked gate?": "Yes", "Frequency": "One-time"}


def test_priority_lock_interactive_quote_jumps_the_batch_queue():
    # A2: while an interactive (high) quote is waiting, a batch (low) address must not grab
    # the pipeline — the homeowner never waits behind more than the one in-flight quote.
    import threading

    lock = api.PriorityLock()
    order = []

    assert lock.acquire(priority="low")   # a batch address is mid-run, holding the pipeline

    def interactive():
        assert lock.acquire(priority="high")
        order.append("interactive")
        lock.release()

    def next_batch_address():
        assert lock.acquire(priority="low")
        order.append("batch")
        lock.release()

    hi = threading.Thread(target=interactive)
    lo = threading.Thread(target=next_batch_address)
    hi.start()
    while lock._waiting_hi == 0:       # ensure the interactive quote is queued first
        time.sleep(0.001)
    lo.start()
    time.sleep(0.02)                  # let the next batch address reach its wait
    lock.release()                    # the in-flight batch address finishes
    hi.join(2)
    lo.join(2)
    assert order == ["interactive", "batch"]   # interactive jumped ahead of the queued batch


def test_priority_lock_times_out_to_503_signal():
    # A single /quote caps its wait: if the pipeline stays busy, acquire returns False
    # (the caller raises PipelineBusy -> 503) rather than blocking forever.
    lock = api.PriorityLock()
    assert lock.acquire(priority="low")             # something else holds it
    assert lock.acquire(timeout=0.05, priority="high") is False
    lock.release()
    assert lock.acquire(timeout=0.05, priority="high") is True   # free now


# ---------------------------------------------------------------------------
# B3 — Cloud Tasks async batch (routing + the /tasks/quote handler)
# ---------------------------------------------------------------------------
class _FakeDispatcher:
    """Stand-in for tasks.TASKS. `enabled` toggles the batch routing; enqueues are recorded."""

    def __init__(self, enabled):
        self.enabled = enabled
        self.enqueued = []

    def enqueue_address(self, *, job_id, idx, address, imagery, client, submitted_by=""):
        self.enqueued.append({"job_id": job_id, "idx": idx, "address": address,
                              "imagery": imagery, "client": client, "submitted_by": submitted_by})

    def verify_request(self, token):
        return token == "tasktoken"


def test_batch_enqueues_cloud_tasks_when_enabled(client, mock_run, monkeypatch):
    # With Cloud Tasks on, /quote/batch enqueues one task per address and does NOT touch the
    # in-process queue (which wouldn't survive on serverless).
    fake = _FakeDispatcher(enabled=True)
    monkeypatch.setattr(api, "TASKS", fake)
    puts = []
    monkeypatch.setattr(api._JOB_QUEUE, "put", lambda x: puts.append(x))
    r = client.post("/quote/batch", json={"addresses": ["1 A St", "2 B St", "3 C St"]}, headers=KEY)
    assert r.status_code == 200
    jid = r.json()["job_id"]
    assert [e["idx"] for e in fake.enqueued] == [1, 2, 3]
    assert all(e["job_id"] == jid for e in fake.enqueued)
    assert puts == []   # the in-process queue is bypassed under Cloud Tasks


def test_dispatch_job_falls_back_to_in_process_when_disabled(monkeypatch):
    fake = _FakeDispatcher(enabled=False)
    monkeypatch.setattr(api, "TASKS", fake)
    puts = []
    monkeypatch.setattr(api._JOB_QUEUE, "put", lambda x: puts.append(x))
    api._dispatch_job("jX", imagery="auto", client="c", addresses=["a", "b"])
    assert puts == ["jX"] and fake.enqueued == []   # in-process path, no tasks


def test_tasks_quote_rejects_missing_or_bad_token(client, monkeypatch):
    monkeypatch.setattr(api, "TASKS", _FakeDispatcher(enabled=True))
    body = {"job_id": "x", "idx": 1, "address": "1 A St"}
    assert client.post("/tasks/quote", json=body).status_code == 403                       # no header
    assert client.post("/tasks/quote", json=body,
                       headers={"X-Tasks-Token": "wrong"}).status_code == 403               # bad token


def test_tasks_quote_runs_addresses_and_marks_done_on_last(client, mock_run, monkeypatch):
    monkeypatch.setattr(api, "TASKS", _FakeDispatcher(enabled=True))
    jid = "b3job"
    api.JOBS_STORE.create(jid, client="tenantX", imagery="auto", addresses=["1 A St", "2 B St"])
    hdr = {"X-Tasks-Token": "tasktoken"}

    r1 = client.post("/tasks/quote",
                     json={"job_id": jid, "idx": 1, "address": "1 A St", "client": "tenantX"}, headers=hdr)
    assert r1.status_code == 200
    mid = api.JOBS_STORE.get(jid, "tenantX")
    assert mid["completed"] == 1 and mid["status"] != "done"   # not done until the last address

    r2 = client.post("/tasks/quote",
                     json={"job_id": jid, "idx": 2, "address": "2 B St", "client": "tenantX"}, headers=hdr)
    assert r2.status_code == 200
    fin = api.JOBS_STORE.get(jid, "tenantX")
    assert fin["completed"] == 2 and fin["status"] == "done"
    assert [res["status"] for res in fin["results"]] == ["ok", "ok"]


def test_tasks_quote_is_idempotent_on_retry(client, mock_run, monkeypatch):
    # Cloud Tasks may redeliver a task; re-running the same idx overwrites, never duplicates.
    monkeypatch.setattr(api, "TASKS", _FakeDispatcher(enabled=True))
    jid = "b3retry"
    api.JOBS_STORE.create(jid, client="t", imagery="auto", addresses=["1 A St"])
    hdr = {"X-Tasks-Token": "tasktoken"}
    body = {"job_id": jid, "idx": 1, "address": "1 A St", "client": "t"}
    assert client.post("/tasks/quote", json=body, headers=hdr).status_code == 200
    assert client.post("/tasks/quote", json=body, headers=hdr).status_code == 200   # redelivery
    fin = api.JOBS_STORE.get(jid, "t")
    assert fin["completed"] == 1 and len(fin["results"]) == 1 and fin["status"] == "done"


def test_record_batch_lead_files_and_dedups(monkeypatch):
    from lawn_estimator.db import LeadStore, connect
    monkeypatch.setattr(api, "LEADS", LeadStore(connect(":memory:")))
    result = {
        "status": "ok", "result_id": "r1", "formatted_address": "1 A St", "zip_code": "68106",
        "lawn_sqft": 4200.0, "confidence": "high",
        "pricing": {"line_items": [{"service": "mow", "price": 48.0}, {"service": "fert", "price": 32.0}]},
        "lawn_polygon_px": [[0, 0], [10, 0], [10, 10]], "pixel_area_sqft": 0.5, "image_size_px": [1280, 1280],
    }
    api._record_batch_lead("acme", "runner@acme.com", result)
    leads = api.LEADS.recent("acme")
    assert len(leads) == 1
    lead = leads[0]
    assert lead["source"] == "batch" and lead["email"] == "runner@acme.com" and lead["result_id"] == "r1"
    assert lead["price_total"] == 80.0                       # summed from the line items
    import json as _json
    detail = _json.loads(lead["detail"])
    assert detail["pixel_area_sqft"] == 0.5 and detail["image_size_px"] == [1280, 1280]
    assert detail["pricing"]["line_items"] and detail["lawn_polygon_px"]   # geometry for the popup

    api._record_batch_lead("acme", "runner@acme.com", result)   # same result_id -> deduped
    assert len(api.LEADS.recent("acme")) == 1
    api._record_batch_lead("acme", "x", {"status": "error", "address": "2 B St"})   # non-ok skipped
    assert len(api.LEADS.recent("acme")) == 1


def test_measurement_edit_persists_to_lead_and_keeps_history(client, mock_run, monkeypatch):
    import json as _json

    from lawn_estimator.db import AuditLog, LeadStore, MeasurementEditStore, RunLedger, UserStore, connect
    db = connect(":memory:")
    monkeypatch.setattr(api, "USERS", UserStore(db))
    monkeypatch.setattr(api, "METER", RunLedger(db))
    monkeypatch.setattr(api, "MEDITS", MeasurementEditStore(db))
    monkeypatch.setattr(api, "LEADS", LeadStore(db))
    monkeypatch.setattr(api, "AUDIT", AuditLog(db))
    api.USERS.create("acme", "clerk_acme", "a@acme.com", "owner", status="active")
    api.METER.record("acme", "1 Batch St", api.BATCH, lawn_sqft=4000.0, result_id="rb", zip="68106")
    api.LEADS.record("acme", "batch", "1 Batch St", email="runner@acme.com", lawn_sqft=4000.0,
                     result_id="rb", detail=_json.dumps({"original_sqft": 4000.0}))
    monkeypatch.setattr(api, "verify_clerk_session", lambda t: "clerk_acme")
    hdr = {"Authorization": "Bearer x"}

    # An edit persists onto the lead (no lost work) and is recorded as a tenant edit.
    assert client.post("/dashboard/measurement/edit", headers=hdr,
                       json={"result_id": "rb", "edited_sqft": 4500}).status_code == 200
    assert api.LEADS.get_by_result("acme", "rb")["lawn_sqft"] == 4500.0

    h = client.get("/dashboard/measurement/history/rb", headers=hdr).json()
    assert [s["kind"] for s in h["states"]] == ["original", "tenant"]
    assert h["states"][0]["sqft"] == 4000.0 and h["current_sqft"] == 4500.0

    # Revert = re-apply the original; it persists and the history keeps every state.
    assert client.post("/dashboard/measurement/edit", headers=hdr,
                       json={"result_id": "rb", "edited_sqft": 4000}).status_code == 200
    assert api.LEADS.get_by_result("acme", "rb")["lawn_sqft"] == 4000.0
    h2 = client.get("/dashboard/measurement/history/rb", headers=hdr).json()
    assert h2["current_sqft"] == 4000.0 and len(h2["states"]) == 3   # original + two edits


def test_book_records_customer_measurement_edit(client, monkeypatch):
    from lawn_estimator.db import LeadStore, MeasurementEditStore, RunLedger, connect
    _register_widget(monkeypatch)
    db = connect(":memory:")
    monkeypatch.setattr(api, "LEADS", LeadStore(db))
    monkeypatch.setattr(api, "METER", RunLedger(db))
    monkeypatch.setattr(api, "MEDITS", MeasurementEditStore(db))
    monkeypatch.setattr(api, "notify_booking", lambda **kw: True)
    api.METER.record("wco", "1 A St", api.SINGLE, lawn_sqft=4000.0, result_id="r1", zip="68106")
    wco = {"X-API-Key": "pub-k", "Origin": "https://wco.com"}

    # Booking with an adjusted outline records a 'customer' edit + persists the booked area.
    r = client.post("/book", headers=wco, json={"address": "1 A St", "email": "a@x.com",
        "result_id": "r1", "services": ["Mowing"], "total": 90.0, "edited_sqft": 4600})
    assert r.status_code == 200
    edits = api.MEDITS.history("wco", "r1")
    assert len(edits) == 1 and edits[0]["kind"] == "customer"
    assert edits[0]["edited_sqft"] == 4600.0 and edits[0]["original_sqft"] == 4000.0
    lead = api.LEADS.get_by_result("wco", "r1")
    assert lead["source"] == "booking" and lead["lawn_sqft"] == 4600.0

    # Booking without a real change (edited == original) records no edit.
    api.METER.record("wco", "2 B St", api.SINGLE, lawn_sqft=3000.0, result_id="r2", zip="68106")
    client.post("/book", headers=wco, json={"address": "2 B St", "email": "b@x.com",
        "result_id": "r2", "services": ["Mowing"], "total": 70.0, "edited_sqft": 3000})
    assert api.MEDITS.history("wco", "r2") == []


def test_book_stores_line_items_long_format_with_purchased_flag(client, monkeypatch):
    import json as _json

    from lawn_estimator.db import BookingItemStore, LeadStore, connect
    _register_widget(monkeypatch)
    db = connect(":memory:")
    monkeypatch.setattr(api, "LEADS", LeadStore(db))
    monkeypatch.setattr(api, "BOOKING_ITEMS", BookingItemStore(db))
    monkeypatch.setattr(api, "notify_booking", lambda **kw: True)
    wco = {"X-API-Key": "pub-k", "Origin": "https://wco.com"}
    items = [{"service": "mow", "label": "Mowing", "price": 48.0, "purchased": True},
             {"service": "fert", "label": "Fertilizer", "price": 32.0, "purchased": False}]
    r = client.post("/book", headers=wco, json={"address": "1 A St", "email": "a@x.com",
        "result_id": "r1", "services": ["Mowing"], "total": 48.0, "line_items": items})
    assert r.status_code == 200

    stored = api.BOOKING_ITEMS.for_result("wco", "r1")   # one row per offered item, long format
    assert len(stored) == 2
    by = {s["service"]: s for s in stored}
    assert by["mow"]["purchased"] == 1 and by["mow"]["price"] == 48.0 and by["mow"]["label"] == "Mowing"
    assert by["fert"]["purchased"] == 0                  # offered but not bought

    lead = api.LEADS.get_by_result("wco", "r1")          # popup breakdown = the purchased items
    detail = _json.loads(lead["detail"])
    assert [li["service"] for li in detail["pricing"]["line_items"]] == ["mow"]


def test_dashboard_overview(client, mock_run, monkeypatch):
    from lawn_estimator.db import LeadStore, RunLedger, UserStore, connect
    db = connect(":memory:")
    monkeypatch.setattr(api, "USERS", UserStore(db))
    monkeypatch.setattr(api, "METER", RunLedger(db))
    monkeypatch.setattr(api, "LEADS", LeadStore(db))
    api.USERS.create("acme", "clerk_a", "a@acme.com", "owner", status="active")
    api.METER.record("acme", "1 A St", api.BATCH, zip="68106", lawn_sqft=4200.0, duration_s=22.0, result_id="r1")
    api.LEADS.record("acme", "widget", "1 A St", email="a@x")
    api.LEADS.record("acme", "booking", "1 A St", price_total=90.0, result_id="r1")
    monkeypatch.setattr(api, "verify_clerk_session", lambda t: "clerk_a")

    d = client.get("/dashboard/overview", headers={"Authorization": "Bearer x"}).json()
    assert d["tenant"] == "acme" and d["billable_quotes"] == 1 and d["quotes"] == 1
    assert d["by_surface"]["batch"] == 1 and len(d["volume_series"]) == 30
    assert d["leads_summary"]["leads"] == 2 and d["leads_summary"]["bookings"] == 1