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