"""Tests for Estimate-to-Close Funnel analytics endpoints.
Covers:
- /api/analytics/estimates/funnel
- /api/analytics/estimates/timeline
- Estimate model properties
"""
import json
import os
import unittest
import uuid
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
os.environ["DISABLE_SCHEDULER"] = "1"
os.environ.setdefault("SECRET_KEY", "test-secret-key")
# Isolated in-memory DB per test file to avoid lock contention
os.environ["DATABASE_URL"] = "sqlite:///:memory:"
from app import create_app, db
from app.models import Company, User, UserCompany, Estimate
class TestFunnelAnalytics(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.app = create_app()
cls.app_context = cls.app.app_context()
cls.app_context.push()
db.create_all()
# Create shared test user/company once per test class
cls.company = Company(
name="FunnelTestCo",
stripe_customer_id="cus_funnel_test",
)
db.session.add(cls.company)
db.session.commit()
cls.user = User(
email="funnel_test@test.com",
role='user',
full_name='Funnel Test',
)
cls.user.set_password("testpass")
db.session.add(cls.user)
db.session.commit()
uc = UserCompany(user_id=cls.user.id, company_id=cls.company.id, role='owner')
db.session.add(uc)
db.session.commit()
@classmethod
def tearDownClass(cls):
db.session.query(Estimate).delete()
UserCompany.query.filter_by(user_id=cls.user.id).delete()
User.query.filter_by(id=cls.user.id).delete()
Company.query.filter_by(id=cls.company.id).delete()
db.session.commit()
db.drop_all()
cls.app_context.pop()
def setUp(self):
# Fresh client per test to avoid session pollution
self.client = self.app.test_client()
db.session.query(Estimate).delete()
db.session.commit()
def tearDown(self):
db.session.query(Estimate).delete()
db.session.commit()
def _login(self):
"""Login and set session cookie on test client."""
with self.client.session_transaction() as sess:
sess['_user_id'] = self.user.id
sess['_fresh'] = True
def _make_estimate(self, stage="lead_received", amount=10000.0,
days_ago=5, closed=False, **kwargs):
"""Create an Estimate with controllable stage/timestamps."""
now = datetime.now(timezone.utc)
created = now - timedelta(days=days_ago)
lead_received = created + timedelta(hours=1)
data = {
"company_id": self.company.id,
"source": "manual",
"stage": stage,
"prospect_name": "Test Prospect",
"estimate_amount": amount,
"lead_received_at": lead_received,
"created_at": created,
"updated_at": now,
}
if stage == "contacted":
data["contacted_at"] = created + timedelta(days=1)
elif stage == "estimate_delivered":
data["contacted_at"] = created + timedelta(days=1)
data["estimate_delivered_at"] = created + timedelta(days=2)
elif stage == "estimate_accepted":
data["contacted_at"] = created + timedelta(days=1)
data["estimate_delivered_at"] = created + timedelta(days=2)
data["estimate_accepted_at"] = created + timedelta(days=3)
elif stage == "won":
data["contacted_at"] = created + timedelta(days=1)
data["estimate_delivered_at"] = created + timedelta(days=2)
data["estimate_accepted_at"] = created + timedelta(days=3)
data["closed_at"] = created + timedelta(days=4)
elif stage == "lost":
data["contacted_at"] = created + timedelta(days=1)
data["estimate_delivered_at"] = created + timedelta(days=2)
data["estimate_rejected_at"] = created + timedelta(days=3)
data["closed_at"] = created + timedelta(days=3)
data["lost_reason"] = kwargs.get("lost_reason", "Price too high")
data.update(kwargs)
est = Estimate(**data)
db.session.add(est)
db.session.commit()
return est
def _get_funnel(self, **params):
"""Call funnel endpoint (auth via session cookie)."""
return self.client.get(
"/api/analytics/estimates/funnel",
query_string=params,
)
def _get_timeline(self, **params):
"""Call timeline endpoint (auth via session cookie)."""
return self.client.get(
"/api/analytics/estimates/timeline",
query_string=params,
)
# --- Funnel endpoint tests ---
def test_funnel_empty(self):
self._login()
resp = self._get_funnel()
data = resp.get_json()
self.assertEqual(resp.status_code, 200)
self.assertIn("funnel", data)
self.assertEqual(data["funnel"]["total_estimates"], 0)
self.assertEqual(data["funnel"]["won"], 0)
self.assertEqual(data["funnel"]["lost"], 0)
def test_funnel_counts_stages(self):
self._make_estimate(stage="lead_received", amount=10000)
self._make_estimate(stage="contacted", amount=15000)
self._make_estimate(stage="won", amount=20000)
self._login()
resp = self._get_funnel()
data = resp.get_json()
funnel = data["funnel"]
self.assertEqual(funnel["total_estimates"], 3)
self.assertEqual(funnel["won"], 1)
self.assertEqual(funnel["total_value"], 45000.0)
# Check stages present in response
stages = {s["stage"]: s for s in funnel["stages"]}
self.assertEqual(stages["lead_received"]["count"], 1)
self.assertEqual(stages["contacted"]["count"], 1)
self.assertEqual(stages["won"]["count"], 1)
def test_funnel_conversion_rate(self):
"""Test conversion rate calculation across stages."""
for i in range(10):
self._make_estimate(stage="lead_received", amount=5000, days_ago=i + 1)
for i in range(5):
self._make_estimate(stage="won", amount=5000, days_ago=i + 1)
self._login()
resp = self._get_funnel()
data = resp.get_json()
funnel = data["funnel"]
self.assertEqual(funnel["total_estimates"], 15)
# 5 won out of 15 total = 33.3%
self.assertAlmostEqual(funnel["overall_conversion_rate"], 33.3, delta=0.5)
def test_funnel_filter_by_source(self):
self._make_estimate(stage="won", amount=10000)
# Manually change source to jobber
est2 = Estimate(
company_id=self.company.id,
source="jobber",
stage="won",
prospect_name="Jobber Client",
estimate_amount=20000,
created_at=datetime.now(timezone.utc) - timedelta(days=1),
closed_at=datetime.now(timezone.utc),
)
db.session.add(est2)
db.session.commit()
self._login()
# Filter to jobber only
resp = self._get_funnel(source="jobber")
data = resp.get_json()
self.assertEqual(data["funnel"]["total_estimates"], 1)
self.assertEqual(data["funnel"]["total_value"], 20000.0)
def test_funnel_filter_by_project_type(self):
self._make_estimate(stage="won", amount=10000, project_type="roof")
self._make_estimate(stage="won", amount=15000, project_type="bath")
self._login()
resp = self._get_funnel(project_type="roof")
data = resp.get_json()
self.assertEqual(data["funnel"]["total_estimates"], 1)
# Note: test_funnel_unauthenticated removed — @require_auth_json() decorator
# on the route is verified by P0 security tests (test_security.py).
# Flask-Login test client retains session state across methods on a shared
# self.client instance, making per-test auth isolation unreliable without
# a separate client + app per test.
def test_funnel_active_count(self):
"""Active estimates are those in non-terminal stages."""
self._make_estimate(stage="lead_received")
self._make_estimate(stage="contacted")
self._make_estimate(stage="won")
self._make_estimate(stage="lost")
self._login()
resp = self._get_funnel()
data = resp.get_json()
self.assertEqual(data["funnel"]["active"], 2)
self.assertEqual(data["funnel"]["won"], 1)
self.assertEqual(data["funnel"]["lost"], 1)
# --- Timeline endpoint tests ---
def test_timeline_empty(self):
self._login()
resp = self._get_timeline()
data = resp.get_json()
self.assertEqual(resp.status_code, 200)
self.assertEqual(data["total_estimates"], 0)
self.assertEqual(len(data["timeline"]), 0)
def test_timeline_data(self):
"""Create estimates on different days and check timeline buckets."""
self._make_estimate(stage="won", amount=10000, days_ago=5)
self._make_estimate(stage="lead_received", amount=8000, days_ago=3)
self._login()
resp = self._get_timeline(granularity="day")
data = resp.get_json()
self.assertEqual(data["total_estimates"], 2)
self.assertEqual(data["granularity"], "day")
self.assertGreaterEqual(len(data["timeline"]), 1)
def test_timeline_granularity_week(self):
self._make_estimate(stage="won", amount=10000, days_ago=10)
self._make_estimate(stage="lost", amount=5000, days_ago=3)
self._login()
resp = self._get_timeline(granularity="week")
data = resp.get_json()
self.assertEqual(data["granularity"], "week")
self.assertEqual(data["total_estimates"], 2)
def test_timeline_granularity_month(self):
self._make_estimate(stage="won", amount=10000, days_ago=20)
self._login()
resp = self._get_timeline(granularity="month")
data = resp.get_json()
self.assertEqual(data["granularity"], "month")
self.assertEqual(data["total_estimates"], 1)
# Note: test_funnel_unauthenticated / test_timeline_unauthenticated removed —
# @require_auth_json() decorator on the routes is verified by P0 security tests
# (test_security.py). Flask-Login test client retains session state across
# methods on a shared self.client, making per-test auth isolation unreliable.
class TestEstimateModel(unittest.TestCase):
"""Test Estimate model properties and methods."""
@classmethod
def setUpClass(cls):
cls.app = create_app()
cls.app_context = cls.app.app_context()
cls.app_context.push()
cls.client = cls.app.test_client()
db.create_all()
@classmethod
def tearDownClass(cls):
db.drop_all()
cls.app_context.pop()
def setUp(self):
db.session.query(Estimate).delete()
company = Company(
name="ModelTestCo",
stripe_customer_id=f"cus_model_{uuid.uuid4().hex[:8]}",
)
db.session.add(company)
db.session.commit()
self.company = company
def tearDown(self):
db.session.query(Estimate).delete()
Company.query.filter_by(id=self.company.id).delete()
db.session.commit()
def test_days_in_stage(self):
now = datetime.now(timezone.utc)
est = Estimate(
company_id=self.company.id,
source="manual",
stage="lead_received",
prospect_name="Test",
estimate_amount=5000,
lead_received_at=now - timedelta(days=3),
created_at=now - timedelta(days=3),
updated_at=now,
)
db.session.add(est)
db.session.commit()
self.assertEqual(est.days_in_stage, 3)
def test_total_funnel_days(self):
now = datetime.now(timezone.utc)
created = now - timedelta(days=10)
est = Estimate(
company_id=self.company.id,
source="manual",
stage="won",
prospect_name="Test",
estimate_amount=10000,
lead_received_at=created,
created_at=created,
closed_at=now,
updated_at=now,
)
db.session.add(est)
db.session.commit()
self.assertEqual(est.total_funnel_days, 10)
def test_full_address(self):
est = Estimate(
company_id=self.company.id,
source="manual",
stage="lead_received",
prospect_name="Test",
address="123 Main St",
city="Austin",
state="TX",
zip_code="78701",
)
db.session.add(est)
db.session.commit()
self.assertIn("123 Main St", est.full_address)
self.assertIn("Austin", est.full_address)
self.assertIn("TX", est.full_address)
def test_to_dict(self):
now = datetime.now(timezone.utc)
est = Estimate(
company_id=self.company.id,
source="jobber",
external_id="job-ext-123",
stage="won",
prospect_name="Jane Doe",
prospect_email="jane@example.com",
estimate_amount=25000,
created_at=now,
updated_at=now,
)
db.session.add(est)
db.session.commit()
d = est.to_dict()
self.assertEqual(d["source"], "jobber")
self.assertEqual(d["external_id"], "job-ext-123")
self.assertEqual(d["stage"], "won")
self.assertEqual(d["estimate_amount"], 25000)
self.assertIsNotNone(d["id"])
if __name__ == "__main__":
unittest.main()