"""Smoke test for Phase 1 leak detectors — in-memory SQLite, seeded data."""
import os
os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
os.environ.setdefault("SQLALCHEMY_DATABASE_URI", "sqlite:///:memory:")
os.environ.setdefault("RATELIMIT_STORAGE_URI", "memory://")
os.environ.setdefault("SECRET_KEY", "test")
os.environ.setdefault("DISABLE_SCHEDULER", "1")
from datetime import datetime, timedelta, timezone
from flask import Flask
from app.models import (
db, Company, Connector, QuickbooksInvoice, CrmContact, CrmDeal,
AdCampaign, AdMetric, ExternalSyncRecord, RevenueLeak,
)
from app.services.leak_detectors import run_all_detectors, DETECTORS
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
db.init_app(app)
NOW = datetime.now(timezone.utc)
with app.app_context():
db.create_all()
co = Company(name="Test Co")
db.session.add(co)
db.session.flush()
cid = co.id
for svc in ("quickbooks", "hubspot", "google_ads", "leadperfection"):
db.session.add(Connector(company_id=cid, service=svc, status="connected"))
# QB invoices: 45d overdue, 75d overdue, 120d overdue, paid old, draft 20d
for i, (status, due_days, tx_days, amt) in enumerate([
("Sent", 45, 60, 1000.0),
("Sent", 75, 90, 2000.0),
("Sent", 120, 130, 3000.0),
("Paid", 120, 130, 999.0), # excluded
("Draft", None, 20, 500.0), # draft detector
("Draft", None, 5, 400.0), # too fresh, excluded
]):
db.session.add(QuickbooksInvoice(
company_id=cid, qb_doc_id=f"qb-{i}", invoice_num=f"INV-{i}",
total_amount=amt, status=status,
due_date=(NOW - timedelta(days=due_days)) if due_days else None,
tx_date=NOW - timedelta(days=tx_days),
))
# CRM contact: unworked lead (3 days old, no deal)
lead = CrmContact(company_id=cid, external_id="c-1", first_name="Lonely",
last_name="Lead", lifecycle_stage="lead",
created_at=NOW - timedelta(days=3))
# Contact with a deal (should NOT be flagged)
worked = CrmContact(company_id=cid, external_id="c-2", first_name="Busy",
last_name="Contact", lifecycle_stage="lead",
created_at=NOW - timedelta(days=5))
db.session.add_all([lead, worked])
db.session.flush()
# Deals: stale 10d (medium), stale 20d (high), slipped close, orphaned, closed (excluded)
deals = [
CrmDeal(company_id=cid, external_id="d-1", name="Stale Ten", amount=10000,
probability=0.4, stage="proposal", deal_owner_id="own1",
updated_at=NOW - timedelta(days=10)),
CrmDeal(company_id=cid, external_id="d-2", name="Stale Twenty", amount=5000,
probability=None, stage="negotiation", deal_owner_id="own1",
updated_at=NOW - timedelta(days=20)),
CrmDeal(company_id=cid, external_id="d-3", name="Slipped", amount=8000,
probability=0.6, stage="proposal", deal_owner_id="own2",
expected_close_date=NOW - timedelta(days=12), updated_at=NOW),
CrmDeal(company_id=cid, external_id="d-4", name="Orphan", amount=7000,
probability=0.5, stage="qualified", deal_owner_id="",
updated_at=NOW),
CrmDeal(company_id=cid, external_id="d-5", name="Won", amount=9000,
stage="closedwon", deal_owner_id="", updated_at=NOW - timedelta(days=30)),
CrmDeal(company_id=cid, external_id="d-6", name="Busy Deal", amount=1000,
stage="proposal", deal_owner_id="own3", contact_id=worked.id,
updated_at=NOW),
]
db.session.add_all(deals)
# Ads: campaign spends $600 over 7d, 0 conversions -> high
camp = AdCampaign(company_id=cid, external_id="camp-1", source_service="google_ads",
name="Wasted Search")
db.session.add(camp)
for d in range(3):
db.session.add(AdMetric(company_id=cid, external_campaign_id="camp-1",
source_service="google_ads",
metric_date=(NOW - timedelta(days=d)).date(),
spend=200.0, conversions=0.0))
# converting campaign — excluded
db.session.add(AdMetric(company_id=cid, external_campaign_id="camp-2",
source_service="google_ads",
metric_date=NOW.date(), spend=300.0, conversions=2.0))
# External lead 4d old, no deal
db.session.add(ExternalSyncRecord(company_id=cid, external_id="ext-1",
source_service="leadperfection",
entity_type="lead", name="Ext Lead",
created_at=NOW - timedelta(days=4)))
db.session.commit()
# ---- Run 1 ----
r1 = run_all_detectors(cid)
print("RUN1:", {k: r1[k] for k in ("scanned", "new_leaks", "updated_leaks", "errors")})
for l in sorted(r1["leaks"], key=lambda x: x["detector_id"]):
print(f" [{l['detector_id']}] {l['severity']:8s} ${l['estimated_loss']} {l['description']}")
# ---- Run 2 (idempotency) ----
r2 = run_all_detectors(cid)
print("RUN2:", {k: r2[k] for k in ("scanned", "new_leaks", "updated_leaks", "errors")})
total = RevenueLeak.query.filter_by(company_id=cid, detection_type="auto").count()
print("total auto leaks in DB:", total)
# Assertions
assert r1["scanned"] == 7
assert not r1["errors"], r1["errors"]
assert r2["new_leaks"] == 0, "second run must not create duplicates"
assert r2["updated_leaks"] == r1["new_leaks"]
ids = sorted({l["detector_id"] for l in r1["leaks"]})
assert ids == sorted([d.id for d in DETECTORS]), ids
# spot-check values
by_id = {}
for l in r1["leaks"]:
by_id.setdefault(l["detector_id"], []).append(l)
assert len(by_id["qb_overdue_30d"]) == 3 # three buckets
sevs = sorted(l["severity"] for l in by_id["qb_overdue_30d"])
assert sevs == ["critical", "high", "medium"], sevs
assert by_id["qb_draft_invoices"][0]["estimated_loss"] == 500.0
assert len(by_id["crm_stale_deals"]) == 2
stale_sevs = {l["description"].split("'")[1]: l["severity"] for l in by_id["crm_stale_deals"]}
assert stale_sevs == {"Stale Ten": "medium", "Stale Twenty": "high"}, stale_sevs
assert by_id["crm_slipped_close"][0]["estimated_loss"] == 4800.0 # 8000*0.6
assert by_id["crm_orphaned_deals"][0]["estimated_loss"] == 3500.0 # 7000*0.5
assert by_id["ads_zero_conversion"][0]["severity"] == "high" # $600
assert by_id["ads_zero_conversion"][0]["estimated_loss"] == 600.0
assert len(by_id["leads_unworked"]) == 2 # crm lead + external lead
# Resolve one leak, re-run: should create a fresh one (created_after_resolved)
leak = RevenueLeak.query.filter_by(company_id=cid, detector_id="crm_orphaned_deals").first()
leak.resolved = True
db.session.commit()
r3 = run_all_detectors(cid)
assert r3["new_leaks"] == 1 and r3["already_resolved"] == 1, (r3["new_leaks"], r3["already_resolved"])
# Disconnect quickbooks connector -> QB detectors return nothing new
Connector.query.filter_by(company_id=cid, service="quickbooks").update({"status": "error"})
db.session.commit()
r4 = run_all_detectors(cid)
qb_leaks = [l for l in r4["leaks"] if l["detector_id"].startswith("qb_")]
assert qb_leaks == [], qb_leaks
print("ALL ASSERTIONS PASSED")