from datetime import datetime, timedelta, timezone from decimal import Decimal import pytest from sqlalchemy import select from app.models.experiment import ExperimentEvent from app.models.settings import SiteSettings from tests.conftest import TestSessionLocal @pytest.mark.asyncio async def test_list_experiments_returns_seeded_registry(client): response = await client.get("/api/experiments") assert response.status_code == 200, response.text body = response.json() assert [item["experiment_key"] for item in body] == [ "homepage_hero_test", "pricing_cta_test", ] assert body[0]["cookie_name"] == "exp_homepage_hero" assert body[0]["variants"][0]["variant_key"] == "control" @pytest.mark.asyncio async def test_ingest_experiment_event_persists_event(client): now = datetime.now(timezone.utc).isoformat() response = await client.post( "/api/experiments/event", json={ "experiment_key": "homepage_hero_test", "variant_key": "control", "session_id": "session_abcd1234", "path": "/", "event_name": "cta_click", "timestamp": now, "metadata": { "element": "Explore Pack Walks", "slot": "primary", }, }, headers={"user-agent": "Mozilla/5.0"}, ) assert response.status_code == 202, response.text assert response.json() == {"ok": True, "accepted": True} async with TestSessionLocal() as session: result = await session.execute(select(ExperimentEvent)) event = result.scalar_one() assert event.experiment_key == "homepage_hero_test" assert event.variant_key == "control" assert event.event_type == "cta_click" assert event.metadata_["element"] == "Explore Pack Walks" @pytest.mark.asyncio async def test_ingest_experiment_event_filters_bots(client): response = await client.post( "/api/experiments/impression", json={ "experiment_key": "homepage_hero_test", "variant_key": "control", "session_id": "session_abcd1234", "path": "/", "event_name": "impression", "timestamp": datetime.now(timezone.utc).isoformat(), }, headers={"user-agent": "Googlebot/2.1"}, ) assert response.status_code == 202, response.text assert response.json() == {"ok": True, "accepted": False} async with TestSessionLocal() as session: result = await session.execute(select(ExperimentEvent)) assert result.scalars().all() == [] @pytest.mark.asyncio async def test_experiment_results_aggregate_by_variant(client, admin_token): now = datetime.now(timezone.utc) async with TestSessionLocal() as session: session.add_all( [ ExperimentEvent( experiment_key="pricing_cta_test", variant_key="control", session_id="session-1", path="/our-pricing", event_type="impression", created_at=(now - timedelta(minutes=5)).replace(tzinfo=None), ), ExperimentEvent( experiment_key="pricing_cta_test", variant_key="control", session_id="session-1", path="/our-pricing", event_type="cta_click", created_at=(now - timedelta(minutes=4)).replace(tzinfo=None), ), ExperimentEvent( experiment_key="pricing_cta_test", variant_key="control", session_id="session-1", path="/our-pricing", event_type="conversion", conversion_value=Decimal("1.00"), created_at=(now - timedelta(minutes=3)).replace(tzinfo=None), ), ExperimentEvent( experiment_key="pricing_cta_test", variant_key="meet_greet_emphasis", session_id="session-2", path="/our-pricing", event_type="impression", created_at=(now - timedelta(minutes=2)).replace(tzinfo=None), ), ExperimentEvent( experiment_key="pricing_cta_test", variant_key="meet_greet_emphasis", session_id="session-3", path="/our-pricing", event_type="impression", created_at=(now - timedelta(minutes=1)).replace(tzinfo=None), ), ] ) await session.commit() response = await client.get( "/api/v1/experiments/results?experiment_key=pricing_cta_test", headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200, response.text body = response.json() assert len(body) == 1 assert body[0]["experiment_key"] == "pricing_cta_test" assert body[0]["variants"][0] == { "variant_key": "control", "impressions": 1, "cta_clicks": 1, "form_starts": 0, "form_submits": 0, "conversions": 1, "unique_sessions": 1, "conversion_rate": 1.0, "conversion_value_total": 1.0, } @pytest.mark.asyncio async def test_admin_can_update_backend_managed_experiment_definition(client, admin_token): response = await client.put( "/api/admin/experiments/homepage_hero_test", headers={"Authorization": f"Bearer {admin_token}"}, json={ "cookie_name": "exp_homepage_hero", "name": "Homepage hero test", "description": "Updated from admin", "enabled": False, "eligible_routes": ["/", "/contact"], "variants": [ { "variant_key": "control", "label": "Original", "allocation": 20, "is_control": True, }, { "variant_key": "tiny_gang_social_proof", "label": "Social proof", "allocation": 80, "is_control": False, }, ], }, ) assert response.status_code == 200, response.text body = response.json() assert body["enabled"] is False assert body["eligible_routes"] == ["/", "/contact"] assert body["variants"][1]["allocation"] == 80 public_response = await client.get("/api/experiments") assert public_response.status_code == 200, public_response.text public_body = public_response.json() updated = next(item for item in public_body if item["experiment_key"] == "homepage_hero_test") assert updated["enabled"] is False assert updated["description"] == "Updated from admin" @pytest.mark.asyncio async def test_public_experiments_return_empty_when_globally_disabled(client): async with TestSessionLocal() as session: session.add(SiteSettings(site_name="", experiments_enabled=False)) await session.commit() response = await client.get("/api/experiments") assert response.status_code == 200, response.text assert response.json() == [] @pytest.mark.asyncio async def test_experiment_ingest_is_ignored_when_globally_disabled(client): async with TestSessionLocal() as session: session.add(SiteSettings(site_name="", experiments_enabled=False)) await session.commit() now = datetime.now(timezone.utc).isoformat() response = await client.post( "/api/experiments/event", json={ "experiment_key": "homepage_hero_test", "variant_key": "control", "session_id": "session_abcd1234", "path": "/", "event_name": "cta_click", "timestamp": now, }, headers={"user-agent": "Mozilla/5.0"}, ) assert response.status_code == 202, response.text assert response.json() == {"ok": True, "accepted": False} async with TestSessionLocal() as session: result = await session.execute(select(ExperimentEvent)) assert result.scalars().all() == [] @pytest.mark.asyncio async def test_admin_experiments_returns_404_when_globally_disabled(client, admin_token): async with TestSessionLocal() as session: session.add(SiteSettings(site_name="", experiments_enabled=False)) await session.commit() response = await client.get( "/api/admin/experiments", headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 404, response.text