"""Angi Lead Management tests.
Tests webhook processing, lead CRUD, actions tracking, analytics,
and API endpoint authentication/validation.
"""
import os
import sys
import unittest
import json
import time
from datetime import datetime, timezone, timedelta
from unittest.mock import patch, MagicMock
from decimal import Decimal
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
os.environ.setdefault('DATABASE_URL', 'sqlite:///test_angi_leads.db')
os.environ.setdefault('SECRET_KEY', 'test-secret-key')
from app import create_app
from app.models import db, Company, User, AngiLead, AngiLeadAction, CrmContact
class AngiLeadsTest(unittest.TestCase):
"""Angi lead management tests"""
@classmethod
def setUpClass(cls):
cls.app = create_app()
cls.app.config['TESTING'] = True
cls.app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test_angi_leads.db'
with cls.app.app_context():
db.create_all()
@classmethod
def tearDownClass(cls):
with cls.app.app_context():
db.drop_all()
def setUp(self):
self.app = self.__class__.app
self.client = self.app.test_client()
self.company_id = None
self.user_id = None
self.auth_header = None
self._setup_auth()
# Build connector for this test's DB
from app.connectors import build_connector
self.connector = build_connector(
'angi',
self.company_id,
{'spid': '12345', 'webhook_url': 'https://example.com/webhook'},
)
def tearDown(self):
with self.app.app_context():
db.session.rollback()
# Delete test data in correct order (FK constraints)
CrmContact.query.filter_by(company_id=self.company_id).delete()
AngiLeadAction.query.filter_by(company_id=self.company_id).delete()
AngiLead.query.filter_by(company_id=self.company_id).delete()
Company.query.filter_by(id=self.company_id).delete()
User.query.filter_by(id=self.user_id).delete()
db.session.commit()
def _setup_auth(self):
with self.app.app_context():
# Clean up any stale test data from previous runs
Company.query.filter_by(id='test-company-angi').delete()
User.query.filter_by(id='test-user-angi').delete()
db.session.commit()
user = User(
id='test-user-angi',
email='test@angi.com',
role='partner',
full_name='Test User',
)
user.set_password('test_password')
company = Company(id='test-company-angi', name='Angi Test Co.')
db.session.add_all([company, user])
db.session.commit()
self.company_id = company.id
self.user_id = user.id
self.auth_header = {
'Authorization': f'Bearer test-user-angi:test-company-angi',
}
def _create_lead_via_webhook(self, payload, expect_success=True):
with self.app.app_context():
result = self.connector.process_webhook(payload)
if expect_success:
assert result['success'], f"Webhook failed: {result.get('error')}"
return result
def _get_lead(self, lead_id):
with self.app.app_context():
lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id=lead_id,
).first()
return lead
def _get_lead_by_db_id(self, db_id):
with self.app.app_context():
lead = AngiLead.query.get(db_id)
db.session.expunge_all()
return lead
# -- Test data --
def _sample_lead(self, lead_id='LEAD-001', **overrides):
base = {
'lead_id': lead_id,
'first_name': 'John',
'last_name': 'Smith',
'email': 'john@example.com',
'phone': '555-123-4567',
'project_type': 'Roofing',
'description': 'Need roof replacement',
'budget': '$5,000-$10,000',
'timeline': 'Within 3 months',
'address': '123 Main St',
'city': 'Detroit',
'state': 'MI',
'zip': '48201',
'timestamp': (datetime.now(timezone.utc) - timedelta(minutes=30)).isoformat(),
}
base.update(overrides)
return base
# -- Tests --
def test_webhook_creates_lead(self):
lead_data = self._sample_lead()
result = self._create_lead_via_webhook({'lead': lead_data})
self.assertTrue(result['success'])
self.assertEqual(result['lead_id'], 'LEAD-001')
with self.app.app_context():
leads = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='LEAD-001',
).all()
self.assertEqual(len(leads), 1)
lead = leads[0]
self.assertEqual(lead.first_name, 'John')
self.assertEqual(lead.last_name, 'Smith')
self.assertEqual(lead.project_type, 'Roofing')
self.assertEqual(lead.status, 'new')
def test_webhook_creates_received_action(self):
self._create_lead_via_webhook({'lead': self._sample_lead()})
with self.app.app_context():
actions = AngiLeadAction.query.filter_by(
company_id=self.company_id,
action_type='received',
).all()
self.assertEqual(len(actions), 1)
self.assertEqual(actions[0].action_type, 'received')
def test_webhook_duplicate_idempotent(self):
lead_data = self._sample_lead()
self._create_lead_via_webhook({'lead': lead_data})
self._create_lead_via_webhook({'lead': lead_data})
with self.app.app_context():
leads = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='LEAD-001',
).all()
self.assertEqual(len(leads), 1)
def test_webhook_creates_crm_contact(self):
from app.models import CrmContact
lead_data = self._sample_lead()
self._create_lead_via_webhook({'lead': lead_data})
with self.app.app_context():
contacts = CrmContact.query.filter_by(
company_id=self.company_id,
external_id='angi_LEAD-001',
).all()
self.assertEqual(len(contacts), 1)
self.assertEqual(contacts[0].first_name, 'John')
def test_webhook_invalid_payload(self):
result = self._create_lead_via_webhook({}, expect_success=False)
self.assertFalse(result['success'])
self.assertIn('error', result)
def test_webhook_missing_lead_id(self):
lead_data = self._sample_lead(lead_id='')
result = self._create_lead_via_webhook({'lead': lead_data}, expect_success=False)
self.assertFalse(result['success'])
def test_webhook_budget_premium_priority(self):
lead_data = self._sample_lead(
lead_id='PREMIUM-001',
is_premium=True,
budget='$50,000',
)
self._create_lead_via_webhook({'lead': lead_data})
with self.app.app_context():
lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='PREMIUM-001',
).first()
self.assertEqual(lead.priority, 'high')
self.assertTrue(lead.is_premium)
self.assertGreaterEqual(lead.budget_value, 50000)
def test_webhook_budget_low_priority(self):
lead_data = self._sample_lead(
lead_id='LOW-001',
budget='$500',
)
self._create_lead_via_webhook({'lead': lead_data})
with self.app.app_context():
lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='LOW-001',
).first()
self.assertEqual(lead.priority, 'medium')
def test_webhook_timestamp_parsed(self):
ts = '2025-06-15T14:30:00Z'
lead_data = self._sample_lead(timestamp=ts)
self._create_lead_via_webhook({'lead': lead_data})
with self.app.app_context():
lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='LEAD-001',
).first()
self.assertEqual(lead.received_at.hour, 14)
self.assertEqual(lead.received_at.minute, 30)
# -- Respond to lead --
def test_respond_to_lead(self):
self._create_lead_via_webhook({'lead': self._sample_lead()})
lead = self._get_lead('LEAD-001')
self.assertIsNotNone(lead)
with self.app.app_context():
db_lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='LEAD-001',
).first()
result = self.connector.respond_to_lead(
db_lead.id,
'Thanks for your inquiry!',
'test-user-angi',
)
self.assertTrue(result['success'])
self.assertIn('data', result)
with self.app.app_context():
db_lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='LEAD-001',
).first()
self.assertEqual(db_lead.status, 'responded')
self.assertIsNotNone(db_lead.responded_at)
self.assertIsNotNone(db_lead.response_time_minutes)
actions = AngiLeadAction.query.filter_by(
company_id=self.company_id,
action_type='responded',
).all()
self.assertEqual(len(actions), 1)
def test_respond_to_lead_not_found(self):
with self.app.app_context():
result = self.connector.respond_to_lead('nonexistent', 'test')
self.assertFalse(result['success'])
self.assertIn('error', result)
# -- Update lead status --
def test_update_lead_status(self):
self._create_lead_via_webhook({'lead': self._sample_lead()})
with self.app.app_context():
db_lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='LEAD-001',
).first()
result = self.connector.update_lead_status(
db_lead.id,
'accepted',
'Good lead',
'test-user-angi',
)
self.assertTrue(result['success'])
with self.app.app_context():
db_lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='LEAD-001',
).first()
self.assertEqual(db_lead.status, 'accepted')
def test_update_lead_status_invalid(self):
self._create_lead_via_webhook({'lead': self._sample_lead()})
with self.app.app_context():
db_lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='LEAD-001',
).first()
result = self.connector.update_lead_status(
db_lead.id,
'invalid_status',
)
self.assertFalse(result['success'])
self.assertIn('error', result)
def test_update_lead_status_action_logged(self):
self._create_lead_via_webhook({'lead': self._sample_lead()})
with self.app.app_context():
db_lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='LEAD-001',
).first()
self.connector.update_lead_status(
db_lead.id,
'contacted',
performed_by='test-user-angi',
)
with self.app.app_context():
actions = AngiLeadAction.query.filter_by(
company_id=self.company_id,
action_type='status_change',
).all()
self.assertEqual(len(actions), 1)
self.assertIn('new', actions[0].details)
self.assertIn('contacted', actions[0].details)
# -- Accept / Reject --
def test_accept_lead(self):
self._create_lead_via_webhook({'lead': self._sample_lead()})
with self.app.app_context():
db_lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='LEAD-001',
).first()
result = self.connector.accept_lead(db_lead.id, 'test-user-angi')
self.assertTrue(result['success'])
with self.app.app_context():
db_lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='LEAD-001',
).first()
self.assertEqual(db_lead.status, 'accepted')
def test_reject_lead(self):
self._create_lead_via_webhook({'lead': self._sample_lead()})
with self.app.app_context():
db_lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='LEAD-001',
).first()
result = self.connector.reject_lead(
db_lead.id,
'Out of area',
'test-user-angi',
)
self.assertTrue(result['success'])
with self.app.app_context():
db_lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='LEAD-001',
).first()
self.assertEqual(db_lead.status, 'rejected')
# -- Analytics --
def test_lead_analytics(self):
for i in range(5):
self._create_lead_via_webhook({'lead': self._sample_lead(f'ANALYTICS-{i}')})
# Respond to 3 of them
with self.app.app_context():
leads = AngiLead.query.filter_by(company_id=self.company_id).all()
for lead in leads[:3]:
self.connector.respond_to_lead(lead.id, 'Response', 'test-user-angi')
with self.app.app_context():
analytics = self.connector.get_lead_analytics(period_days=30)
self.assertEqual(analytics['total_leads'], 5)
self.assertEqual(analytics['leads_responded'], 3)
self.assertEqual(analytics['response_rate'], 60.0)
self.assertEqual(analytics['leads_new'], 2)
self.assertGreater(analytics['avg_response_time_minutes'], 0)
def test_lead_analytics_empty(self):
with self.app.app_context():
analytics = self.connector.get_lead_analytics(period_days=30)
self.assertEqual(analytics['total_leads'], 0)
self.assertEqual(analytics['response_rate'], 0)
# -- API Endpoints --
def test_api_list_leads_no_auth(self):
resp = self.client.get('/api/v1/partner/angi/leads')
self.assertIn(resp.status_code, [401, 403])
def test_api_list_leads(self):
self._create_lead_via_webhook({'lead': self._sample_lead()})
self._create_lead_via_webhook({'lead': self._sample_lead('LEAD-002')})
resp = self.client.get('/api/v1/partner/angi/leads', headers=self.auth_header)
self.assertEqual(resp.status_code, 200)
data = json.loads(resp.data)
self.assertEqual(data['total'], 2)
self.assertEqual(len(data['leads']), 2)
def test_api_list_leads_filter(self):
self._create_lead_via_webhook({'lead': self._sample_lead()})
with self.app.app_context():
db_lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='LEAD-001',
).first()
self.connector.update_lead_status(db_lead.id, 'accepted')
resp = self.client.get(
'/api/v1/partner/angi/leads?status=accepted',
headers=self.auth_header,
)
self.assertEqual(resp.status_code, 200)
data = json.loads(resp.data)
self.assertEqual(data['total'], 1)
def test_api_get_lead(self):
self._create_lead_via_webhook({'lead': self._sample_lead()})
with self.app.app_context():
db_lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='LEAD-001',
).first()
lead_id = db_lead.id
resp = self.client.get(f'/api/v1/partner/angi/leads/{lead_id}', headers=self.auth_header)
self.assertEqual(resp.status_code, 200)
data = json.loads(resp.data)
self.assertEqual(data['data']['first_name'], 'John')
def test_api_get_lead_not_found(self):
resp = self.client.get('/api/v1/partner/angi/leads/nonexistent', headers=self.auth_header)
self.assertIn(resp.status_code, [404, 400])
def test_api_lead_actions(self):
self._create_lead_via_webhook({'lead': self._sample_lead()})
with self.app.app_context():
db_lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='LEAD-001',
).first()
self.connector.respond_to_lead(db_lead.id, 'Response', 'test-user-angi')
resp = self.client.get(f'/api/v1/partner/angi/leads/{db_lead.id}/actions', headers=self.auth_header)
self.assertEqual(resp.status_code, 200)
data = json.loads(resp.data)
self.assertGreaterEqual(len(data['actions']), 2) # received + responded
def test_api_analytics(self):
self._create_lead_via_webhook({'lead': self._sample_lead()})
resp = self.client.get('/api/v1/partner/angi/analytics', headers=self.auth_header)
self.assertEqual(resp.status_code, 200)
data = json.loads(resp.data)
self.assertIn('total_leads', data['data'])
self.assertIn('response_rate', data['data'])
# -- Connector status --
def test_connector_status(self):
status = self.connector.status()
self.assertEqual(status['service'], 'angi')
self.assertIn('spid', status)
def test_connector_connect(self):
result = self.connector.connect()
self.assertEqual(result['status'], 'connected')
self.assertEqual(result['spid'], '12345')
def test_connector_disconnect(self):
self.connector.connect()
result = self.connector.disconnect()
self.assertEqual(result['status'], 'disconnected')
# -- Budget parsing edge cases --
def test_budget_single_value(self):
lead_data = self._sample_lead(lead_id='BUDGET-SINGLE', budget='$5,000')
self._create_lead_via_webhook({'lead': lead_data})
with self.app.app_context():
lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='BUDGET-SINGLE',
).first()
self.assertEqual(lead.budget_value, 5000)
def test_budget_range(self):
lead_data = self._sample_lead(lead_id='BUDGET-RANGE', budget='$5,000-$10,000')
self._create_lead_via_webhook({'lead': lead_data})
with self.app.app_context():
lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='BUDGET-RANGE',
).first()
self.assertEqual(lead.budget_value, 10000)
def test_budget_no_money(self):
lead_data = self._sample_lead(lead_id='BUDGET-NONE', budget='Not sure yet')
self._create_lead_via_webhook({'lead': lead_data})
with self.app.app_context():
lead = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id='BUDGET-NONE',
).first()
self.assertIsNone(lead.budget_value)
if __name__ == '__main__':
unittest.main()