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