Source identity
| Field | Recorded value |
|---|---|
| Source type | GitHub source |
| Original record | Janady13/AI_Brain / tests/test_marketing_intelligence.py |
| Date | Not specified in source |
| Repository | Janady13/AI_Brain |
| Revision | 5cd0bbf744b0aee9445913337e852889d9a8098e |
| Branch | codex/build-online-github-ani-7v4kug |
| Path or archive member | tests/test_marketing_intelligence.py |
| Original SHA-256 | 02eef1196ce2bc800769647feafdd3f4717b930e6402b34aba4c96646369c455 |
| Git blob SHA | 03a3709b49b5d09b6dd5c9898204cf29bcd76dde |
| Reading-copy SHA-256 | 02eef1196ce2bc800769647feafdd3f4717b930e6402b34aba4c96646369c455 |
Recovered source transcript
Download complete source text ↓958 lines · 39.3 KB
from services.seeder_api.app import (
MarketingQuery,
marketing_diagnostics,
marketing_intelligence,
marketing_processes,
marketing_statistical_intelligence,
)
import base64
import json
from services.seeder_api.intelligence import (
AmazonMarketplaceProcess,
AWSRevenueProcess,
BudgetAllocatorProcess,
CJAffiliateProcess,
CryptoMarketProcess,
EbayMarketplaceProcess,
GoogleAdsProcess,
MarketingCampaign,
MarketingIntelligenceEngine,
AutomationSprintPlanner,
LearningLoopProcess,
CompetitiveIntelligenceProcess,
ScenarioPlanningProcess,
StripeProcess,
PayPalProcess,
InfluencerCollaborationProcess,
StatisticalReasoningProcess,
AgenticSystemsProcess,
QuantumSimulationProcess,
PhysicsEconomicsProcess,
)
def test_marketing_campaign_parsing_handles_standard_format():
raw = (
"Campaign: social ads targeting climate tech founders leveraging video testimonial to drive product demos."
)
campaign = MarketingCampaign.parse(raw)
assert campaign.channel == "social ads"
assert campaign.audience == "climate tech founders"
assert campaign.offer == "video testimonial"
assert campaign.objective == "drive product demos"
def test_marketing_engine_recommendations_use_prompt_keywords():
engine = MarketingIntelligenceEngine.from_file()
results = engine.recommend("B2B SaaS founders looking for pipeline velocity", limit=2)
assert len(results) == 2
assert any("B2B SaaS founders" in campaign.audience for campaign in results)
def test_marketing_statistical_engine_enriches_results_with_metrics():
engine = MarketingIntelligenceEngine.from_file()
scored = engine.recommend_statistical(
"Need ROI positive programs for B2B SaaS founders", limit=2
)
assert len(scored) == 2
assert all(score.metrics.expected_roi > 0 for score in scored)
assert any(score.keyword_coverage["audience"] > 0 for score in scored)
def test_statistical_reasoning_summary_blends_priors_and_campaigns():
engine = MarketingIntelligenceEngine.from_file()
prompt = "Enterprise pipeline needs statistical ROI certainty"
ranked = engine.recommend_statistical(prompt, limit=2)
stock_markets = engine.stock_market_enrichment(prompt, ranked, limit=2)
crypto_markets = engine.crypto_market_enrichment(prompt, limit=2)
budget_plan = engine.budget_optimizer(prompt, ranked, limit=2)
summary = engine.statistical_reasoning_summary(
prompt,
ranked,
budget=budget_plan,
stock_markets=stock_markets,
crypto_markets=crypto_markets,
limit=2,
)
assert summary.expected_roi > 0
assert summary.success_probability > 0
assert summary.evidence
assert summary.driver_confidence
assert "success" in summary.narrative.lower()
def test_marketing_intelligence_endpoint_returns_structured_payload():
payload = marketing_intelligence(
MarketingQuery(prompt="Need to accelerate demos for B2B SaaS founders", limit=2)
)
assert payload.query == "Need to accelerate demos for B2B SaaS founders"
assert len(payload.recommendations) == 2
audiences = {entry.audience for entry in payload.recommendations}
assert any("B2B SaaS founders" in audience for audience in audiences)
assert "accelerate" in payload.insight.lower()
def test_marketing_statistical_endpoint_returns_portfolio_metrics():
payload = marketing_statistical_intelligence(
MarketingQuery(prompt="Optimize ROI for AI founders", limit=2)
)
assert payload.query == "Optimize ROI for AI founders"
assert len(payload.recommendations) == 2
assert payload.portfolio.expected_roi > 0
assert payload.portfolio.confidence > 0
assert payload.portfolio.strongest_drivers
def test_marketing_engine_diagnostics_detects_stage_and_focus():
engine = MarketingIntelligenceEngine.from_file()
diagnostics = engine.diagnose("Need urgent conversion plays for enterprise SaaS founders")
assert diagnostics.stage in {"conversion", "consideration"}
assert diagnostics.urgency != "standard"
assert diagnostics.focus.startswith("enterprise") or "startup" in diagnostics.focus
assert diagnostics.recommended_kpis
assert diagnostics.keyword_summary["tokens"]
def test_marketing_diagnostics_endpoint_returns_structured_analysis():
payload = marketing_diagnostics(
MarketingQuery(prompt="Pipeline is stalled for B2B founders, need fast ROI", limit=3)
)
assert payload.query == "Pipeline is stalled for B2B founders, need fast ROI"
assert payload.stage in {"conversion", "consideration"}
assert "pipeline" in payload.recommended_kpis[0].lower() or payload.recommended_kpis
assert payload.opportunities
assert "Performance plateau" in " ".join(payload.risk_signals) or not payload.risk_signals
def test_marketing_engine_expansion_grid_generates_egg_experiments():
engine = MarketingIntelligenceEngine.from_file()
scored = engine.recommend_statistical(
"Need urgent conversion plays for enterprise SaaS founders", limit=2
)
experiments = engine.expansion_grid("Need urgent conversion plays for enterprise SaaS founders", scored)
assert experiments
first = experiments[0]
assert first.primary_metric
assert first.playbook
assert 0.2 <= first.confidence <= 0.99
def test_marketing_engine_google_ads_fallback_without_credentials(monkeypatch):
monkeypatch.delenv("GOOGLE_ADS_DEVELOPER_TOKEN", raising=False)
monkeypatch.delenv("GOOGLE_ADS_CLIENT_ID", raising=False)
monkeypatch.delenv("GOOGLE_ADS_CLIENT_SECRET", raising=False)
monkeypatch.delenv("GOOGLE_ADS_REFRESH_TOKEN", raising=False)
monkeypatch.delenv("GOOGLE_ADS_CUSTOMER_ID", raising=False)
engine = MarketingIntelligenceEngine.from_file()
scored = engine.recommend_statistical(
"Need urgent conversion plays for enterprise SaaS founders", limit=2
)
insights = engine.google_ads_enrichment(
"Need urgent conversion plays for enterprise SaaS founders", scored, limit=2
)
assert len(insights) == 2
assert all(entry.source == "simulated" for entry in insights)
assert all(entry.clicks > 0 for entry in insights)
def test_google_ads_process_loads_service_account_credentials(monkeypatch):
info = {
"type": "service_account",
"project_id": "test-project",
"private_key_id": "abc123",
"private_key": "-----BEGIN PRIVATE KEY-----\nabc\n-----END PRIVATE KEY-----\n",
"client_email": "ai-403@encoded-equator-335209.iam.gserviceaccount.com",
"client_id": "109575360154363072675",
"token_uri": "https://oauth2.googleapis.com/token",
}
encoded = base64.b64encode(json.dumps(info).encode("utf-8")).decode("utf-8")
monkeypatch.setenv("GOOGLE_ADS_DEVELOPER_TOKEN", "dev-token")
monkeypatch.setenv("GOOGLE_ADS_CUSTOMER_ID", "1234567890")
monkeypatch.setenv("GOOGLE_ADS_SERVICE_ACCOUNT_B64", encoded)
process = GoogleAdsProcess()
assert process._config["developer_token"] == "dev-token"
assert process._config["customer_id"] == "1234567890"
assert process._config["service_account_info"]["client_email"] == info["client_email"]
assert process.configured is False # without SDK the client remains unconfigured
def test_cj_affiliate_process_catalog_fallback(monkeypatch):
monkeypatch.delenv("CJ_DEVELOPER_KEY", raising=False)
monkeypatch.delenv("CJ_WEBSITE_ID", raising=False)
monkeypatch.delenv("CJ_ACCOUNT_ID", raising=False)
engine = MarketingIntelligenceEngine.from_file()
scored = engine.recommend_statistical(
"Affiliate monetization ideas for AI SaaS founders", limit=3
)
process = CJAffiliateProcess()
recommendations = process.offers(
"Affiliate monetization ideas for AI SaaS founders", scored, limit=2
)
assert recommendations
assert all(entry.source == "cj_catalog" for entry in recommendations)
assert all(entry.projected_monthly_revenue > 0 for entry in recommendations)
def test_cj_affiliate_process_harvests_env_configuration(monkeypatch):
monkeypatch.setenv("CJ_DEVELOPER_KEY", "dev-key")
monkeypatch.setenv("CJ_WEBSITE_ID", "site-123")
process = CJAffiliateProcess()
assert process.configured is True
assert process._config["developer_key"] == "dev-key"
assert process._config["website_id"] == "site-123"
def test_crypto_market_process_fallback_without_requests():
process = CryptoMarketProcess(allow_network=False)
snapshots = process.market_data("Need BTC ETH and SOL price intel", limit=3)
assert len(snapshots) == 3
assert all(snapshot.source in {"simulated", "coingecko", "coincap"} for snapshot in snapshots)
assert any(snapshot.asset == "bitcoin" for snapshot in snapshots)
def test_stripe_process_generates_checkout_plans(monkeypatch):
monkeypatch.delenv("STRIPE_API_KEY", raising=False)
monkeypatch.delenv("STRIPE_ACCOUNT_ID", raising=False)
engine = MarketingIntelligenceEngine.from_file()
scored = engine.recommend_statistical("Launch subscription for enterprise AI", limit=2)
process = StripeProcess()
plans = process.plans("Launch subscription for enterprise AI", scored, limit=2)
assert plans
assert all(plan.source in {"simulated", "stripe_sdk"} for plan in plans)
assert plans[0].billing_interval in {"monthly", "annual", "one_time"}
def test_paypal_process_generates_checkout_plans(monkeypatch):
monkeypatch.delenv("PAYPAL_CLIENT_ID", raising=False)
monkeypatch.delenv("PAYPAL_CLIENT_SECRET", raising=False)
monkeypatch.delenv("PAYPAL_MERCHANT_ID", raising=False)
engine = MarketingIntelligenceEngine.from_file()
scored = engine.recommend_statistical("Need to monetize community via digital products", limit=2)
process = PayPalProcess()
plans = process.plans("Need to monetize community via digital products", scored, limit=2)
assert plans
assert all(plan.source in {"simulated", "paypal_sdk"} for plan in plans)
assert plans[0].checkout_flow in {"subscription", "invoice", "one_time"}
assert plans[0].payout_schedule in {"daily", "weekly", "rolling"}
def test_influencer_process_recommends_partner_reach():
engine = MarketingIntelligenceEngine.from_file()
scored = engine.recommend_statistical("Launch AI automation suite with influencer blitz", limit=3)
process = InfluencerCollaborationProcess()
recommendations = process.recommendations(
"Launch AI automation suite with influencer blitz", scored, limit=2
)
assert recommendations
assert all(rec.estimated_reach > 0 for rec in recommendations)
assert all(rec.projected_roi > 0 for rec in recommendations)
assert {rec.source for rec in recommendations}.issubset(
{"influencer_catalog", "influencer_api"}
)
def test_agentic_systems_process_scores_agent_workflows(monkeypatch):
monkeypatch.delenv("DATABASE_URL", raising=False)
monkeypatch.delenv("FEATURE_STORE_URL", raising=False)
monkeypatch.delenv("VECTOR_DB_URL", raising=False)
engine = MarketingIntelligenceEngine.from_file()
scored = engine.recommend_statistical(
"Need agent workflows orchestrating databases and trading automation", limit=3
)
process = AgenticSystemsProcess()
plans = process.plans(
"Need agent workflows orchestrating databases and trading automation", scored, limit=2
)
assert plans
assert all(plan.system_name for plan in plans)
assert any("database" in " ".join(plan.data_interfaces).lower() for plan in plans)
assert all(plan.source in {"agentic_systems_catalog", "agentic_systems_live"} for plan in plans)
def test_knowledge_workflow_process_returns_rag_and_kag():
engine = MarketingIntelligenceEngine.from_file()
scored = engine.recommend_statistical(
"Need rag pipelines and knowledge graphs for revenue prompts", limit=3
)
plans = engine.knowledge_workflow_enrichment(
"Need rag pipelines and knowledge graphs for revenue prompts", scored, limit=3
)
assert plans
categories = {plan.category for plan in plans}
assert "rag" in categories
assert "kag" in categories
assert all(plan.indexing_plan for plan in plans)
def test_marketing_knowledge_process_surfaces_sync_automation():
engine = MarketingIntelligenceEngine.from_file()
prompt = "Refresh marketing wiki, airtable dashboards, and creative libraries automatically"
scored = engine.recommend_statistical(prompt, limit=3)
plans = engine.marketing_knowledge_enrichment(prompt, scored, limit=3)
assert plans
assert all(plan.ingestion_steps for plan in plans)
assert all(plan.coverage_assets for plan in plans)
assert all(plan.update_frequency for plan in plans)
def test_mixture_of_experts_activation_selects_experts():
engine = MarketingIntelligenceEngine.from_file()
scored = engine.recommend_statistical(
"Blend google ads, stripe pricing, amazon inventory, and CJ affiliates", limit=3
)
moe_plans = engine.mixture_of_experts_activation(
"Blend google ads, stripe pricing, amazon inventory, and CJ affiliates", scored, limit=3
)
assert moe_plans
plan = moe_plans[0]
assert plan.activations
assert any(act.provider in {"google", "stripe", "amazon", "cj"} for act in plan.activations)
assert plan.evaluation_metrics
def test_agentic_loop_blueprint_available():
engine = MarketingIntelligenceEngine.from_file()
scored = engine.recommend_statistical(
"Automate revenue loops with autonomous agents", limit=3
)
loops = engine.agentic_loop_blueprints("Automate revenue loops with autonomous agents", scored)
assert loops
loop = loops[0]
assert loop.maturity in {"foundational", "advanced", "autonomous"}
assert loop.stages
assert loop.stages[0].stage == "perception"
def test_amazon_marketplace_process_offline_catalog(monkeypatch):
monkeypatch.delenv("AMAZON_PAAPI_ACCESS_KEY", raising=False)
monkeypatch.delenv("AMAZON_PAAPI_SECRET_KEY", raising=False)
monkeypatch.delenv("AMAZON_PAAPI_PARTNER_TAG", raising=False)
engine = MarketingIntelligenceEngine.from_file()
scored = engine.recommend_statistical(
"Need retail and aws marketplace offers for ai founders", limit=3
)
process = AmazonMarketplaceProcess()
listings = process.listings(
"Need retail and aws marketplace offers for ai founders", scored, limit=2
)
assert listings
assert all(entry.source in {"amazon_catalog", "amazon_paapi"} for entry in listings)
assert listings[0].projected_monthly_revenue > 0
def test_ebay_marketplace_process_offline_catalog(monkeypatch):
monkeypatch.delenv("EBAY_APP_ID", raising=False)
monkeypatch.delenv("EBAY_CERT_ID", raising=False)
monkeypatch.delenv("EBAY_DEV_ID", raising=False)
engine = MarketingIntelligenceEngine.from_file()
scored = engine.recommend_statistical(
"Need ecommerce partnerships and wearable funnels", limit=3
)
process = EbayMarketplaceProcess()
listings = process.listings(
"Need ecommerce partnerships and wearable funnels", scored, limit=2
)
assert listings
assert all(entry.source in {"ebay_catalog", "ebay_partner_api"} for entry in listings)
assert listings[0].projected_monthly_revenue > 0
assert listings[0].projected_profit_margin > 0
def test_aws_revenue_process_generates_playbooks(monkeypatch):
monkeypatch.delenv("AWS_ACCESS_KEY_ID", raising=False)
monkeypatch.delenv("AWS_SECRET_ACCESS_KEY", raising=False)
monkeypatch.delenv("AWS_SESSION_TOKEN", raising=False)
engine = MarketingIntelligenceEngine.from_file()
scored = engine.recommend_statistical("Scale conversion with aws pinpoint journeys", limit=3)
process = AWSRevenueProcess()
programs = process.plays("Scale conversion with aws pinpoint journeys", scored, limit=2)
assert programs
assert all(entry.source in {"aws_playbooks", "aws_sdk"} for entry in programs)
assert programs[0].projected_roi > 0
def test_budget_allocator_process_generates_plan(monkeypatch):
engine = MarketingIntelligenceEngine.from_file()
scored = engine.recommend_statistical(
"Deploy $45k across enterprise pipeline accelerators", limit=3
)
process = BudgetAllocatorProcess()
plan = process.plan("Deploy $45k across enterprise pipeline accelerators", scored, limit=2)
assert plan.total_budget >= 45000
assert plan.allocations
assert abs(sum(allocation.allocation_usd for allocation in plan.allocations) - plan.total_budget) <= 1.0
assert plan.expected_revenue >= plan.total_budget
assert 0 < plan.expected_roi
def test_automation_sprint_planner_generates_three_hour_schedule():
engine = MarketingIntelligenceEngine.from_file()
prompt = "Need urgent conversion plays for enterprise SaaS founders"
scored = engine.recommend_statistical(prompt, limit=3)
diagnostics = engine.diagnose(prompt)
planner = AutomationSprintPlanner()
sprint = planner.plan(prompt, diagnostics, scored, limit=3)
assert sprint.total_minutes == 180
assert len(sprint.tasks) == sprint.total_minutes // 30
assert sprint.tasks[0].start_minute == 0
assert sprint.tasks[-1].end_minute == sprint.total_minutes
assert "automation" in sprint.summary.lower()
def test_learning_loop_process_surfaces_closed_loop_recommendations():
engine = MarketingIntelligenceEngine.from_file()
prompt = "Need enterprise AI pipeline recovery with stronger ROI"
scored = engine.recommend_statistical(prompt, limit=3)
diagnostics = engine.diagnose(prompt)
process = LearningLoopProcess()
insights = process.recommend(prompt, diagnostics, scored, limit=2)
assert insights
first = insights[0]
assert first.learning
assert first.next_actions
assert first.source == "performance_memory"
def test_marketing_engine_automation_sprint_infers_from_prompt():
engine = MarketingIntelligenceEngine.from_file()
sprint = engine.automation_sprint(
"Need urgent conversion plays for enterprise SaaS founders", limit=2
)
assert sprint.stage in {"conversion", "consideration", "awareness", "retention"}
assert sprint.total_minutes == 180
assert all(task.start_minute < task.end_minute for task in sprint.tasks)
def test_marketing_engine_learning_loop_aligns_with_prompt():
engine = MarketingIntelligenceEngine.from_file()
insights = engine.learning_loop(
"Need enterprise AI pipeline recovery with stronger ROI", limit=2
)
assert insights
assert any(insight.stage.lower() in {"conversion", "consideration"} for insight in insights)
assert all(insight.next_actions for insight in insights)
def test_competitive_process_surfaces_counter_moves():
engine = MarketingIntelligenceEngine.from_file()
prompt = "Need to outmaneuver paid search competitors for AI security buyers"
scored = engine.recommend_statistical(prompt, limit=2)
diagnostics = engine.diagnose(prompt)
process = CompetitiveIntelligenceProcess()
signals = process.signals(prompt, diagnostics, scored, limit=2)
assert signals
assert all(signal.threat_level in {"low", "medium", "high"} for signal in signals)
assert all(0.0 <= signal.focus_alignment <= 1.0 for signal in signals)
def test_language_capability_process_surfaces_conversational_readiness():
engine = MarketingIntelligenceEngine.from_file()
prompt = "Deploy conversational AI with NLP, NLU, and NLG to close inbound leads"
ranked = engine.recommend_statistical(prompt, limit=3)
plans = engine.language_capability_enrichment(prompt, ranked, limit=3)
assert plans
assert {plan.category for plan in plans}.issubset(
{"nlp", "nlu", "nlg", "conversational_ai"}
)
assert any("language" in plan.capability.lower() for plan in plans)
assert plans[0].maturity in {"foundational", "advanced", "autonomous"}
assert plans[0].pipeline_components
def test_code_automation_process_maps_codefix_capabilities():
engine = MarketingIntelligenceEngine.from_file()
prompt = "Use AI CodeFix, intelligent automation, and agentic loops to fix bugs and scale releases"
ranked = engine.recommend_statistical(prompt, limit=3)
plans = engine.code_automation_enrichment(prompt, ranked, limit=2)
assert plans
assert {plan.category for plan in plans}.issubset(
{"ai_codefix", "intelligent_automation", "agentic_automation"}
)
assert any(
any(token in plan.capability.lower() for token in ("code", "automation"))
for plan in plans
)
assert plans[0].automation_stages
assert plans[0].guardrails
assert plans[0].maturity in {"foundational", "advanced", "autonomous", "elite"}
def test_meta_learning_process_surfaces_adaptive_loops():
engine = MarketingIntelligenceEngine.from_file()
prompt = "Continuously learn from campaigns with meta-learning, reinforcement loops, and governance"
ranked = engine.recommend_statistical(prompt, limit=3)
plans = engine.meta_learning_enrichment(prompt, ranked, limit=3)
assert plans
assert any("learn" in plan.capability.lower() for plan in plans)
assert plans[0].learning_modes
assert plans[0].deployment_modes
assert plans[0].maturity in {"foundational", "advanced", "elite", "autonomous"}
def test_payment_integration_process_captures_provider_playbooks():
engine = MarketingIntelligenceEngine.from_file()
prompt = "Stand up Stripe and Cash App pay flows for new offers"
ranked = engine.recommend_statistical(prompt, limit=3)
plans = engine.payment_integration_enrichment(prompt, ranked, limit=3)
assert plans
providers = {plan.provider.lower() for plan in plans}
assert "stripe" in providers
assert any("cash" in provider for provider in providers)
assert plans[0].integration_focus
assert plans[0].troubleshooting
assert plans[0].codefix
def test_investing_process_surfaces_multi_platform_strategies():
engine = MarketingIntelligenceEngine.from_file()
prompt = "Launch Yieldstreet and SoFi investing funnels with compliant automation"
ranked = engine.recommend_statistical(prompt, limit=3)
plans = engine.investing_enrichment(prompt, ranked, limit=3)
assert plans
providers = {plan.provider.lower() for plan in plans}
assert "yieldstreet" in providers
assert any("sofi" in provider for provider in providers)
top_plan = plans[0]
assert top_plan.asset_classes
assert top_plan.monetization
assert top_plan.risk_controls
assert top_plan.maturity in {"foundational", "advanced", "autonomous"}
def test_quantum_simulation_process_covers_required_domains():
engine = MarketingIntelligenceEngine.from_file()
prompt = "Stress test google ads, aws, payments, and markets with quantum"
ranked = engine.recommend_statistical(prompt, limit=3)
process = QuantumSimulationProcess()
plans = process.plans(prompt, ranked, limit=1)
assert plans
plan = plans[0]
coverage = {test.domain.lower() for test in plan.stress_tests}
required = {
"google_ads",
"aws",
"marketing",
"bitcoin",
"stockmarket",
"stripe",
"cashapp",
"venmo",
"paypal",
"agentic",
}
assert required.issubset(coverage)
assert all(0.0 <= test.measurement_probability <= 1.0 for test in plan.stress_tests)
assert plan.orchestration_steps
assert plan.coverage_ratio >= 0.9
assert required.issubset(plan.domain_stability.keys())
assert all(0.0 <= value <= 1.0 for value in plan.domain_stability.values())
assert plan.agents_per_node >= 1
assert plan.nodes_for_one_million >= 1
assert plan.compression_index >= 0
assert plan.shots >= 1
assert isinstance(plan.continuous_operation, bool)
assert all(test.agents_per_node >= 1 for test in plan.stress_tests)
assert all(test.nodes_for_one_million >= 1 for test in plan.stress_tests)
def test_physics_economics_process_generates_multidisciplinary_plan():
engine = MarketingIntelligenceEngine.from_file()
prompt = "Blend physics, quantum, and economics to scale marketing mechanics"
ranked = engine.recommend_statistical(prompt, limit=3)
process = PhysicsEconomicsProcess()
plans = process.plans(prompt, ranked, limit=2)
assert plans
plan = plans[0]
assert plan.physics_principles
assert plan.economic_models
assert plan.orchestration_steps
assert plan.metrics
def test_quantum_simulation_enrichment_integrates_with_engine():
engine = MarketingIntelligenceEngine.from_file()
prompt = "Need quantum stress telemetry for payments and market plays"
ranked = engine.recommend_statistical(prompt, limit=2)
plans = engine.quantum_simulation_enrichment(prompt, ranked, limit=1)
assert plans
plan = plans[0]
assert plan.training_focus
assert plan.stress_tests
assert plan.metrics
assert plan.coverage_ratio > 0
assert plan.domain_stability
assert plan.agents_per_node >= 1
assert plan.nodes_for_one_million >= 1
assert plan.shots >= 1
def test_marketing_processes_endpoint_combines_rag_and_egg():
payload = marketing_processes(
MarketingQuery(prompt="Need urgent conversion plays for enterprise SaaS founders", limit=2)
)
assert payload.query == "Need urgent conversion plays for enterprise SaaS founders"
assert payload.rag.process == "rag"
assert len(payload.rag.recommendations) == 2
assert payload.egg
assert payload.egg[0].playbook
assert payload.google_ads
assert payload.google_ads[0].source in {"simulated", "google_ads_api"}
assert payload.cj_affiliate
assert payload.cj_affiliate[0].projected_monthly_revenue > 0
assert payload.crypto_markets
assert payload.crypto_markets[0].asset
assert payload.stock_markets
assert payload.stock_markets[0].symbol
assert payload.stripe
assert payload.stripe[0].product_name
assert payload.paypal
assert payload.paypal[0].checkout_flow
assert payload.payment_integrations
assert payload.payment_integrations[0].provider
assert payload.investing
assert payload.investing[0].provider
assert payload.cognitive_architecture
assert payload.cognitive_architecture[0].capability
assert payload.agentic_systems
assert payload.agentic_systems[0].system_name
assert payload.agentic_systems[0].toolchain
assert payload.rag_workflows
assert payload.knowledge_graphs
assert payload.marketing_knowledge
assert payload.marketing_knowledge[0].coverage_assets
assert payload.language_intelligence
assert payload.language_intelligence[0].capability
assert payload.code_automation
assert payload.code_automation[0].automation_stages
assert payload.meta_learning
assert payload.meta_learning[0].learning_modes
assert payload.mixture_of_experts
assert payload.agentic_loops
assert payload.quantum_simulation
quantum_plan = payload.quantum_simulation[0]
assert quantum_plan.coverage_ratio > 0
assert "google_ads" in {domain for domain in quantum_plan.domain_stability}
assert payload.quantum_simulation[0].stress_tests
assert payload.quantum_simulation[0].agents_per_node >= 1
assert payload.quantum_simulation[0].nodes_for_one_million >= 1
assert payload.quantum_simulation[0].shots >= 1
assert isinstance(payload.quantum_simulation[0].continuous_operation, bool)
assert payload.physics_economics
assert payload.physics_economics[0].physics_principles
assert payload.influencer_collaborations
assert payload.influencer_collaborations[0].projected_roi > 0
assert payload.amazon_marketplace
assert payload.amazon_marketplace[0].estimated_monthly_sales > 0
assert payload.amazon_marketplace[0].fulfillment_model
assert payload.ebay_marketplace
assert payload.ebay_marketplace[0].projected_profit_margin > 0
assert payload.aws_cloud
assert payload.budget_optimizer.total_budget >= sum(
allocation.allocation_usd for allocation in payload.budget_optimizer.allocations
)
assert payload.budget_optimizer.expected_revenue >= payload.budget_optimizer.total_budget
assert payload.automation.total_minutes == 180
assert payload.automation.tasks
assert payload.automation.tasks[0].start_minute == 0
assert payload.learning_loop
assert payload.learning_loop[0].next_actions
assert payload.competitive_landscape
assert payload.competitive_landscape[0].threat_level in {"low", "medium", "high"}
assert payload.excellence.maturity_level in {"emerging", "scaling", "elite", "world-class"}
assert 0 <= payload.excellence.excellence_index <= 100
assert "rag" in payload.excellence.process_coverage
assert "ebay_marketplace" in payload.excellence.process_coverage
assert "paypal" in payload.excellence.process_coverage
assert "payment_integrations" in payload.excellence.process_coverage
assert "investing" in payload.excellence.process_coverage
assert "influencer_collaborations" in payload.excellence.process_coverage
assert "stock_markets" in payload.excellence.process_coverage
assert "cognitive_architecture" in payload.excellence.process_coverage
assert "agentic_systems" in payload.excellence.process_coverage
assert "rag_workflows" in payload.excellence.process_coverage
assert "knowledge_graphs" in payload.excellence.process_coverage
assert "marketing_knowledge" in payload.excellence.process_coverage
assert "language_intelligence" in payload.excellence.process_coverage
assert "code_automation" in payload.excellence.process_coverage
assert "meta_learning" in payload.excellence.process_coverage
assert "mixture_of_experts" in payload.excellence.process_coverage
assert "agentic_loops" in payload.excellence.process_coverage
assert "quantum_simulation" in payload.excellence.process_coverage
assert "physics_economics" in payload.excellence.process_coverage
assert payload.excellence.strengths
assert payload.excellence.watchouts
assert payload.executive_playbook.headline
assert payload.executive_playbook.directives
top_directive = payload.executive_playbook.directives[0]
assert top_directive.supporting_channels
assert payload.executive_playbook.north_star_metric in top_directive.supporting_metrics
assert top_directive.action_steps
assert top_directive.risk_mitigations
assert any(channel.startswith("Markets -") for channel in top_directive.supporting_channels)
assert payload.scenario_planning.headline
assert payload.scenario_planning.projections
scenario_labels = {projection.scenario for projection in payload.scenario_planning.projections}
assert {"optimistic", "base", "conservative"}.issubset(scenario_labels)
assert payload.scenario_planning.projections[0].probability > 0
assert payload.scenario_planning.accelerators
assert payload.scenario_planning.stabilizers
assert payload.statistical_reasoning.expected_roi > 0
assert payload.statistical_reasoning.success_probability > 0
assert payload.statistical_reasoning.evidence
def test_marketing_engine_executive_playbook_blends_processes():
engine = MarketingIntelligenceEngine.from_file()
prompt = "Accelerate enterprise pipeline with automation and partnerships"
rag_campaigns = engine.recommend(prompt, limit=3)
ranked = engine.recommend_statistical(prompt, limit=3)
egg = engine.expansion_grid(prompt, ranked)
google_ads = engine.google_ads_enrichment(prompt, ranked, limit=3)
cj_affiliate = engine.cj_affiliate_enrichment(prompt, ranked, limit=3)
crypto_markets = engine.crypto_market_enrichment(prompt, limit=3)
stock_markets = engine.stock_market_enrichment(prompt, ranked, limit=3)
stripe_plans = engine.stripe_checkout_enrichment(prompt, ranked, limit=3)
paypal_plans = engine.paypal_checkout_enrichment(prompt, ranked, limit=3)
payment_plans = engine.payment_integration_enrichment(prompt, ranked, limit=3)
investing_plans = engine.investing_enrichment(prompt, ranked, limit=3)
influencers = engine.influencer_collaboration_enrichment(prompt, ranked, limit=3)
amazon = engine.amazon_marketplace_enrichment(prompt, ranked, limit=3)
ebay = engine.ebay_marketplace_enrichment(prompt, ranked, limit=3)
aws = engine.aws_cloud_enrichment(prompt, ranked, limit=3)
budget = engine.budget_optimizer(prompt, ranked, limit=3)
automation = engine.automation_sprint(prompt, ranked, limit=3)
learning = engine.learning_loop(prompt, ranked, limit=3)
competitive = engine.competitive_landscape(prompt, ranked, limit=3)
cognitive = engine.cognitive_architecture_enrichment(prompt, ranked, limit=3)
agentic = engine.agentic_system_enrichment(prompt, ranked, limit=3)
knowledge = engine.knowledge_workflow_enrichment(prompt, ranked, limit=3)
marketing_knowledge = engine.marketing_knowledge_enrichment(prompt, ranked, limit=3)
language = engine.language_capability_enrichment(prompt, ranked, limit=3)
meta_learning = engine.meta_learning_enrichment(prompt, ranked, limit=3)
moe_plans = engine.mixture_of_experts_activation(prompt, ranked, limit=3)
agentic_loops = engine.agentic_loop_blueprints(prompt, ranked)
quantum = engine.quantum_simulation_enrichment(prompt, ranked, limit=3)
physics = engine.physics_economics_enrichment(prompt, ranked, limit=3)
assert agentic
excellence = engine.excellence_summary(
prompt,
ranked,
limit=3,
rag=rag_campaigns,
egg=egg,
google_ads=google_ads,
cj_affiliate=cj_affiliate,
crypto_markets=crypto_markets,
stocks=stock_markets,
stripe=stripe_plans,
paypal=paypal_plans,
payments=payment_plans,
investing=investing_plans,
influencers=influencers,
amazon=amazon,
ebay=ebay,
aws=aws,
budget=budget,
automation=automation,
learning_loop=learning,
competitive=competitive,
cognitive=cognitive,
agentic=agentic,
knowledge=knowledge,
marketing_knowledge=marketing_knowledge,
language=language,
meta_learning=meta_learning,
moe=moe_plans,
agentic_loops=agentic_loops,
quantum=quantum,
physics_economics=physics,
)
playbook = engine.executive_playbook(
prompt,
ranked,
limit=3,
rag=rag_campaigns,
egg=egg,
google_ads=google_ads,
cj_affiliate=cj_affiliate,
amazon=amazon,
ebay=ebay,
paypal=paypal_plans,
payments=payment_plans,
investing=investing_plans,
influencers=influencers,
stocks=stock_markets,
budget=budget,
automation=automation,
learning_loop=learning,
competitive=competitive,
excellence=excellence,
statistics=None,
cognitive=cognitive,
agentic=agentic,
knowledge=knowledge,
marketing_knowledge=marketing_knowledge,
language=language,
meta_learning=meta_learning,
moe=moe_plans,
agentic_loops=agentic_loops,
quantum=quantum,
physics_economics=physics,
)
assert playbook.directives
assert playbook.north_star_metric
assert playbook.confidence >= 0.35
assert playbook.notes
directive = playbook.directives[0]
assert directive.priority == 1
assert directive.primary_channel in directive.supporting_channels[0]
assert directive.supporting_metrics
assert any("Knowledge" in channel for channel in directive.supporting_channels)
assert any("Language" in channel for channel in directive.supporting_channels)
assert any("Meta-Learning" in channel for channel in directive.supporting_channels)
assert playbook.north_star_metric in directive.supporting_metrics
assert directive.dependencies
assert any("eBay" in channel for channel in directive.supporting_channels)
assert any("PayPal" in channel for channel in directive.supporting_channels)
assert any("Influencer" in channel for channel in directive.supporting_channels)
assert any("Investing" in channel for channel in directive.supporting_channels)
assert any(channel.startswith("Markets -") for channel in directive.supporting_channels)
assert any("Cognitive" in channel for channel in directive.supporting_channels)
assert any("Agentic" in channel for channel in directive.supporting_channels)
assert any("Quantum" in channel for channel in directive.supporting_channels)
assert any("Physics/Economics" in channel for channel in directive.supporting_channels)
def test_scenario_planning_process_generates_three_outcomes():
engine = MarketingIntelligenceEngine.from_file()
prompt = "Drive enterprise conversions with integrated paid and partner motions"
ranked = engine.recommend_statistical(prompt, limit=3)
budget = engine.budget_optimizer(prompt, ranked, limit=3)
google_ads = engine.google_ads_enrichment(prompt, ranked, limit=3)
cj_affiliate = engine.cj_affiliate_enrichment(prompt, ranked, limit=3)
amazon = engine.amazon_marketplace_enrichment(prompt, ranked, limit=3)
aws = engine.aws_cloud_enrichment(prompt, ranked, limit=3)
stripe = engine.stripe_checkout_enrichment(prompt, ranked, limit=3)
paypal = engine.paypal_checkout_enrichment(prompt, ranked, limit=3)
crypto = engine.crypto_market_enrichment(prompt, limit=3)
stocks = engine.stock_market_enrichment(prompt, ranked, limit=3)
ebay = engine.ebay_marketplace_enrichment(prompt, ranked, limit=3)
influencers = engine.influencer_collaboration_enrichment(prompt, ranked, limit=3)
cognitive = engine.cognitive_architecture_enrichment(prompt, ranked, limit=3)
language = engine.language_capability_enrichment(prompt, ranked, limit=3)
meta_learning = engine.meta_learning_enrichment(prompt, ranked, limit=3)
quantum = engine.quantum_simulation_enrichment(prompt, ranked, limit=3)
physics = engine.physics_economics_enrichment(prompt, ranked, limit=3)
payments = engine.payment_integration_enrichment(prompt, ranked, limit=3)
investing = engine.investing_enrichment(prompt, ranked, limit=3)
scenario = engine.scenario_planning(
prompt,
ranked,
limit=3,
budget=budget,
google_ads=google_ads,
cj_affiliate=cj_affiliate,
amazon=amazon,
ebay=ebay,
aws=aws,
stripe=stripe,
paypal=paypal,
payments=payments,
investing=investing,
crypto=crypto,
influencers=influencers,
stocks=stocks,
statistics=None,
cognitive=cognitive,
agentic=None,
knowledge=None,
code_automation=None,
language=language,
meta_learning=meta_learning,
moe=None,
agentic_loops=None,
quantum=quantum,
physics_economics=physics,
)
assert scenario.projections
assert len(scenario.projections) == 3
assert {proj.scenario for proj in scenario.projections} == {"optimistic", "base", "conservative"}
assert all(proj.expected_revenue > 0 for proj in scenario.projections)
assert scenario.accelerators
assert any("Equity" in accel or "market" in accel.lower() for accel in scenario.accelerators)
assert any("cognitive" in accel.lower() for accel in scenario.accelerators)
assert any("language" in accel.lower() for accel in scenario.accelerators)
assert any("meta" in accel.lower() for accel in scenario.accelerators)
assert any("quantum" in accel.lower() for accel in scenario.accelerators)
assert any("physics" in accel.lower() for accel in scenario.accelerators)
assert any("investing" in accel.lower() for accel in scenario.accelerators)
assert scenario.stabilizers
assert 0 < scenario.confidence <= 1