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