"""Lead Generation engine tests.
Tests against the actual model schemas and API routes for Phase 1 lead gen.
"""
import os
import sys
import unittest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
os.environ.setdefault('DATABASE_URL', 'sqlite:///test_lead_gen.db')
os.environ.setdefault('SECRET_KEY', 'test-secret-key-lead-gen')
from app import create_app
from app.models import db, Company, User, AdTemplate, AdKeyword, AdCreative, \
AdCampaign, AdMetric, OptimizationRule, OptimizationLog, LeadAttribution, UserCompany
class LeadGenTest(unittest.TestCase):
"""Lead gen engine model and API tests"""
@classmethod
def setUpClass(cls):
cls.app = create_app()
cls.app.config['TESTING'] = True
cls.app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test_lead_gen.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._setup_auth()
def tearDown(self):
with self.app.app_context():
db.session.rollback()
LeadAttribution.query.filter_by(company_id=self.company_id).delete()
OptimizationLog.query.filter_by(company_id=self.company_id).delete()
OptimizationRule.query.filter_by(company_id=self.company_id).delete()
AdMetric.query.filter_by(company_id=self.company_id).delete()
AdCreative.query.filter_by(company_id=self.company_id).delete()
AdKeyword.query.filter_by(company_id=self.company_id).delete()
AdCampaign.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():
Company.query.filter_by(id=self.company_id).delete() if hasattr(self, 'company_id') else None
User.query.filter_by(id=self.user_id).delete() if hasattr(self, 'user_id') else None
db.session.commit()
user = User(
id='test-user-lg',
email='test@leadgen.com',
role='partner',
full_name='Test User',
)
user.set_password('test_password')
company = Company(
id='test-company-lg',
name='Test Lead Gen Company',
)
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 {self.user_id}:{self.company_id}',
}
# =========================================================================
# AdTemplate tests (global - no company_id)
# =========================================================================
def test_01_create_template(self):
with self.app.app_context():
template = AdTemplate(
name='Emergency Roof Repair',
slug='roofing-emergency-repair-test',
vertical='roofing',
budget_recommendation=2000.0,
target_cpa=75.0,
structure_json={
"ad_groups": [
{
"name": "Emergency Repair",
"keywords": [
{"text": "emergency roof repair", "match_type": "PHRASE"},
],
}
]
},
is_active=True,
)
db.session.add(template)
db.session.commit()
self.assertIsNotNone(template.id)
self.assertEqual(template.vertical, 'roofing')
def test_02_template_query_by_vertical(self):
with self.app.app_context():
# Clean up any existing HVAC templates first (global table)
AdTemplate.query.filter_by(vertical='hvac').delete()
db.session.commit()
# Create a couple templates
t1 = AdTemplate(
name='AC Install',
slug='hvac-ac-install-test',
vertical='hvac',
is_active=True,
)
t2 = AdTemplate(
name='Furnace Repair',
slug='hvac-furnace-repair-test',
vertical='hvac',
is_active=True,
)
db.session.add_all([t1, t2])
db.session.commit()
hvac_templates = AdTemplate.query.filter_by(vertical='hvac').all()
self.assertEqual(len(hvac_templates), 2)
# =========================================================================
# AdKeyword tests
# =========================================================================
def test_03_create_keyword(self):
with self.app.app_context():
kw = AdKeyword(
company_id=self.company_id,
source_service='google_ads',
external_campaign_id='campaign-test-001',
text="emergency roof repair",
match_type="phrase",
max_cpc=3.50,
target_cpa=75.0,
status='enabled',
)
db.session.add(kw)
db.session.commit()
self.assertIsNotNone(kw.id)
self.assertEqual(kw.text, "emergency roof repair")
def test_04_keyword_query_by_campaign(self):
with self.app.app_context():
for i in range(3):
kw = AdKeyword(
company_id=self.company_id,
source_service='google_ads',
external_campaign_id='test-campaign-multi',
text=f"keyword {i}",
match_type="broad",
)
db.session.add(kw)
db.session.commit()
kws = AdKeyword.query.filter_by(
company_id=self.company_id,
external_campaign_id='test-campaign-multi',
).all()
self.assertEqual(len(kws), 3)
# =========================================================================
# AdCreative tests
# =========================================================================
def test_05_create_creative(self):
with self.app.app_context():
creative = AdCreative(
company_id=self.company_id,
vertical='roofing',
service_type='emergency-repair',
ad_type='responsive_search',
headlines=[
"Emergency Roof Repair",
"24/7 Service Available",
],
descriptions=[
"Fast, reliable roof repair. Call now.",
],
is_active=True,
)
db.session.add(creative)
db.session.commit()
self.assertIsNotNone(creative.id)
self.assertEqual(creative.vertical, 'roofing')
# =========================================================================
# OptimizationRule tests (uses params_json, is_active)
# =========================================================================
def test_06_create_optimization_rule(self):
with self.app.app_context():
rule = OptimizationRule(
company_id=self.company_id,
name='Pause if CPA > $150',
rule_type='kill_high_cpa',
target_level='campaign',
params_json={
"cpa_multiplier": 2.0,
"min_spend": 100,
"lookback_days": 14,
"action": "pause",
},
is_active=True,
)
db.session.add(rule)
db.session.commit()
self.assertIsNotNone(rule.id)
self.assertTrue(rule.is_active)
def test_07_optimization_rule_query(self):
with self.app.app_context():
for i, rtype in enumerate(['kill_high_cpa', 'scale_low_cpa', 'pause_zero_conv']):
rule = OptimizationRule(
company_id=self.company_id,
name=f'Rule {i}',
rule_type=rtype,
params_json={"action": "pause"},
is_active=True,
)
db.session.add(rule)
db.session.commit()
rules = OptimizationRule.query.filter_by(
company_id=self.company_id,
is_active=True,
).all()
self.assertEqual(len(rules), 3)
def test_08_optimization_rule_multi_tenant(self):
with self.app.app_context():
other = Company(
id='other-company-rule-test',
name='Other Rule Co',
)
db.session.add(other)
db.session.flush()
rule = OptimizationRule(
company_id=other.id,
name='Other Rule',
rule_type='kill_high_cpa',
params_json={"action": "scale"},
is_active=True,
)
db.session.add(rule)
db.session.commit()
our_rules = OptimizationRule.query.filter_by(company_id=self.company_id).all()
for r in our_rules:
self.assertNotEqual(r.company_id, other.id)
OptimizationRule.query.filter_by(company_id=other.id).delete()
Company.query.filter_by(id=other.id).delete()
db.session.commit()
# =========================================================================
# OptimizationLog tests
# =========================================================================
def test_09_create_optimization_log(self):
with self.app.app_context():
log = OptimizationLog(
company_id=self.company_id,
source_service='google_ads',
target_type='campaign',
target_id='campaign-ext-001',
target_name='High CPA Campaign',
action='paused',
reason='Paused due to CPA exceeding threshold',
)
db.session.add(log)
db.session.commit()
self.assertIsNotNone(log.id)
self.assertEqual(log.action, 'paused')
# =========================================================================
# LeadAttribution tests
# =========================================================================
def test_10_record_attribution(self):
with self.app.app_context():
attr = LeadAttribution(
company_id=self.company_id,
external_lead_id='lead-12345',
lead_source='web_form',
utm_source='google',
utm_campaign='Emergency Roof Repair - Q3',
utm_medium='cpc',
utm_term='emergency roof repair',
click_cost=4.50,
attribution_status='lead',
)
db.session.add(attr)
db.session.commit()
self.assertIsNotNone(attr.id)
def test_11_attribution_with_deal(self):
with self.app.app_context():
attr = LeadAttribution(
company_id=self.company_id,
external_lead_id='lead-67890',
lead_source='web_form',
utm_source='google',
utm_campaign='Emergency Roof Repair - Q3',
utm_medium='cpc',
click_cost=5.00,
deal_amount=3500.00,
external_deal_id='deal-001',
roas=700.0,
attribution_status='deal',
)
db.session.add(attr)
db.session.commit()
self.assertEqual(attr.attribution_status, 'deal')
def test_12_attribution_multi_tenant(self):
with self.app.app_context():
other = Company(
id='other-company-attr-test',
name='Other Attr Co',
)
db.session.add(other)
db.session.flush()
attr = LeadAttribution(
company_id=other.id,
external_lead_id='other-lead-1',
utm_source='facebook',
utm_campaign='Other Campaign',
)
db.session.add(attr)
db.session.commit()
our_attrs = LeadAttribution.query.filter_by(company_id=self.company_id).all()
for a in our_attrs:
self.assertNotEqual(a.company_id, other.id)
LeadAttribution.query.filter_by(company_id=other.id).delete()
Company.query.filter_by(id=other.id).delete()
db.session.commit()
# =========================================================================
# Attribution helper tests
# =========================================================================
def test_13_utm_url_generation(self):
from app.services.lead_gen.attribution import AttributionHelper
from unittest.mock import MagicMock
campaign = MagicMock()
campaign.source_service = 'google_ads'
campaign.name = 'Emergency Roof Repair - Q3 2026'
url = AttributionHelper.generate_landing_url(
"https://example.com/landing",
campaign,
keyword="emergency roof repair",
)
self.assertIn("utm_source=google", url)
self.assertIn("utm_campaign=Emergency+Roof+Repair", url)
def test_14_utm_extraction(self):
from app.services.lead_gen.attribution import AttributionHelper
url = (
"https://example.com/landing?"
"utm_source=google&utm_campaign=test&"
"utm_term=roof+repair&utm_medium=cpc"
)
params = AttributionHelper.extract_utm_params(url)
self.assertEqual(params['utm_source'], 'google')
self.assertEqual(params['utm_campaign'], 'test')
self.assertEqual(params['utm_term'], 'roof repair')
# =========================================================================
# Campaign generator tests
# =========================================================================
def test_15_campaign_generator_import(self):
from app.services.lead_gen.campaign_generator import CampaignGenerator
self.assertTrue(hasattr(CampaignGenerator, 'generate_campaign'))
# =========================================================================
# Optimization engine tests
# =========================================================================
def test_16_optimization_engine_import(self):
from app.services.lead_gen.optimization_engine import OptimizationEngine
self.assertTrue(hasattr(OptimizationEngine, 'run_all_active'))
# =========================================================================
# Scheduler tests
# =========================================================================
def test_17_scheduler_has_optimization(self):
from app.scheduler import SyncScheduler
self.assertTrue(hasattr(SyncScheduler, '_schedule_optimization'))
self.assertTrue(hasattr(SyncScheduler, '_run_optimization'))
# =========================================================================
# Auth gating tests
# =========================================================================
def test_18_unauthenticated_access(self):
resp = self.client.get('/api/lead-gen/templates')
self.assertEqual(resp.status_code, 401)
class LeadGenAPITest(unittest.TestCase):
"""Lead gen API endpoint tests"""
@classmethod
def setUpClass(cls):
cls.app = create_app()
cls.app.config['TESTING'] = True
cls.app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test_lead_gen_api.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._setup_auth()
self._login()
def _login(self):
"""Log in via the auth endpoint to establish a Flask-Login session."""
with self.app.app_context():
user = User.query.get(self.user_id)
password = user.password_hash if hasattr(user, 'password_hash') else 'test_password'
# Use Flask-Login directly in test context
with self.client.session_transaction() as sess:
sess['_user_id'] = self.user_id
sess['_fresh'] = True
def _setup_auth(self):
with self.app.app_context():
import uuid
uid = uuid.uuid4().hex[:8]
user = User(
id=f'test-user-api-{uid}',
email=f'api{uid}@test.com',
role='partner',
full_name=f'API Test {uid}',
)
user.set_password('test_password')
company = Company(
id=f'test-company-api-{uid}',
name=f'API Test Company {uid}',
)
db.session.add_all([company, user])
db.session.commit()
# Create company membership so _check_company_access passes
membership = UserCompany(
user_id=user.id,
company_id=company.id,
role='owner',
)
db.session.add(membership)
db.session.commit()
self.company_id = company.id
self.user_id = user.id
def tearDown(self):
with self.app.app_context():
db.session.rollback()
LeadAttribution.query.filter_by(company_id=self.company_id).delete()
OptimizationLog.query.filter_by(company_id=self.company_id).delete()
OptimizationRule.query.filter_by(company_id=self.company_id).delete()
AdMetric.query.filter_by(company_id=self.company_id).delete()
AdCreative.query.filter_by(company_id=self.company_id).delete()
AdKeyword.query.filter_by(company_id=self.company_id).delete()
AdCampaign.query.filter_by(company_id=self.company_id).delete()
UserCompany.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 test_01_api_templates_endpoint(self):
with self.app.app_context():
template = AdTemplate(
name='API Template',
slug='api-template-test',
vertical='roofing',
is_active=True,
)
db.session.add(template)
db.session.commit()
resp = self.client.get('/api/lead-gen/templates')
self.assertEqual(resp.status_code, 200)
def test_02_api_rules_endpoint(self):
resp = self.client.get(
f'/api/company/{self.company_id}/lead-gen/rules',
)
self.assertEqual(resp.status_code, 200)
def test_03_api_campaigns_endpoint(self):
resp = self.client.get(
f'/api/company/{self.company_id}/lead-gen/dashboard',
)
def test_04_api_optimization_logs(self):
resp = self.client.get(
f'/api/company/{self.company_id}/lead-gen/optimization-log',
)
self.assertEqual(resp.status_code, 200)
def test_05_api_attribution(self):
resp = self.client.get(
f'/api/company/{self.company_id}/lead-gen/attribution',
)
self.assertEqual(resp.status_code, 200)
if __name__ == '__main__':
unittest.main()