Janady13/AI_Brain / tests/test_marketing_intelligence.py

1from services.seeder_api.app import ( 2 MarketingQuery, 3 marketing_diagnostics, 4 marketing_intelligence, 5 marketing_processes, 6 marketing_statistical_intelligence, 7) 8from services.seeder_api.intelligence import ( 9 MarketingCampaign, 10 MarketingIntelligenceEngine, 11) 12 13 14def test_marketing_campaign_parsing_handles_standard_format(): 15 raw = ( 16 "Campaign: social ads targeting climate tech founders leveraging video testimonial to drive product demos." 17 ) 18 campaign = MarketingCampaign.parse(raw) 19 assert campaign.channel == "social ads" 20 assert campaign.audience == "climate tech founders" 21 assert campaign.offer == "video testimonial" 22 assert campaign.objective == "drive product demos" 23 24 25def test_marketing_engine_recommendations_use_prompt_keywords(): 26 engine = MarketingIntelligenceEngine.from_file() 27 results = engine.recommend("B2B SaaS founders looking for pipeline velocity", limit=2) 28 assert len(results) == 2 29 assert any("B2B SaaS founders" in campaign.audience for campaign in results) 30 31 32def test_marketing_statistical_engine_enriches_results_with_metrics(): 33 engine = MarketingIntelligenceEngine.from_file() 34 scored = engine.recommend_statistical( 35 "Need ROI positive programs for B2B SaaS founders", limit=2 36 ) 37 assert len(scored) == 2 38 assert all(score.metrics.expected_roi > 0 for score in scored) 39 assert any(score.keyword_coverage["audience"] > 0 for score in scored) 40 41 42def test_marketing_intelligence_endpoint_returns_structured_payload(): 43 payload = marketing_intelligence( 44 MarketingQuery(prompt="Need to accelerate demos for B2B SaaS founders", limit=2) 45 ) 46 assert payload.query == "Need to accelerate demos for B2B SaaS founders" 47 assert len(payload.recommendations) == 2 48 audiences = {entry.audience for entry in payload.recommendations} 49 assert any("B2B SaaS founders" in audience for audience in audiences) 50 assert "accelerate" in payload.insight.lower() 51 52 53def test_marketing_statistical_endpoint_returns_portfolio_metrics(): 54 payload = marketing_statistical_intelligence( 55 MarketingQuery(prompt="Optimize ROI for AI founders", limit=2) 56 ) 57 assert payload.query == "Optimize ROI for AI founders" 58 assert len(payload.recommendations) == 2 59 assert payload.portfolio.expected_roi > 0 60 assert payload.portfolio.confidence > 0 61 assert payload.portfolio.strongest_drivers 62 63 64def test_marketing_engine_diagnostics_detects_stage_and_focus(): 65 engine = MarketingIntelligenceEngine.from_file() 66 diagnostics = engine.diagnose("Need urgent conversion plays for enterprise SaaS founders") 67 assert diagnostics.stage in {"conversion", "consideration"} 68 assert diagnostics.urgency != "standard" 69 assert diagnostics.focus.startswith("enterprise") or "startup" in diagnostics.focus 70 assert diagnostics.recommended_kpis 71 assert diagnostics.keyword_summary["tokens"] 72 73 74def test_marketing_diagnostics_endpoint_returns_structured_analysis(): 75 payload = marketing_diagnostics( 76 MarketingQuery(prompt="Pipeline is stalled for B2B founders, need fast ROI", limit=3) 77 ) 78 assert payload.query == "Pipeline is stalled for B2B founders, need fast ROI" 79 assert payload.stage in {"conversion", "consideration"} 80 assert "pipeline" in payload.recommended_kpis[0].lower() or payload.recommended_kpis 81 assert payload.opportunities 82 assert "Performance plateau" in " ".join(payload.risk_signals) or not payload.risk_signals 83 84 85def test_marketing_engine_expansion_grid_generates_egg_experiments(): 86 engine = MarketingIntelligenceEngine.from_file() 87 scored = engine.recommend_statistical( 88 "Need urgent conversion plays for enterprise SaaS founders", limit=2 89 ) 90 experiments = engine.expansion_grid("Need urgent conversion plays for enterprise SaaS founders", scored) 91 assert experiments 92 first = experiments[0] 93 assert first.primary_metric 94 assert first.playbook 95 assert 0.2 <= first.confidence <= 0.99 96 97 98def test_marketing_engine_google_ads_fallback_without_credentials(monkeypatch): 99 monkeypatch.delenv("GOOGLE_ADS_DEVELOPER_TOKEN", raising=False) 100 monkeypatch.delenv("GOOGLE_ADS_CLIENT_ID", raising=False) 101 monkeypatch.delenv("GOOGLE_ADS_CLIENT_SECRET", raising=False) 102 monkeypatch.delenv("GOOGLE_ADS_REFRESH_TOKEN", raising=False) 103 monkeypatch.delenv("GOOGLE_ADS_CUSTOMER_ID", raising=False) 104 105 engine = MarketingIntelligenceEngine.from_file() 106 scored = engine.recommend_statistical( 107 "Need urgent conversion plays for enterprise SaaS founders", limit=2 108 ) 109 insights = engine.google_ads_enrichment( 110 "Need urgent conversion plays for enterprise SaaS founders", scored, limit=2 111 ) 112 assert len(insights) == 2 113 assert all(entry.source == "simulated" for entry in insights) 114 assert all(entry.clicks > 0 for entry in insights) 115 116 117def test_marketing_processes_endpoint_combines_rag_and_egg(): 118 payload = marketing_processes( 119 MarketingQuery(prompt="Need urgent conversion plays for enterprise SaaS founders", limit=2) 120 ) 121 assert payload.query == "Need urgent conversion plays for enterprise SaaS founders" 122 assert payload.rag.process == "rag" 123 assert len(payload.rag.recommendations) == 2 124 assert payload.egg 125 assert payload.egg[0].playbook 126 assert payload.google_ads 127 assert payload.google_ads[0].source in {"simulated", "google_ads_api"} 128