Janady13/AI_Brain / tests/test_marketing_intelligence.py

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	840945d70b88674d8f40d330001cc63d871dc8e2
Branch	codex/build-online-github-ani-nbgsix
Path or archive member	tests/test_marketing_intelligence.py
Original SHA-256	a8d9509ce461071cb162b361d96ed6b484758c9b0be1d5aac34f4f30ea9f726c
Git blob SHA	fb807217b29ce4e1a75cd97b27c94364f606eec9
Reading-copy SHA-256	a8d9509ce461071cb162b361d96ed6b484758c9b0be1d5aac34f4f30ea9f726c
Original source ↗
Recovered source transcript

Download complete source text ↓257 lines · 10.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,
    CJAffiliateProcess,
    CryptoMarketProcess,
    GoogleAdsProcess,
    MarketingCampaign,
    MarketingIntelligenceEngine,
    StripeProcess,
)


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_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_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_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_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.stripe
    assert payload.stripe[0].product_name
    assert payload.amazon_marketplace
    assert payload.aws_cloud