Spaces:
Running
Running
| """Run ledger (db.RunLedger) — the observability columns + real durability (P1a).""" | |
| from __future__ import annotations | |
| from lawn_estimator.db import ( | |
| ApiKeyStore, | |
| AuditLog, | |
| ConfigStore, | |
| JobStore, | |
| LeadStore, | |
| MeasurementEditStore, | |
| RunLedger, | |
| TenantStore, | |
| UserStore, | |
| connect, | |
| ) | |
| from lawn_estimator.metering import SINGLE | |
| def test_record_stores_the_observability_columns(): | |
| conn = connect(":memory:") | |
| RunLedger(conn).record( | |
| "acme", "5534 Mayberry St", SINGLE, | |
| zip="68106", method="lidar+rgb", confidence="high", duration_s=21.4, result_id="abc123", | |
| ) | |
| row = conn.execute("SELECT * FROM runs").fetchone() | |
| assert row["tenant_id"] == "acme" | |
| assert row["surface"] == SINGLE | |
| assert row["zip"] == "68106" | |
| assert row["method"] == "lidar+rgb" | |
| assert row["confidence"] == "high" | |
| assert row["duration_s"] == 21.4 | |
| assert row["result_id"] == "abc123" | |
| assert row["billable"] == 1 | |
| assert row["outcome"] == "ok" # set explicitly (not via the column default) | |
| assert row["created_at"] # server-stamped | |
| def test_record_sets_outcome_ok_without_a_column_default(): | |
| # Reproduce the migrated-Neon condition: `outcome` exists but has NO default (added via | |
| # ADD COLUMN). record() must write 'ok' explicitly, else `ok` stays 0 on prod forever. | |
| db = connect(":memory:") | |
| db.execute("DROP TABLE runs") | |
| db.execute( | |
| "CREATE TABLE runs (id INTEGER PRIMARY KEY AUTOINCREMENT, tenant_id TEXT, surface TEXT, " | |
| "address TEXT, address_norm TEXT, zip TEXT, billable INTEGER, method TEXT, confidence TEXT, " | |
| "lawn_sqft REAL, duration_s REAL, result_id TEXT, outcome TEXT, error_class TEXT, created_at TEXT)" | |
| ) | |
| db.commit() | |
| RunLedger(db).record("acme", "1 A St", "single", result_id="r1") | |
| row = db.execute("SELECT outcome FROM runs WHERE result_id = 'r1'").fetchone() | |
| assert row["outcome"] == "ok" # explicit — there is no default to fall back on | |
| def test_ensure_columns_backfills_legacy_null_outcome_as_ok(): | |
| # A pre-observability row (outcome NULL) counts in total but is invisible to ok — the | |
| # `total=2, ok=0` gap. Re-running the migration backfills it to 'ok'. | |
| db = connect(":memory:") | |
| led = RunLedger(db) | |
| # Recreate runs with a NULLABLE outcome (the migrated Neon column has no NOT NULL / default), | |
| # then insert a pre-observability completed quote as outcome=NULL. | |
| db.execute("DROP TABLE runs") | |
| db.execute( | |
| "CREATE TABLE runs (id INTEGER PRIMARY KEY AUTOINCREMENT, tenant_id TEXT, surface TEXT, " | |
| "address TEXT, address_norm TEXT, zip TEXT, billable INTEGER, method TEXT, confidence TEXT, " | |
| "lawn_sqft REAL, duration_s REAL, result_id TEXT, outcome TEXT, error_class TEXT, " | |
| "created_at TEXT DEFAULT '2020-01-01T00:00:00Z')" | |
| ) | |
| db.execute("INSERT INTO runs (tenant_id, surface, address, address_norm, billable, outcome) " | |
| "VALUES ('acme', 'batch', '1 A St', '1 a st', 1, NULL)") | |
| db.commit() | |
| assert led.stats("acme")["total"] == 1 and led.stats("acme")["ok"] == 0 # the gap | |
| db._ensure_columns() # re-running the migration backfills | |
| s = led.stats("acme") | |
| assert s["total"] == 1 and s["ok"] == 1 # now counted as ok | |
| def test_usage_survives_a_new_connection_to_the_same_file(tmp_path): | |
| db = str(tmp_path / "app.db") | |
| RunLedger(connect(db)).record("acme", "123 A St", SINGLE) | |
| # A brand-new ledger over the same file (i.e. a "redeploy") still sees the count. | |
| assert RunLedger(connect(db)).usage("acme") == 1 | |
| def test_run_ledger_records_failures_and_stats(): | |
| led = RunLedger(connect(":memory:")) | |
| led.record("acme", "1 A St", "single", duration_s=20.0) # ok | |
| led.record("acme", "2 B St", "batch", duration_s=30.0) # ok | |
| led.record_failure("acme", "bad addr", "single", error_class="ValueError", outcome="rejected") | |
| led.record_failure("acme", "3 C St", "batch", error_class="RuntimeError", outcome="error") | |
| s = led.stats("acme") | |
| assert s["total"] == 4 and s["ok"] == 2 and s["rejected"] == 1 and s["errors"] == 1 | |
| assert s["failure_rate"] == 0.25 # errors / total | |
| assert s["avg_duration_s"] == 25.0 # avg over the two ok runs (failures excluded) | |
| assert RunLedger(connect(":memory:")).stats("nobody")["total"] == 0 | |
| def test_stats_by_tenant_is_cross_tenant(): | |
| led = RunLedger(connect(":memory:")) | |
| led.record("acme", "1", "batch", duration_s=20.0) | |
| led.record("acme", "2", "single", duration_s=30.0) | |
| led.record_failure("acme", "x", "single", error_class="ValueError", outcome="rejected") | |
| led.record("beta", "9", "batch", duration_s=10.0) | |
| rows = led.stats_by_tenant() | |
| by = {r["tenant"]: r for r in rows} | |
| assert by["acme"]["total"] == 3 and by["acme"]["ok"] == 2 and by["acme"]["rejected"] == 1 | |
| assert by["acme"]["billable"] == 2 # 2 billable ok runs; the rejected one isn't | |
| assert by["beta"]["total"] == 1 | |
| assert rows[0]["tenant"] == "acme" # most active first | |
| def test_run_ledger_stores_lawn_sqft_and_fetches_by_result_id(): | |
| led = RunLedger(connect(":memory:")) | |
| led.record("acme", "1 A St", "batch", lawn_sqft=4200.0, result_id="r1", zip="68106") | |
| row = led.get_by_result_id("acme", "r1") | |
| assert row["lawn_sqft"] == 4200.0 and row["surface"] == "batch" and row["zip"] == "68106" | |
| assert led.get_by_result_id("acme", "nope") is None | |
| assert led.get_by_result_id("beta", "r1") is None # tenant-scoped | |
| def test_measurement_edit_store(): | |
| st = MeasurementEditStore(connect(":memory:")) | |
| st.record("acme", "r1", 4500.0, original_sqft=4000.0, address="1 A St", user_id=1) | |
| got = st.recent("acme") | |
| assert got[0]["edited_sqft"] == 4500.0 and got[0]["original_sqft"] == 4000.0 | |
| assert st.recent("beta") == [] | |
| def test_lead_store_records_and_lists_newest_first(): | |
| store = LeadStore(connect(":memory:")) | |
| store.record("acme", "widget", "1 A St", name="Sam", email="s@x.com", phone="402-555-0100") | |
| store.record("acme", "quote", "2 B St", email="b@x.com", zip="68106", | |
| lawn_sqft=4200.0, price_total=90.0, confidence="high", result_id="r1") | |
| store.record("beta", "widget", "9 Z St", email="z@x.com") # other tenant, filtered out | |
| leads = store.recent("acme") | |
| assert [lead["address"] for lead in leads] == ["2 B St", "1 A St"] # newest first | |
| quote_lead = leads[0] | |
| assert quote_lead["source"] == "quote" | |
| assert quote_lead["lawn_sqft"] == 4200.0 | |
| assert quote_lead["price_total"] == 90.0 | |
| assert quote_lead["result_id"] == "r1" | |
| widget_lead = leads[1] | |
| assert widget_lead["phone"] == "402-555-0100" | |
| assert widget_lead["lawn_sqft"] is None # captured before measurement | |
| assert store.recent("gamma") == [] # unknown tenant | |
| def test_tenant_store_roundtrips_and_upserts(): | |
| store = TenantStore(connect(":memory:")) | |
| assert store.get("acme") is None # never stored | |
| store.put("acme", {"company": "Acme", "currency": "USD"}) | |
| assert store.get("acme") == {"company": "Acme", "currency": "USD"} | |
| store.put("acme", {"company": "Acme Renamed"}) # upsert replaces | |
| assert store.get("acme") == {"company": "Acme Renamed"} | |
| assert store.all_ids() == ["acme"] | |
| def test_user_store_maps_clerk_user_to_tenant_and_role(): | |
| users = UserStore(connect(":memory:")) | |
| uid = users.create("acme", "clerk_abc", "owner@acme.com", "owner", status="active") | |
| got = users.by_clerk_id("clerk_abc") | |
| assert got["tenant_id"] == "acme" and got["role"] == "owner" and got["email"] == "owner@acme.com" | |
| assert users.by_clerk_id("clerk_nope") is None # unprovisioned Clerk user → denied | |
| users.set_role(uid, "staff") | |
| assert users.by_clerk_id("clerk_abc")["role"] == "staff" | |
| users.create("acme", "clerk_def", "staff@acme.com", "staff") | |
| assert {u["email"] for u in users.by_tenant("acme")} == {"owner@acme.com", "staff@acme.com"} | |
| def test_audit_log_is_append_only_scoped_and_newest_first(): | |
| audit = AuditLog(connect(":memory:")) | |
| audit.record("config.publish", tenant_id="acme", user_id=1, | |
| before={"rate": 5}, after={"rate": 6}, ip="1.2.3.4") | |
| audit.record("user.invite", tenant_id="acme", user_id=1, target_type="user", target_id="7") | |
| audit.record("config.publish", tenant_id="beta", user_id=9) # other tenant | |
| rows = audit.recent("acme") | |
| assert [r["action"] for r in rows] == ["user.invite", "config.publish"] # newest first | |
| assert '"rate": 6' in rows[1]["after"] and rows[1]["ip"] == "1.2.3.4" | |
| assert audit.recent("beta")[0]["action"] == "config.publish" # scoped | |
| def test_config_store_drafts_and_versions(): | |
| cs = ConfigStore(connect(":memory:")) | |
| assert cs.get_draft("acme") is None | |
| cs.save_draft("acme", {"company": "Acme", "rate": 5}, user_id=1) | |
| assert cs.get_draft("acme") == {"company": "Acme", "rate": 5} | |
| cs.save_draft("acme", {"company": "Acme", "rate": 6}) # upsert | |
| assert cs.get_draft("acme")["rate"] == 6 | |
| cs.add_version("acme", {"v": 1}) | |
| cs.add_version("acme", {"v": 2}) | |
| versions = cs.versions("acme") | |
| assert [v["config"]["v"] for v in versions] == [2, 1] # newest first | |
| cs.clear_draft("acme") | |
| assert cs.get_draft("acme") is None | |
| class _Dropped(Exception): | |
| """Stand-in for psycopg.OperationalError/InterfaceError (psycopg isn't installed locally).""" | |
| def _fake_pg_retry(db): | |
| """Retrofit a real SQLite Database into the postgres retry path with fake connections.""" | |
| db.is_postgres = True | |
| db._conn_errors = (_Dropped,) | |
| def test_execute_reconnects_once_on_a_dropped_connection(): | |
| # Neon drops idle connections; a dropped conn must reconnect + retry, not surface a 500. | |
| db = connect(":memory:") | |
| _fake_pg_retry(db) | |
| calls = {"n": 0} | |
| class DeadConn: | |
| def execute(self, sql, params): | |
| calls["n"] += 1 | |
| raise _Dropped("server closed the connection unexpectedly") | |
| class LiveConn: | |
| def execute(self, sql, params): | |
| calls["n"] += 1 | |
| return "ok-cursor" | |
| db._conn = DeadConn() | |
| db._connect_pg = lambda: setattr(db, "_conn", LiveConn()) # "reconnect" swaps in a live conn | |
| assert db.execute("SELECT 1") == "ok-cursor" # retried on the reconnected connection | |
| assert calls["n"] == 2 # exactly one failed attempt + one retry | |
| def test_execute_retries_only_once_then_propagates(): | |
| # A real outage (still dead after reconnect) must raise, not loop forever. | |
| import pytest | |
| db = connect(":memory:") | |
| _fake_pg_retry(db) | |
| calls = {"n": 0} | |
| class DeadConn: | |
| def execute(self, sql, params): | |
| calls["n"] += 1 | |
| raise _Dropped("still down") | |
| db._conn = DeadConn() | |
| db._connect_pg = lambda: setattr(db, "_conn", DeadConn()) # reconnect, but still dead | |
| with pytest.raises(_Dropped): | |
| db.execute("SELECT 1") | |
| assert calls["n"] == 2 # original + exactly one retry, then propagate | |
| def test_user_store_invite_activate_and_caps(): | |
| users = UserStore(connect(":memory:")) | |
| users.create("acme", "clerk_owner", "owner@acme.com", "owner", status="active") | |
| # A pending invite has no Clerk id and is found by email until it's accepted. | |
| iid = users.invite("acme", "Staff@Acme.com", "staff", invited_by=1) | |
| assert users.by_clerk_id("clerk_new") is None | |
| pending = users.pending_by_email("staff@acme.com") # case-insensitive | |
| assert pending["id"] == iid and pending["role"] == "staff" and pending["clerk_user_id"] is None | |
| # First sign-in binds the Clerk id + activates it. | |
| users.activate(iid, "clerk_new") | |
| assert users.pending_by_email("staff@acme.com") is None # no longer pending | |
| assert users.by_clerk_id("clerk_new")["status"] == "active" | |
| # Seats count active + pending (not disabled); owners are counted for the last-owner guard. | |
| assert users.seat_count("acme") == 2 and users.active_owner_count("acme") == 1 | |
| users.set_status(iid, "disabled") | |
| assert users.seat_count("acme") == 1 | |
| def test_job_store_persist_resume_scope_and_evict(): | |
| jobs = JobStore(connect(":memory:")) | |
| jobs.create("j1", client="acme", imagery="google", addresses=["1 A St", "2 B St", "3 C St"]) | |
| # Freshly created -> queued, worker sees it with completed=0 and decoded addresses. | |
| assert jobs.pending_ids() == ["j1"] | |
| w = jobs.for_worker("j1") | |
| assert w["client"] == "acme" and w["imagery"] == "google" and w["completed"] == 0 | |
| assert w["addresses"] == ["1 A St", "2 B St", "3 C St"] | |
| # Worker runs the first address, then the app "restarts" mid-batch (nothing lost). | |
| jobs.mark_running("j1") | |
| jobs.append_result("j1", 1, {"address": "1 A St", "status": "ok", "result_id": "r1"}) | |
| assert jobs.pending_ids() == ["j1"] # still resumable after a restart | |
| assert jobs.for_worker("j1")["completed"] == 1 # resumes from index 1, not 0 | |
| # Finish the batch; the poll view is scoped to the owning client. | |
| jobs.append_result("j1", 2, {"address": "2 B St", "status": "ok", "result_id": "r2"}) | |
| jobs.append_result("j1", 3, {"address": "3 C St", "status": "error"}) | |
| jobs.mark_done("j1") | |
| assert jobs.pending_ids() == [] | |
| view = jobs.get("j1", "acme") | |
| assert view["status"] == "done" and view["total"] == 3 and view["completed"] == 3 | |
| assert [r.get("result_id") for r in view["results"]] == ["r1", "r2", None] | |
| assert jobs.get("j1", "someone-else") is None # a job never leaks across clients | |
| # Idempotent replay of an index overwrites rather than duplicating. | |
| jobs.append_result("j1", 2, {"address": "2 B St", "status": "ok", "result_id": "r2b"}) | |
| assert jobs.get("j1", "acme")["results"][1]["result_id"] == "r2b" | |
| # Eviction returns the result_ids to forget and only removes DONE jobs beyond `keep`. | |
| jobs.create("j2", client="acme", imagery="google", addresses=["x"]) # queued, must survive | |
| forget = jobs.evict_old(keep=0) | |
| assert set(forget) == {"r1", "r2b"} # None result_ids are skipped | |
| assert jobs.get("j1", "acme") is None # the done job is gone | |
| assert jobs.pending_ids() == ["j2"] # the queued job is untouched | |
| def test_run_ledger_overview_aggregates(): | |
| led = RunLedger(connect(":memory:")) | |
| led.record("acme", "1 A St", "single", zip="68106", duration_s=20.0, lawn_sqft=4000.0) | |
| led.record("acme", "2 B St", "batch", zip="68106", duration_s=40.0, lawn_sqft=6000.0) | |
| led.record_failure("acme", "x", "single", error_class="ValueError", outcome="rejected", zip="68164") | |
| led.record("beta", "9 Z", "batch", zip="68007") # other tenant — excluded | |
| ov = led.overview("acme", days=30) | |
| assert ov["quotes"] == 3 and ov["ok"] == 2 and ov["rejected"] == 1 and ov["errors"] == 0 | |
| assert ov["by_surface"] == {"single": 2, "batch": 1} | |
| assert ov["avg_duration_s"] == 30.0 and ov["avg_lawn_sqft"] == 5000 # ok runs only | |
| assert ov["top_zips"][0] == {"zip": "68106", "count": 2} | |
| assert len(ov["volume_series"]) == 30 | |
| assert ov["volume_series"][-1]["count"] == 3 and sum(p["count"] for p in ov["volume_series"]) == 3 | |
| assert RunLedger(connect(":memory:")).overview("nobody")["quotes"] == 0 | |
| def test_api_key_store_issue_and_lookup(): | |
| store = ApiKeyStore(connect(":memory:")) | |
| store.issue("acme", "pub-abc", kind="publishable", label="acme") | |
| store.issue("acme", "sec-xyz", kind="secret") | |
| rows = store.all() | |
| assert {"pub-abc", "sec-xyz"} <= {r["api_key"] for r in rows} | |
| assert store.key_for_tenant("acme", "publishable") == "pub-abc" | |
| assert store.key_for_tenant("acme", "secret") == "sec-xyz" | |
| assert store.key_for_tenant("nobody") is None | |
| def test_lead_store_summary_conversion(): | |
| store = LeadStore(connect(":memory:")) | |
| store.record("acme", "widget", "1 A", email="a@x") | |
| store.record("acme", "booking", "1 A", price_total=90.0, result_id="r1") | |
| store.record("acme", "booking", "2 B", price_total=110.0, result_id="r2") | |
| store.record("acme", "batch", "3 C", price_total=200.0) # operator quote — NOT a customer lead | |
| store.record("beta", "widget", "9 Z") # other tenant | |
| s = store.summary("acme", days=30) | |
| assert s["leads"] == 3 and s["bookings"] == 2 and s["conversion"] == round(2 / 3, 3) # batch excluded | |
| assert s["by_source"] == {"widget": 1, "booking": 2} and "batch" not in s["by_source"] | |
| assert s["avg_price"] == 100.0 # avg of BOOKED prices (90, 110); the batch 200 is ignored | |
| def test_lead_store_summary_top_packages(): | |
| import json | |
| store = LeadStore(connect(":memory:")) | |
| spring = json.dumps({"bundle": {"id": "spring", "name": "Spring Package", "cadence": "seasonal"}}) | |
| store.record("acme", "booking", "1 A", price_total=90.0, result_id="r1", detail=spring) | |
| store.record("acme", "booking", "2 B", price_total=90.0, result_id="r2", detail=spring) | |
| store.record("acme", "booking", "3 C", price_total=50.0, result_id="r3") # à-la-carte, no bundle | |
| top = store.summary("acme", days=30)["top_packages"] | |
| assert top and top[0] == {"name": "Spring Package", "count": 2} | |