"""Tests for the Backlog Burndown 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_backlog.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, Connector
from app.services.leak_detectors.backlog_burndown import BacklogBurndownDetector
class BacklogBurndownTestCase(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 = BacklogBurndownDetector()
def setUp(self):
self._ctx = self.app.app_context()
self._ctx.push()
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", "active"),
budget=kwargs.pop("budget", 50000),
revenue=kwargs.pop("revenue", 0),
actual_cost=kwargs.pop("actual_cost", 0),
start_date=kwargs.pop("start_date", datetime.now(timezone.utc) - timedelta(days=30)),
end_date=kwargs.pop("end_date", datetime.now(timezone.utc) + timedelta(days=30)),
completed_date=kwargs.pop("completed_date", None),
)
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", 25000),
stage=kwargs.pop("stage", "appointments_set"),
probability=kwargs.pop("probability", 0.5),
)
db.session.add(d)
db.session.commit()
return d
# -- Tests ----------------------------------------------------------------
def test_empty_backlog_returns_critical(self):
"""No projects, no deals → critical alert."""
results = self.detector.check(self.company.id)
self.assertEqual(len(results), 1)
self.assertEqual(results[0].severity, "critical")
self.assertIn("No active backlog", results[0].description)
def test_healthy_backlog_returns_low(self):
"""Sufficient backlog → low severity."""
# Add 12 weeks of completed projects to establish burn rate
for i in range(12):
self._make_project(
id=f"completed-{i}",
name=f"Completed {i}",
status="completed",
budget=10000,
revenue=10000,
completed_date=datetime.now(timezone.utc) - timedelta(weeks=i),
)
# Add backlog worth 8 weeks (above 4-week threshold)
for i in range(4):
self._make_project(
id=f"backlog-{i}",
name=f"Backlog {i}",
status="active",
budget=20000,
)
results = self.detector.check(self.company.id)
self.assertEqual(len(results), 1)
self.assertIn("low", results[0].severity)
self.assertIn("above", results[0].description.lower())
def test_low_backlog_returns_alert(self):
"""Backlog below threshold → alert with severity."""
# Add 12 weeks of completed projects for burn rate
for i in range(12):
self._make_project(
id=f"completed-{i}",
name=f"Completed {i}",
status="completed",
budget=10000,
revenue=10000,
completed_date=datetime.now(timezone.utc) - timedelta(weeks=i),
)
# Only 2 weeks of backlog (below 4-week threshold)
self._make_project(
id="backlog-1",
name="Backlog 1",
status="active",
budget=20000,
)
results = self.detector.check(self.company.id)
# Should emit main candidate + revenue-at-risk candidate
self.assertGreaterEqual(len(results), 2)
# Main alert should be medium or higher
self.assertIn(
results[0].severity,
("medium", "high", "critical"),
)
self.assertIn("below", results[0].description.lower())
def test_revenue_at_risk_included(self):
"""Revenue-at-risk candidate is emitted when coverage is low."""
for i in range(12):
self._make_project(
id=f"completed-{i}",
name=f"Completed {i}",
status="completed",
budget=10000,
revenue=10000,
completed_date=datetime.now(timezone.utc) - timedelta(weeks=i),
)
# Zero active backlog
results = self.detector.check(self.company.id)
revenue_at_risk = [r for r in results if "revenue_at_risk" in r.metadata_json]
self.assertEqual(len(revenue_at_risk), 1)
self.assertGreater(revenue_at_risk[0].estimated_loss or 0, 0)
def test_includes_crm_deals_in_backlog(self):
"""CRM pipeline deals contribute to backlog."""
for i in range(12):
self._make_project(
id=f"completed-{i}",
status="completed",
budget=10000,
revenue=10000,
completed_date=datetime.now(timezone.utc) - timedelta(weeks=i),
)
self._make_deal(
id="deal-1",
external_id="ext-1",
amount=50000,
stage="appointments_set",
)
results = self.detector.check(self.company.id)
metadata = results[0].metadata_json
self.assertEqual(metadata["pipeline_deal_count"], 1)
def test_no_double_count_won_deals(self):
"""Won deals are excluded from backlog."""
for i in range(12):
self._make_project(
id=f"completed-{i}",
status="completed",
budget=10000,
revenue=10000,
completed_date=datetime.now(timezone.utc) - timedelta(weeks=i),
)
self._make_deal(id="won-1", external_id="ext-won", amount=50000, stage="closedwon")
self._make_deal(id="lost-1", external_id="ext-lost", amount=50000, stage="closedlost")
results = self.detector.check(self.company.id)
metadata = results[0].metadata_json
self.assertEqual(metadata["pipeline_deal_count"], 0)
def test_metadata_has_required_keys(self):
"""All expected keys present in metadata."""
results = self.detector.check(self.company.id)
keys = {
"dedupe_key", "total_backlog", "weekly_burn_rate",
"weeks_coverage", "active_project_count",
}
self.assertTrue(keys.issubset(results[0].metadata_json.keys()))
if __name__ == '__main__':
unittest.main()