"""Tests for the Project Schedule Variance detector (P3 — slippage tracking)."""
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_slippage.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, Connector
from app.services.leak_detectors.project_schedule_variance import (
ProjectScheduleVarianceDetector,
)
class ProjectSlippageTestCase(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 = ProjectScheduleVarianceDetector()
def setUp(self):
self._ctx = self.app.app_context()
self._ctx.push()
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", None),
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_project_on_schedule_returns_empty(self):
"""Project completed on or before planned end -> no leak."""
now = datetime.now(timezone.utc)
self._make_project(
id="p1",
name="On Schedule",
budget=50000,
start_date=now - timedelta(days=90),
end_date=now - timedelta(days=30),
completed_date=now - timedelta(days=30), # completed on planned end
)
results = self.detector.check(self.company.id)
self.assertEqual(len(results), 0)
def test_project_early_returns_empty(self):
"""Project completed early -> no leak."""
now = datetime.now(timezone.utc)
self._make_project(
id="p1",
name="Early Finish",
budget=50000,
start_date=now - timedelta(days=90),
end_date=now - timedelta(days=30),
completed_date=now - timedelta(days=40), # completed 10 days early
)
results = self.detector.check(self.company.id)
self.assertEqual(len(results), 0)
def test_slipped_project_detected(self):
"""Project completed 20 days late -> flagged."""
now = datetime.now(timezone.utc)
self._make_project(
id="p1",
name="Slipped Kitchen",
budget=100000,
actual_cost=90000,
start_date=now - timedelta(days=90),
end_date=now - timedelta(days=30), # planned end 30 days ago
completed_date=now - timedelta(days=10), # actually finished 10 days ago
status="completed",
)
# Planned duration: 60 days, actual: 80 days -> 33% slippage
results = self.detector.check(self.company.id)
self.assertGreater(len(results), 0)
agg = results[0]
self.assertIn("Slipped Kitchen", agg.description)
self.assertIn("behind", agg.description.lower())
self.assertGreater(agg.estimated_loss, 0)
def test_below_threshold_not_flagged(self):
"""Small slippage (< 10%) -> not flagged."""
now = datetime.now(timezone.utc)
# 100 day duration, completed 5 days late -> 5% slippage
self._make_project(
id="p1",
name="Minor Slip",
budget=50000,
start_date=now - timedelta(days=105),
end_date=now - timedelta(days=5),
completed_date=now, # 5 days late
status="completed",
)
results = self.detector.check(self.company.id)
self.assertEqual(len(results), 0)
def test_below_min_slippage_days_not_flagged(self):
"""Slippage < 3 days -> noise floor, not flagged."""
now = datetime.now(timezone.utc)
# 60 day duration, completed 2 days late -> 3.3% slippage, < 3 days
self._make_project(
id="p1",
name="Tiny Slip",
budget=50000,
start_date=now - timedelta(days=62),
end_date=now - timedelta(days=2),
completed_date=now,
status="completed",
)
results = self.detector.check(self.company.id)
self.assertEqual(len(results), 0)
# -- Active project tests ------------------------------------------------
def test_active_project_past_deadline_detected(self):
"""Active project past planned end -> flagged as behind."""
now = datetime.now(timezone.utc)
self._make_project(
id="p1",
name="Active Overdue",
budget=75000,
start_date=now - timedelta(days=60),
end_date=now - timedelta(days=10), # deadline was 10 days ago
completed_date=None,
status="active",
)
results = self.detector.check(self.company.id)
self.assertGreater(len(results), 0)
agg = results[0]
self.assertIn("Active Overdue", agg.description)
def test_active_project_near_deadline_at_risk(self):
"""Active project near deadline -> flagged as at risk."""
now = datetime.now(timezone.utc)
# Started 58 days ago, 60 day duration, 2 days left
self._make_project(
id="p1",
name="Near Deadline",
budget=50000,
start_date=now - timedelta(days=58),
end_date=now + timedelta(days=2), # 2 days left (< min_slippage_days=3)
completed_date=None,
status="active",
)
results = self.detector.check(self.company.id)
self.assertGreater(len(results), 0)
combined_desc = " ".join(r.description for r in results)
self.assertIn("at risk", combined_desc.lower())
# -- Severity tests ------------------------------------------------------
def test_severity_scales_with_impact(self):
"""Severity increases with total margin impact."""
now = datetime.now(timezone.utc)
# Create 5 projects with significant slippage
for i in range(5):
self._make_project(
id=f"p{i}",
name=f"Slipped Project {i}",
budget=100000,
actual_cost=95000,
start_date=now - timedelta(days=90),
end_date=now - timedelta(days=30),
completed_date=now - timedelta(days=10), # 20 days late
status="completed",
)
results = self.detector.check(self.company.id)
agg = results[0]
self.assertIn(agg.severity, ("high", "critical"))
def test_severity_low_for_minor_slippage(self):
"""Small margin impact -> low severity."""
now = datetime.now(timezone.utc)
# Project with small budget, moderate slippage
self._make_project(
id="p1",
name="Small Slipped",
budget=5000,
actual_cost=4000,
start_date=now - timedelta(days=30),
end_date=now - timedelta(days=5),
completed_date=now, # 5 days late on 25-day project -> 20%
status="completed",
)
results = self.detector.check(self.company.id)
agg = results[0]
# With only ~$500/day * 5 days = $2.5K impact, should be low
self.assertEqual(agg.severity, "low")
# -- Metadata tests ------------------------------------------------------
def test_metadata_has_required_keys(self):
"""Metadata includes dedupe_key and project details."""
now = datetime.now(timezone.utc)
self._make_project(
id="p1",
name="Bath Remodel",
budget=80000,
actual_cost=75000,
start_date=now - timedelta(days=60),
end_date=now - timedelta(days=10),
completed_date=now - timedelta(days=1), # 9 days late on 50-day project -> 18%
status="completed",
)
results = self.detector.check(self.company.id)
agg = results[0]
self.assertIn("dedupe_key", agg.metadata_json)
self.assertIn("slipped_count", agg.metadata_json)
self.assertIn("total_slippage_days", agg.metadata_json)
self.assertIn("total_margin_impact", agg.metadata_json)
self.assertIn("threshold_pct", agg.metadata_json)
self.assertIn("projects", agg.metadata_json)
def test_individual_candidates_included(self):
"""Each slipped project gets its own candidate for drill-down."""
now = datetime.now(timezone.utc)
self._make_project(
id="p1",
name="Project A",
budget=60000,
actual_cost=55000,
start_date=now - timedelta(days=60),
end_date=now - timedelta(days=10),
completed_date=now, # 10 days late
status="completed",
)
self._make_project(
id="p2",
name="Project B",
budget=80000,
actual_cost=78000,
start_date=now - timedelta(days=90),
end_date=now - timedelta(days=20),
completed_date=now - timedelta(days=5), # 15 days late
status="completed",
)
results = self.detector.check(self.company.id)
# 1 aggregate + 2 individual = 3 candidates
self.assertEqual(len(results), 3)
# First is aggregate
self.assertEqual(results[0].source, "Project Schedule Variance")
# Rest are individual
sources = [r.source for r in results[1:]]
self.assertTrue(any("Project A" in s for s in sources))
self.assertTrue(any("Project B" in s for s in sources))
# -- active_only param test ----------------------------------------------
def test_active_only_param(self):
"""active_only=True -> only checks active/planning projects."""
now = datetime.now(timezone.utc)
# Completed slipped project (should be ignored with active_only)
self._make_project(
id="p1",
name="Completed Slipped",
budget=100000,
start_date=now - timedelta(days=90),
end_date=now - timedelta(days=30),
completed_date=now - timedelta(days=10),
status="completed",
)
# Active overdue project (should be detected)
self._make_project(
id="p2",
name="Active Overdue",
budget=75000,
start_date=now - timedelta(days=60),
end_date=now - timedelta(days=5),
completed_date=None,
status="active",
)
# Default (active_only=False) should find both
results = self.detector.check(self.company.id)
self.assertGreaterEqual(len(results), 2)
# active_only=True should only find the active one
self.detector.default_params["active_only"] = True
results_active = self.detector.check(self.company.id)
combined_desc = " ".join(r.description for r in results_active)
self.assertIn("Active Overdue", combined_desc)
self.assertNotIn("Completed Slipped", combined_desc)
# Reset
self.detector.default_params["active_only"] = False
# -- Daily cost estimation -----------------------------------------------
def test_daily_cost_from_actual_cost(self):
"""Daily cost estimation prefers actual_cost over budget."""
now = datetime.now(timezone.utc)
proj = self._make_project(
id="p1",
name="Cost Test",
budget=100000,
actual_cost=80000,
start_date=now - timedelta(days=60),
end_date=now - timedelta(days=10),
completed_date=now,
status="completed",
)
daily = ProjectScheduleVarianceDetector._estimate_daily_cost(proj, 60)
self.assertAlmostEqual(daily, 80000 / 60, delta=1)
def test_daily_cost_from_budget_fallback(self):
"""When no actual_cost, falls back to budget."""
now = datetime.now(timezone.utc)
proj = self._make_project(
id="p1",
name="Budget Fallback",
budget=60000,
actual_cost=0,
start_date=now - timedelta(days=60),
end_date=now - timedelta(days=10),
completed_date=now,
status="completed",
)
daily = ProjectScheduleVarianceDetector._estimate_daily_cost(proj, 60)
self.assertAlmostEqual(daily, 60000 / 60, delta=1)
def test_daily_cost_default(self):
"""No budget or actual_cost -> $500/day default."""
now = datetime.now(timezone.utc)
proj = Project(
id="p1",
company_id=self.company.id,
name="No Cost",
status="active",
budget=None,
revenue=None,
actual_cost=None,
start_date=now - timedelta(days=30),
end_date=now + timedelta(days=30),
)
daily = ProjectScheduleVarianceDetector._estimate_daily_cost(proj, 60)
self.assertAlmostEqual(daily, 500.0, delta=1)
if __name__ == '__main__':
unittest.main()