"""
Tests for SMS + Call connector.
Tests cover:
- Outbound SMS with opt-out suppression
- Inbound SMS with keyword detection (STOP, YES, CALL ME)
- Delivery status updates
- Outbound call triggering
- Call disposition
- Timeline aggregation
- Speed-to-lead metrics
- Opt-out management
"""
import json
import unittest
from datetime import datetime, timezone, timedelta
from unittest.mock import patch, MagicMock
from app import create_app, db
from app.connectors.sms import SmsConnector
from app.models import (
User,
UserCompany,
Company,
AngiLead,
SmsMessage,
Call,
LeadTouchpoint,
OptOut,
Connector,
)
class SMSCallTestCase(unittest.TestCase):
"""Test suite for SMS + Call connector functionality."""
def setUp(self):
"""Set up test fixtures."""
self.app = create_app()
self.app.config['TESTING'] = True
self.client = self.app.test_client()
with self.app.app_context():
db.create_all()
# Create company
self.company = Company(
id='test-company-1',
name='Test Roofing Co',
industry='roofing',
size='single',
)
db.session.add(self.company)
# Create user
self.user = User(
id='test-user-1',
email='test@test.com',
password_hash='hashed_password',
role='user',
)
db.session.add(self.user)
# Link user to company via UserCompany
self.uc = UserCompany(
user_id='test-user-1',
company_id='test-company-1',
role='owner',
)
db.session.add(self.uc)
# Create connector record
self.connector = Connector(
id='test-connector-1',
company_id='test-company-1',
service='sms',
config_json={
'provider': 'twilio',
'account_sid': 'AC_test',
'auth_token': 'test_auth',
'from_phone': '+15551000000',
},
status='connected',
)
db.session.add(self.connector)
# Create lead — real digits so normalize_phone() resolves lead lookups
self.lead = AngiLead(
id='test-lead-1',
company_id='test-company-1',
angi_lead_id='angi-test-001',
connector_id='test-connector-1',
first_name='John',
last_name='Doe',
phone='+15551234567',
email='john@example.com',
status='new',
lead_source='angi',
)
self.lead.created_at = datetime.now(timezone.utc) - timedelta(minutes=30)
db.session.add(self.lead)
db.session.commit()
def tearDown(self):
"""Clean up test database."""
with self.app.app_context():
db.session.remove()
db.drop_all()
db.create_all()
def _make_connector(self):
"""Create a concrete SmsConnector with a mocked TwilioProvider."""
from app.connectors.sms import SmsConnector, TwilioProvider
connector = SmsConnector(
connector_id='test-connector-1',
settings={
'company_id': 'test-company-1',
'provider': 'twilio',
'account_sid': 'AC_test',
'auth_token': 'test_auth',
'from_phone': '+15551000000',
},
)
# Replace the real provider with a mock so tests don't hit the API
connector.provider = MagicMock(spec=TwilioProvider)
return connector
# ---------------------------------------------------------------------
# 1. Connector registration
# ---------------------------------------------------------------------
def test_connector_registration(self):
"""Test that SmsConnector is registered in the connector registry."""
from app.connectors import _REGISTRY
self.assertIn('sms', _REGISTRY)
# ---------------------------------------------------------------------
# 2. Outbound SMS
# ---------------------------------------------------------------------
@patch.object(SmsConnector, 'is_opted_out', return_value=False)
def test_send_sms_creates_message_and_touchpoint(self, mock_is_opted_out):
"""Test that outbound SMS creates SmsMessage and LeadTouchpoint."""
connector = self._make_connector()
connector.provider.send_sms.return_value = {
'sid': 'SM_test_sid',
'status': 'queued',
'direction': 'outbound-api',
'from_phone': '+15551000000',
'to_phone': '+15551234567',
'body': 'Hi John, thanks for your inquiry!',
'price': '-0.0075',
}
with self.app.app_context():
result = connector.send_sms(
company_id='test-company-1',
to_phone='+15551234567',
body='Hi John, thanks for your inquiry!',
lead_id='test-lead-1',
)
self.assertEqual(result['provider_sid'], 'SM_test_sid')
self.assertEqual(result['status'], 'queued')
# Check SmsMessage was created
msg = SmsMessage.query.filter_by(id=result['sms_id']).first()
self.assertIsNotNone(msg)
self.assertEqual(msg.to_phone, '+15551234567')
self.assertEqual(msg.direction, 'outbound')
self.assertEqual(msg.lead_id, 'test-lead-1')
# Check LeadTouchpoint was created
tp = LeadTouchpoint.query.filter_by(
lead_id='test-lead-1',
touchpoint_type='sms_sent',
reference_id=result['sms_id'],
).first()
self.assertIsNotNone(tp)
@patch.object(SmsConnector, 'is_opted_out', return_value=True)
def test_send_sms_respects_opt_out(self, mock_is_opted_out):
"""Test that SMS is blocked if phone is opted out."""
connector = self._make_connector()
with self.app.app_context():
# Should raise ConnectorError (suppressed)
from app.connectors import ConnectorError
with self.assertRaises(ConnectorError) as ctx:
connector.send_sms(
company_id='test-company-1',
to_phone='+15551999999',
body='You should not receive this',
)
self.assertIn('suppression', str(ctx.exception).lower())
connector.provider.send_sms.assert_not_called()
# ---------------------------------------------------------------------
# 3. Inbound SMS + keyword detection
# ---------------------------------------------------------------------
def test_inbound_sms_regular_reply(self):
"""Test that regular inbound SMS creates message and touchpoint."""
connector = self._make_connector()
with self.app.app_context():
result = connector.handle_sms_webhook(
company_id='test-company-1',
data={
'MessageSid': 'SM_inbound_001',
'From': '+15551234567',
'To': '+15551000000',
'Body': 'Sounds good!',
'Direction': 'inbound',
},
)
self.assertEqual(result['action'], 'reply')
# Check message
msg = SmsMessage.query.filter_by(
provider_message_sid='SM_inbound_001',
direction='inbound',
).first()
self.assertIsNotNone(msg)
# Check touchpoint
tp = LeadTouchpoint.query.filter_by(
lead_id='test-lead-1',
touchpoint_type='sms_received',
reference_id=msg.id,
).first()
self.assertIsNotNone(tp)
def test_inbound_sms_stop_keyword_creates_opt_out(self):
"""Test that STOP keyword creates opt-out record."""
connector = self._make_connector()
with self.app.app_context():
result = connector.handle_sms_webhook(
company_id='test-company-1',
data={
'MessageSid': 'SM_inbound_002',
'From': '+15551234567',
'To': '+15551000000',
'Body': 'STOP',
'Direction': 'inbound',
},
)
self.assertEqual(result['action'], 'opt_out')
self.assertEqual(result['keyword'], 'STOP')
# Check opt-out was created (phone normalized to digits only)
from app.utils.sms_keywords import normalize_phone
opt_out = OptOut.query.filter_by(
company_id='test-company-1',
phone=normalize_phone('+15551234567'),
).first()
self.assertIsNotNone(opt_out)
self.assertEqual(opt_out.reason, 'stop')
def test_inbound_sms_call_me_triggers_call(self):
"""Test that CALL ME keyword triggers outbound call."""
connector = self._make_connector()
connector.provider.make_call.return_value = {
'sid': 'CA_test_sid',
'status': 'queued',
'from_phone': '+15551000000',
'to_phone': '+15551234567',
}
with self.app.app_context():
result = connector.handle_sms_webhook(
company_id='test-company-1',
data={
'MessageSid': 'SM_inbound_003',
'From': '+15551234567',
'To': '+15551000000',
'Body': 'CALL ME',
'Direction': 'inbound',
},
)
self.assertEqual(result['action'], 'call_triggered')
self.assertEqual(result['keyword'], 'CALL_ME')
# Check call was triggered
connector.provider.make_call.assert_called_once()
# Check Call record was created
call = Call.query.filter_by(
lead_id='test-lead-1',
to_phone='+15551234567',
direction='outbound',
).first()
self.assertIsNotNone(call)
# ---------------------------------------------------------------------
# 4. Delivery status updates
# ---------------------------------------------------------------------
@patch.object(SmsConnector, 'is_opted_out', return_value=False)
def test_delivery_status_update(self, mock_is_opted_out):
"""Test delivery status webhook handler."""
connector = self._make_connector()
with self.app.app_context():
# First create an outbound message
connector.provider.send_sms.return_value = {
'sid': 'SM_test_sid',
'status': 'queued',
'direction': 'outbound-api',
'from_phone': '+15551000000',
'to_phone': '+15551234567',
'body': 'Test',
'price': '-0.0075',
}
connector.send_sms(
company_id='test-company-1',
to_phone='+15551234567',
body='Test',
lead_id='test-lead-1',
)
# Simulate delivery status webhook
result = connector.handle_sms_webhook(
company_id='test-company-1',
data={
'MessageSid': 'SM_test_sid',
'From': '+15551000000',
'To': '+15551234567',
'Body': 'Test',
'MessageStatus': 'delivered',
},
)
self.assertEqual(result['new_status'], 'delivered')
# ---------------------------------------------------------------------
# 5. Opt-out management
# ---------------------------------------------------------------------
def test_opt_out_company_isolation(self):
"""Test that opt-outs are scoped to company."""
with self.app.app_context():
# Create opt-out for test company
opt_out = OptOut(
company_id='test-company-1',
phone='+15551111111',
reason='stop',
)
db.session.add(opt_out)
db.session.commit()
# Check it's there for test company
opt_outs = OptOut.query.filter_by(
company_id='test-company-1'
).all()
self.assertEqual(len(opt_outs), 1)
def test_opt_out_unique_constraint(self):
"""Test duplicate opt-out prevention."""
with self.app.app_context():
phone = '+15551222222'
# First opt-out
existing = OptOut.query.filter_by(
company_id='test-company-1',
phone=phone,
).first()
self.assertIsNone(existing)
opt_out = OptOut(
company_id='test-company-1',
phone=phone,
reason='stop',
)
db.session.add(opt_out)
db.session.commit()
# Second attempt - should still only be one
opt_outs = OptOut.query.filter_by(
company_id='test-company-1',
phone=phone,
).all()
self.assertEqual(len(opt_outs), 1)
# ---------------------------------------------------------------------
# 6. Timeline aggregation
# ---------------------------------------------------------------------
def test_lead_timeline_aggregates_events(self):
"""Test that timeline shows all touchpoints chronologically."""
with self.app.app_context():
now = datetime.now(timezone.utc)
# Create touchpoints
for hours_ago, tp_type in [
(3, 'sms_sent'),
(2, 'sms_received'),
(1, 'call_outbound'),
(0, 'call_connected'),
]:
tp = LeadTouchpoint(
company_id='test-company-1',
lead_id='test-lead-1',
touchpoint_type=tp_type,
direction='outbound' if 'sent' in tp_type or 'call_' in tp_type else 'inbound',
content=f'Test {tp_type}',
created_at=now - timedelta(hours=hours_ago),
)
db.session.add(tp)
db.session.commit()
# Query timeline
timeline = LeadTouchpoint.query.filter_by(
company_id='test-company-1',
lead_id='test-lead-1',
).order_by(LeadTouchpoint.created_at).all()
self.assertEqual(len(timeline), 4)
self.assertEqual(timeline[0].touchpoint_type, 'sms_sent')
self.assertEqual(timeline[-1].touchpoint_type, 'call_connected')
# ---------------------------------------------------------------------
# 7. Speed-to-lead metrics
# ---------------------------------------------------------------------
def test_speed_to_lead_metrics(self):
"""Test time-to-first-contact calculation."""
with self.app.app_context():
now = datetime.now(timezone.utc)
self.lead.created_at = now - timedelta(minutes=45)
self.lead.received_at = now - timedelta(minutes=45)
db.session.commit()
# First SMS sent at 15 min
tp_sms = LeadTouchpoint(
company_id='test-company-1',
lead_id='test-lead-1',
touchpoint_type='sms_sent',
direction='outbound',
created_at=now - timedelta(minutes=15),
)
db.session.add(tp_sms)
self.lead.first_contact_at = now - timedelta(minutes=15)
db.session.commit()
# Calculate time to first contact
first_contact = LeadTouchpoint.query.filter_by(
lead_id='test-lead-1',
direction='outbound',
).order_by(LeadTouchpoint.created_at).first()
self.assertIsNotNone(first_contact)
lead_created = self.lead.created_at or datetime.utcnow() - timedelta(minutes=45)
first_created = first_contact.created_at
# Handle naive/aware comparison
if first_created.tzinfo and not lead_created.tzinfo:
lead_created = lead_created.replace(tzinfo=timezone.utc)
elif lead_created.tzinfo and not first_created.tzinfo:
first_created = first_created.replace(tzinfo=timezone.utc)
time_to_contact = (first_created - lead_created).total_seconds() / 60
self.assertEqual(time_to_contact, 30)
if __name__ == '__main__':
unittest.main()