ponzischeme89
253 lines
8.6 KiB
Python
253 lines
8.6 KiB
Python
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
|