"""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