"""Tests for Multi-Location Rollup Detector (P4).""" import os import sys import unittest from datetime import datetime, timezone, timedelta from uuid import uuid4 sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) os.environ.setdefault('DATABASE_URL', 'sqlite:///test_multi_location.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, Location, RevenueRecord, Project from app.services.leak_detectors.multi_location_rollup import ( MultiLocationRollupDetector, ) class MultiLocationRollupTestCase(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 = MultiLocationRollupDetector() def setUp(self): self._ctx = self.app.app_context() self._ctx.push() db.session.query(RevenueRecord).delete() db.session.query(Project).delete() db.session.query(Location).delete() db.session.query(Company).delete() db.session.commit() self.company = Company( id="test-co", name="Test Multi-Location Co", size="multi", settings_json={"leak_detectors": {}}, ) db.session.add(self.company) db.session.commit() def tearDown(self): db.session.rollback() self._ctx.pop() def _make_location(self, name="Location A", code=None, **kwargs): loc = Location( id=kwargs.pop("id", str(uuid4())), company_id=self.company.id, name=name, code=code or name[:2].upper(), is_active=True, **kwargs, ) db.session.add(loc) db.session.commit() return loc def _make_revenue(self, location, amount, days_ago=10): rr = RevenueRecord( id=str(uuid4()), company_id=self.company.id, location_id=location.id, location_name=location.name, source_service="quickbooks", external_id=f"inv-{uuid4().hex[:8]}", amount=amount, status="completed", transaction_date=datetime.now(timezone.utc) - timedelta(days=days_ago), ) db.session.add(rr) db.session.commit() return rr # -- Location model tests -- def test_create_location(self): loc = self._make_location(name="Downtown", code="DT") saved = Location.query.get(loc.id) self.assertIsNotNone(saved) self.assertEqual(saved.name, "Downtown") self.assertEqual(saved.company_id, self.company.id) self.assertTrue(saved.is_active) def test_location_to_dict(self): loc = self._make_location(name="HQ", code="HQ") d = loc.to_dict() self.assertEqual(d["name"], "HQ") self.assertEqual(d["code"], "HQ") self.assertEqual(d["company_id"], self.company.id) # -- Detector registration -- def test_detector_has_required_attrs(self): self.assertEqual(self.detector.id, "multi_location_rollup") self.assertTrue(self.detector.name) self.assertTrue(self.detector.enabled_by_default) self.assertIn("concentration_threshold", self.detector.default_params) def test_detector_in_registry(self): from app.services.leak_detectors import registry, DETECTORS self.assertIn(MultiLocationRollupDetector, DETECTORS) registered = registry.get("multi_location_rollup") self.assertIsNotNone(registered) # -- Location resolution -- def test_no_locations_returns_empty(self): locations = self.detector._get_locations(self.company.id) self.assertEqual(locations, []) def test_formal_locations_resolved(self): loc_a = self._make_location(name="A") loc_b = self._make_location(name="B") locations = self.detector._get_locations(self.company.id) ids = {loc.id for loc in locations} self.assertIn(loc_a.id, ids) self.assertIn(loc_b.id, ids) def test_denormalized_locations_resolved(self): loc_id_a = str(uuid4()) loc_id_b = str(uuid4()) for loc_id, name in [(loc_id_a, "Denorm A"), (loc_id_b, "Denorm B")]: rr = RevenueRecord( id=str(uuid4()), company_id=self.company.id, location_id=loc_id, location_name=name, source_service="quickbooks", external_id=f"inv-{uuid4().hex[:8]}", amount=1000.0, status="completed", transaction_date=datetime.now(timezone.utc) - timedelta(days=10), ) db.session.add(rr) db.session.commit() locations = self.detector._get_locations(self.company.id) ids = {loc.id for loc in locations} self.assertIn(loc_id_a, ids) self.assertIn(loc_id_b, ids) # -- Metric computation -- def test_computes_per_location_revenue(self): loc_a = self._make_location(name="A") loc_b = self._make_location(name="B") self._make_revenue(loc_a, 5000.0, days_ago=10) self._make_revenue(loc_b, 3000.0, days_ago=5) metrics = self.detector._compute_location_metrics(self.company.id, [loc_a, loc_b]) by_id = {m["location_id"]: m for m in metrics} self.assertEqual(by_id[loc_a.id]["total_revenue"], 5000.0) self.assertEqual(by_id[loc_b.id]["total_revenue"], 3000.0) def test_old_revenue_outside_lookback(self): loc = self._make_location(name="Old") self._make_revenue(loc, 10000.0, days_ago=200) metrics = self.detector._compute_location_metrics(self.company.id, [loc]) self.assertEqual(metrics[0]["total_revenue"], 0.0) # -- Concentration check -- def test_no_concentration_when_equal(self): loc_a = self._make_location(name="A") loc_b = self._make_location(name="B") self._make_revenue(loc_a, 5000.0) self._make_revenue(loc_b, 5000.0) db.session.commit() locations = self.detector._get_locations(self.company.id) metrics = self.detector._compute_location_metrics(self.company.id, locations) findings = self.detector._check_concentration(metrics) self.assertEqual(findings, []) def test_concentration_flagged(self): loc_a = self._make_location(name="Big") loc_b = self._make_location(name="Small") self._make_revenue(loc_a, 9000.0) self._make_revenue(loc_b, 1000.0) db.session.commit() locations = self.detector._get_locations(self.company.id) metrics = self.detector._compute_location_metrics(self.company.id, locations) findings = self.detector._check_concentration(metrics) self.assertEqual(len(findings), 1) self.assertEqual(findings[0]["metadata_json"]["location_id"], loc_a.id) self.assertAlmostEqual(findings[0]["metadata_json"]["revenue_share"], 0.9) def test_severity_critical_above_75pct(self): loc_a = self._make_location(name="Dominant") loc_b = self._make_location(name="Tiny") self._make_revenue(loc_a, 9500.0) self._make_revenue(loc_b, 500.0) db.session.commit() locations = self.detector._get_locations(self.company.id) metrics = self.detector._compute_location_metrics(self.company.id, locations) findings = self.detector._check_concentration(metrics) self.assertEqual(findings[0]["severity"], "critical") # -- Underperformer check -- def test_no_underperformer_when_equal(self): loc_a = self._make_location(name="A") loc_b = self._make_location(name="B") self._make_revenue(loc_a, 5000.0) self._make_revenue(loc_b, 5000.0) db.session.commit() locations = self.detector._get_locations(self.company.id) metrics = self.detector._compute_location_metrics(self.company.id, locations) findings = self.detector._check_underperformers(metrics, self.company) self.assertEqual(findings, []) def test_underperformer_flagged(self): loc_a = self._make_location(name="Good") loc_b = self._make_location(name="Bad") self._make_revenue(loc_a, 10000.0) self._make_revenue(loc_b, 1000.0) db.session.commit() locations = self.detector._get_locations(self.company.id) metrics = self.detector._compute_location_metrics(self.company.id, locations) findings = self.detector._check_underperformers(metrics, self.company) self.assertEqual(len(findings), 1) self.assertEqual(findings[0]["metadata_json"]["location_id"], loc_b.id) # -- Variance check -- def test_low_variance_no_finding(self): loc_a = self._make_location(name="A") loc_b = self._make_location(name="B") self._make_revenue(loc_a, 5000.0) self._make_revenue(loc_b, 4500.0) db.session.commit() locations = self.detector._get_locations(self.company.id) metrics = self.detector._compute_location_metrics(self.company.id, locations) findings = self.detector._check_variance(metrics) self.assertEqual(findings, []) def test_high_variance_flagged(self): loc_a = self._make_location(name="A") loc_b = self._make_location(name="B") self._make_revenue(loc_a, 10000.0) self._make_revenue(loc_b, 3000.0) db.session.commit() locations = self.detector._get_locations(self.company.id) metrics = self.detector._compute_location_metrics(self.company.id, locations) findings = self.detector._check_variance(metrics) self.assertEqual(len(findings), 1) self.assertIn(findings[0]["severity"], ("medium", "high", "critical")) # -- End-to-end check() tests -- def test_single_location_no_findings(self): loc = self._make_location(name="Only") self._make_revenue(loc, 10000.0) candidates = self.detector.check(self.company.id) self.assertEqual(candidates, []) def test_concentration_returns_candidates(self): loc_a = self._make_location(name="Big") loc_b = self._make_location(name="Small") self._make_revenue(loc_a, 9000.0) self._make_revenue(loc_b, 1000.0) db.session.commit() candidates = self.detector.check(self.company.id) self.assertGreater(len(candidates), 0) self.assertTrue(all(c.detector_id == "multi_location_rollup" for c in candidates)) def test_candidates_have_dedupe_keys(self): loc_a = self._make_location(name="A") loc_b = self._make_location(name="B") self._make_revenue(loc_a, 9000.0) self._make_revenue(loc_b, 1000.0) db.session.commit() candidates = self.detector.check(self.company.id) for c in candidates: self.assertIn("dedupe_key", c.metadata_json) self.assertTrue(c.dedupe_key()) # -- Rollup summary -- def test_rollup_summary_structure(self): loc_a = self._make_location(name="A") loc_b = self._make_location(name="B") self._make_revenue(loc_a, 5000.0) self._make_revenue(loc_b, 3000.0) db.session.commit() self.detector.default_params["lookback_days"] = 90 summary = self.detector.get_rollup_summary(self.company.id) self.assertEqual(summary["company_id"], self.company.id) self.assertEqual(summary["num_locations"], 2) self.assertEqual(summary["total_revenue"], 8000.0) self.assertEqual(len(summary["location_metrics"]), 2) self.assertIn("findings_count", summary) def test_rollup_summary_no_company(self): summary = self.detector.get_rollup_summary("nonexistent") self.assertIn("error", summary) if __name__ == '__main__': unittest.main()