"""Tests for the Margin Per Job Tracking detector (P3 — margin monitoring)."""
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_project_margin.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, Estimate, QuickbooksExpense, Connector
from app.services.leak_detectors.project_margin_leak import ProjectMarginLeakDetector
class ProjectMarginTestCase(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 = ProjectMarginLeakDetector()
def setUp(self):
self._ctx = self.app.app_context()
self._ctx.push()
db.session.query(QuickbooksExpense).delete()
db.session.query(Estimate).delete()
db.session.query(Project).delete()
db.session.query(Connector).delete()
db.session.query(Company).delete()
db.session.commit()
self.company = Company(
id="test-co",
name="Test Build 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):
now = datetime.now(timezone.utc)
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", now - timedelta(days=90)),
end_date=kwargs.pop("end_date", now - timedelta(days=30)),
completed_date=kwargs.pop("completed_date", now - timedelta(days=30)),
assigned_to=kwargs.pop("assigned_to", None),
)
db.session.add(p)
db.session.commit()
return p
# -- Basic 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_good_margin_returns_empty(self):
"""Project with margin above threshold -> no leak."""
# 25% margin > 15% threshold
self._make_project(
id="p1",
name="Healthy Margin",
budget=50000,
revenue=50000,
actual_cost=37500, # 25% margin
)
results = self.detector.check(self.company.id)
self.assertEqual(len(results), 0)
def test_low_margin_detected(self):
"""Project with margin below threshold -> flagged."""
# 10% margin < 15% threshold
self._make_project(
id="p1",
name="Low Margin Bath",
budget=40000,
revenue=40000,
actual_cost=36000, # 10% margin
)
results = self.detector.check(self.company.id)
self.assertGreater(len(results), 0)
agg = results[0]
self.assertIn("Low Margin Bath", agg.description)
self.assertIn("margin", agg.description.lower())
def test_negative_margin_critical(self):
"""Project losing money -> critical severity."""
self._make_project(
id="p1",
name="Losing Money Kitchen",
budget=50000,
revenue=48000,
actual_cost=52000, # Lost $4K
)
results = self.detector.check(self.company.id)
self.assertGreater(len(results), 0)
agg = results[0]
self.assertEqual(agg.severity, "critical")
self.assertIn("losing money", agg.description.lower())
self.assertGreater(agg.estimated_loss, 0)
def test_below_min_value_not_flagged(self):
"""Small project below min_project_value -> ignored."""
self._make_project(
id="p1",
name="Tiny Job",
budget=2000,
revenue=2000,
actual_cost=1800, # 10% margin — but under $5K min
)
results = self.detector.check(self.company.id)
self.assertEqual(len(results), 0)
# -- Multiple projects ---------------------------------------------------
def test_aggregate_multiple_projects(self):
"""Multiple low-margin projects -> aggregated correctly."""
# 12% margin
self._make_project(
id="p1",
name="Kitchen A",
budget=50000,
revenue=50000,
actual_cost=44000,
)
# 8% margin
self._make_project(
id="p2",
name="Bath B",
budget=30000,
revenue=30000,
actual_cost=27600,
)
results = self.detector.check(self.company.id)
agg = results[0]
self.assertEqual(agg.metadata_json["low_margin_count"], 2)
self.assertEqual(len(results), 3) # 1 aggregate + 2 individual
self.assertIn("Kitchen A", agg.description)
self.assertIn("Bath B", agg.description)
def test_margin_rollup_present(self):
"""Margin rollup by project type included in metadata."""
self._make_project(
id="p1",
name="Kitchen 1",
budget=60000,
revenue=60000,
actual_cost=54000, # 10%
)
self._make_project(
id="p2",
name="Kitchen 2",
budget=40000,
revenue=40000,
actual_cost=36000, # 10%
)
self._make_project(
id="p3",
name="Bath 1",
budget=25000,
revenue=25000,
actual_cost=23000, # 8%
)
results = self.detector.check(self.company.id)
agg = results[0]
self.assertIn("margin_by_type", agg.metadata_json)
by_type = agg.metadata_json["margin_by_type"]
self.assertIn("general", by_type)
self.assertEqual(by_type["general"]["count"], 3)
# -- Severity scaling ----------------------------------------------------
def test_severity_high_multiple_projects(self):
"""3+ high-severity projects -> high overall severity."""
for i in range(3):
self._make_project(
id=f"p{i}",
name=f"Low Margin {i}",
budget=30000,
revenue=30000,
actual_cost=28500, # 5% margin (< 7.5% → high per-project)
)
results = self.detector.check(self.company.id)
agg = results[0]
self.assertIn(agg.severity, ("high", "critical"))
# -- Estimate cross-reference --------------------------------------------
def test_estimate_low_margin_flagged(self):
"""Won estimate with low estimated margin -> flagged."""
est = Estimate(
id="est-1",
company_id=self.company.id,
prospect_name="Smith Kitchen",
project_type="kitchen",
estimate_amount=50000,
estimated_cost=45000, # 10% margin
stage="won",
)
db.session.add(est)
db.session.commit()
results = self.detector.check(self.company.id)
self.assertGreater(len(results), 0)
combined_desc = " ".join(r.description for r in results)
self.assertIn("Smith Kitchen", combined_desc)
# -- QB expense cross-reference ------------------------------------------
def test_qb_expense_context_added(self):
"""QB expenses aggregated by category when connector 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()
# Add a low-margin project
self._make_project(
id="p1",
name="Kitchen Remodel",
budget=40000,
revenue=40000,
actual_cost=36000,
)
# Add some QB expenses
for i in range(3):
expense = QuickbooksExpense(
id=f"exp-{i}",
company_id=self.company.id,
qb_doc_id=f"qb-exp-{i}",
amount=1000 + i * 500,
category="Materials" if i < 2 else "Labor",
tx_date=datetime.now(timezone.utc) - timedelta(days=30),
)
db.session.add(expense)
db.session.commit()
results = self.detector.check(self.company.id)
agg = results[0]
self.assertIn("_qb_context", agg.metadata_json)
qb_ctx = agg.metadata_json["_qb_context"]
self.assertIn("by_category", qb_ctx)
# -- Metadata tests ------------------------------------------------------
def test_metadata_has_required_keys(self):
"""Metadata includes all required fields."""
self._make_project(
id="p1",
name="Deck Build",
budget=35000,
revenue=35000,
actual_cost=32000, # 8.6% margin
)
results = self.detector.check(self.company.id)
agg = results[0]
self.assertIn("dedupe_key", agg.metadata_json)
self.assertIn("low_margin_count", agg.metadata_json)
self.assertIn("total_revenue", agg.metadata_json)
self.assertIn("total_cost", agg.metadata_json)
self.assertIn("total_profit", agg.metadata_json)
self.assertIn("aggregate_margin_pct", agg.metadata_json)
self.assertIn("margin_threshold_pct", agg.metadata_json)
self.assertIn("projects", agg.metadata_json)
# -- active_only test ----------------------------------------------------
def test_active_projects_included_when_configured(self):
"""include_active=True -> also checks active projects."""
self._make_project(
id="p1",
name="Active Low Margin",
budget=50000,
revenue=45000,
actual_cost=42000, # 6.7% margin
status="active",
completed_date=None,
)
# Default (include_active=False) -> should not find it
results = self.detector.check(self.company.id)
self.assertEqual(len(results), 0)
# include_active=True -> should find it
self.detector.default_params["include_active"] = True
results = self.detector.check(self.company.id)
self.assertGreater(len(results), 0)
self.assertIn("Active Low Margin", results[0].description)
# Reset
self.detector.default_params["include_active"] = False
if __name__ == '__main__':
unittest.main()