"""Tests for the Change Order Revenue Leakage detector (P1 — remodeling sector)."""
import os
import sys
import unittest
from datetime import datetime, timezone, timedelta
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
os.environ.setdefault('DATABASE_URL', 'sqlite:///test_change_order.db')
os.environ.setdefault('SECRET_KEY', 'test-secret-key-for-testing')
os.environ['DISABLE_SCHEDULER'] = '1'
from app import create_app
from app.models import db, Company, Project, CrmDeal, QuickbooksInvoice, Connector
from app.services.leak_detectors.change_order_leak import ChangeOrderLeakDetector
class ChangeOrderLeakTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.app = create_app()
cls.app.config['TESTING'] = True
with cls.app.app_context():
db.create_all()
cls.detector = ChangeOrderLeakDetector()
def setUp(self):
self._ctx = self.app.app_context()
self._ctx.push()
db.session.query(QuickbooksInvoice).delete()
db.session.query(Project).delete()
db.session.query(CrmDeal).delete()
db.session.query(Connector).delete()
db.session.query(Company).delete()
db.session.commit()
self.company = Company(
id="test-co",
name="Test Remodel Co",
settings_json={"leak_detectors": {}},
)
db.session.add(self.company)
db.session.commit()
def tearDown(self):
db.session.rollback()
self._ctx.pop()
def _make_project(self, **kwargs):
p = Project(
id=kwargs.pop("id", "proj-1"),
company_id=self.company.id,
name=kwargs.pop("name", "Test Project"),
status=kwargs.pop("status", "completed"),
budget=kwargs.pop("budget", 50000),
revenue=kwargs.pop("revenue", 50000),
actual_cost=kwargs.pop("actual_cost", 40000),
start_date=kwargs.pop("start_date", datetime.now(timezone.utc) - timedelta(days=60)),
completed_date=kwargs.pop("completed_date", datetime.now(timezone.utc) - timedelta(days=30)),
)
db.session.add(p)
db.session.commit()
return p
def _make_deal(self, **kwargs):
d = CrmDeal(
id=kwargs.pop("id", "deal-1"),
company_id=self.company.id,
external_id=kwargs.pop("external_id", "ext-1"),
name=kwargs.pop("name", "Test Deal"),
amount=kwargs.pop("amount", 50000),
stage=kwargs.pop("stage", "closedwon"),
probability=kwargs.pop("probability", 1.0),
)
d.updated_at = datetime.now(timezone.utc) - timedelta(days=30)
db.session.add(d)
db.session.commit()
return d
def _make_invoice(self, **kwargs):
i = QuickbooksInvoice(
id=kwargs.pop("id", "inv-1"),
company_id=self.company.id,
qb_doc_id=kwargs.pop("qb_doc_id", kwargs.pop("external_id", "ext-1")),
invoice_num=kwargs.pop("invoice_num", kwargs.pop("invoice_number", "INV-001")),
total_amount=kwargs.pop("total_amount", 50000),
status=kwargs.pop("status", "Paid"),
tx_date=kwargs.pop("tx_date", kwargs.pop("invoice_date", datetime.now(timezone.utc) - timedelta(days=30))),
due_date=kwargs.pop("due_date", datetime.now(timezone.utc)),
)
db.session.add(i)
db.session.commit()
return i
# -- Tests ----------------------------------------------------------------
def test_no_projects_returns_empty(self):
"""Empty company → no candidates."""
results = self.detector.check(self.company.id)
self.assertEqual(len(results), 0)
def test_unchanged_estimate_returns_empty(self):
"""Revenue matches budget → no leak detected."""
self._make_project(
id="p1",
name="On Budget",
budget=50000,
revenue=50000,
)
results = self.detector.check(self.company.id)
self.assertEqual(len(results), 0)
def test_uncaptured_change_order_detected(self):
"""Revenue significantly below budget → flagged."""
self._make_project(
id="p1",
name="Kitchen Remodel",
budget=100000,
revenue=75000, # 25% below budget
)
results = self.detector.check(self.company.id)
self.assertGreater(len(results), 0)
# Aggregate candidate
agg = results[0]
self.assertIn("Kitchen Remodel", agg.description)
self.assertIn("uncaptured", agg.description.lower())
self.assertAlmostEqual(agg.estimated_loss, 25000, delta=1)
def test_below_threshold_not_flagged(self):
"""Small variance (< 15%) → not flagged."""
self._make_project(
id="p1",
name="Minor Variance",
budget=100000,
revenue=90000, # 10% below — under 15% threshold
)
results = self.detector.check(self.company.id)
self.assertEqual(len(results), 0)
def test_small_project_ignored(self):
"""Projects under min_project_value threshold are skipped."""
self._make_project(
id="p1",
name="Small Job",
budget=2000,
revenue=1000, # 50% below — but under $5K min
)
results = self.detector.check(self.company.id)
self.assertEqual(len(results), 0)
def test_margin_leak_detected(self):
"""Actual cost exceeds revenue → margin leak flagged."""
self._make_project(
id="p1",
name="Margin Squeeze",
budget=50000,
revenue=45000, # close to budget
actual_cost=52000, # but cost exceeded revenue
)
results = self.detector.check(self.company.id)
self.assertGreater(len(results), 0)
# Should mention margin
combined_desc = " ".join(r.description for r in results)
self.assertIn("margin", combined_desc.lower())
def test_severity_scales_with_amount(self):
"""Severity increases with total uncaptured amount."""
# Critical: > $100K
for i in range(5):
self._make_project(
id=f"p{i}",
name=f"Big Project {i}",
budget=50000,
revenue=30000, # $20K uncaptured each
)
results = self.detector.check(self.company.id)
agg = results[0]
self.assertIn(agg.severity, ("high", "critical"))
def test_crm_deal_with_invoice_mismatch(self):
"""CRM deal estimate vs. QB invoice → detected."""
# Set up connector as available
conn = Connector(
id="conn-qb",
company_id=self.company.id,
service="quickbooks",
status="connected",
config_json={"access_token": "test"},
)
db.session.add(conn)
db.session.commit()
deal = self._make_deal(
id="deal-1",
external_id="job-123",
amount=100000,
name="Deck Build",
)
# Invoice is much lower than the won deal amount
self._make_invoice(
id="inv-1",
external_id="job-123",
total_amount=70000, # 30% below estimate
)
results = self.detector.check(self.company.id)
self.assertGreater(len(results), 0)
combined_desc = " ".join(r.description for r in results)
self.assertIn("Deck Build", combined_desc)
def test_metadata_has_required_keys(self):
"""Metadata includes dedupe_key and project details."""
self._make_project(
id="p1",
name="Bath Remodel",
budget=80000,
revenue=55000,
)
results = self.detector.check(self.company.id)
agg = results[0]
self.assertIn("dedupe_key", agg.metadata_json)
self.assertIn("project_count", agg.metadata_json)
self.assertIn("total_uncaptured", agg.metadata_json)
if __name__ == '__main__':
unittest.main()