scripts/e2e_smoke.py

"""End-to-end smoke: API checks + headless browser against the tabbed mobile-first UI.

Set E2E_BASE_URL (default http://127.0.0.1:8012) to match your dev server.
"""
from __future__ import annotations

import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

from selenium.common.exceptions import TimeoutException
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait

BASE_URL = os.environ.get("E2E_BASE_URL", "http://127.0.0.1:8012").rstrip("/")
DOMINANT_ROLE_ID = "fetlife_kinktionary_dominant_wzwjm"


def get_json(path: str, headers: dict[str, str] | None = None) -> dict:
    request = urllib.request.Request(f"{BASE_URL}{path}", headers=headers or {})
    with urllib.request.urlopen(request, timeout=120) as response:
        return json.loads(response.read().decode("utf-8"))


def post_json(path: str, payload: dict, headers: dict[str, str] | None = None) -> dict:
    request_headers = {"Content-Type": "text/plain;charset=UTF-8"}
    if headers:
        request_headers.update(headers)
    request = urllib.request.Request(
        f"{BASE_URL}{path}",
        data=json.dumps(payload).encode("utf-8"),
        headers=request_headers,
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=120) as response:
        return json.loads(response.read().decode("utf-8"))


def assert_true(condition: bool, message: str) -> None:
    if not condition:
        raise AssertionError(message)


def wait_text(driver: webdriver.Chrome, text: str, timeout: int = 15):
    return WebDriverWait(driver, timeout).until(
        EC.presence_of_element_located((By.XPATH, f"//*[contains(normalize-space(), {json.dumps(text)})]"))
    )


def run_browser_smoke() -> dict:
    options = Options()
    options.add_argument("--headless=new")
    options.add_argument("--disable-gpu")
    options.add_argument("--window-size=1440,1400")
    options.add_argument("--no-sandbox")
    driver = webdriver.Chrome(options=options)
    wait = WebDriverWait(driver, 25)
    long_wait = WebDriverWait(driver, 120)
    results: dict[str, str] = {}
    try:
        driver.get(BASE_URL)

        wait.until(EC.element_to_be_clickable((By.XPATH, "//button[normalize-space()='Create profile']"))).click()
        wait.until(EC.presence_of_element_located((By.XPATH, "//*[contains(., \"Set up your profile\")]")))

        codes = wait.until(lambda d: d.find_elements(By.XPATH, "//div[contains(@class, 'creds')]//code"))
        assert_true(len(codes) >= 2, "create flow did not render profile credentials")
        profile_id = codes[0].text.strip()
        private_pass = codes[1].text.strip()
        results["profile_id"] = profile_id

        role_input = wait.until(
            EC.element_to_be_clickable((By.XPATH, "//input[@placeholder='Search roles...']"))
        )
        role_input.clear()
        role_input.send_keys("dominant")
        try:
            add_btn = long_wait.until(
                EC.element_to_be_clickable(
                    (
                        By.XPATH,
                        "//div[contains(@class,'play-chip')]"
                        "//span[contains(@class,'chip-name') and contains(., 'Dominant')]/following-sibling::button[normalize-space()='Add']",
                    )
                )
            )
            add_btn.click()
        except TimeoutException:
            post_json(
                f"/users/{profile_id}/roles",
                {"kink_id": DOMINANT_ROLE_ID, "selected": True},
                headers={"x-private-token": private_pass},
            )

        wait.until(EC.element_to_be_clickable((By.XPATH, "//button[normalize-space()='Start starter deck']"))).click()

        long_wait.until(
            EC.presence_of_element_located(
                (By.XPATH, "//div[contains(@class,'top-bar-title') and normalize-space()='Starter deck']")
            )
        )
        for _ in range(12):
            long_wait.until(EC.element_to_be_clickable((By.XPATH, "//button[contains(normalize-space(),'Skip')]"))).click()

        wait.until(EC.presence_of_element_located((By.XPATH, "//nav[contains(@class,'tab-bar')]")))
        wait_text(driver, "Discover", timeout=20)

        long_wait.until(
            EC.presence_of_element_located(
                (By.XPATH, "//div[contains(@class,'top-bar-title') and normalize-space()='Discover']")
            )
        )
        long_wait.until(EC.presence_of_element_located((By.XPATH, "//button[contains(normalize-space(),'Love')]")))

        wait.until(EC.element_to_be_clickable((By.XPATH, "//button[contains(@class,'tab-btn') and contains(.,'Discover')]")))

        wait.until(
            EC.element_to_be_clickable((By.XPATH, "//div[contains(@class,'top-bar-right')]//button[contains(normalize-space(),'Search')]"))
        ).click()
        kink_search = wait.until(
            EC.element_to_be_clickable((By.XPATH, "//div[contains(@class,'search-overlay')]//input[@placeholder='Search kinks...']"))
        )
        kink_search.clear()
        kink_search.send_keys("bondage")
        long_wait.until(
            EC.element_to_be_clickable(
                (
                    By.XPATH,
                    "//div[contains(@class,'search-overlay')]//div[contains(@class,'play-chip')][.//span[contains(.,'Bondage')]]",
                )
            )
        ).click()

        wait.until(EC.presence_of_element_located((By.XPATH, "//div[contains(@class,'top-bar-title') and normalize-space()='My Plays']")))

        try:
            wait.until(EC.element_to_be_clickable((By.XPATH, "//button[contains(normalize-space(),'Show')]"))).click()
            wait.until(EC.presence_of_element_located((By.XPATH, "//div[contains(@class,'tinder-card')]//img")))
        except TimeoutException:
            pass

        results["private_pass"] = private_pass
        return results
    finally:
        driver.quit()


def run_api_smoke(credentials: dict[str, str]) -> None:
    headers = {"x-private-token": credentials["private_pass"]}
    user = get_json(f"/users/{credentials['profile_id']}", headers=headers)
    assert_true(isinstance(user.get("roles"), list), "user payload missing roles")

    search = get_json("/search?q=bondage&limit=1")
    top = search["items"][0]["kink"]
    assert_true(top["name"].strip().lower() == "bondage", "bondage search did not return exact bondage row")
    assert_true(bool(top["has_images"]), "canonical bondage result should report images")
    bondage_id = top["id"]

    roles = get_json("/roles/search?q=dominant&limit=3")
    assert_true(all(item["kink"]["content_kind"] in {"role", "dynamic"} for item in roles["items"]), "role search leaked non-role items")

    plays = get_json("/search?q=dominant&limit=3")
    assert_true(plays["items"] == [], "play search should not surface dynamic role terms by default")

    detail = get_json(f"/kinks/{bondage_id}")
    assert_true(detail["filtered_asset_count"] > 0, "bondage detail missing trusted assets")

    recs = get_json(f"/users/{credentials['profile_id']}/recommendations?limit=8", headers=headers)
    items = recs["items"]
    assert_true(len(items) >= 1, "recommendations returned no items")
    assert_true(
        all(item.get("discovery_source") in {"starter", "personalized", "explore", "partner_influenced"} for item in items),
        f"recommendations missing discovery source labels: {[item.get('discovery_source') for item in items]}",
    )
    health = get_json("/admin/fetlife/data-health?limit=5")
    assert_true("starter_priority" in health["debt"], "data health missing prioritized crawl debt")
    assert_true("crawl" in health and "queue" in health, "data health missing crawl/queue readiness")
    assert_true("representative_image_kinks" in health["coverage"], "data health missing representative image coverage")
    debug = get_json(f"/admin/recommendation-debug/{credentials['profile_id']}?limit=5", headers=headers)
    assert_true("signals" in debug and "items" in debug, "recommendation debug payload invalid")

    partner = post_json("/users", {})
    partner_headers = {"x-private-token": partner["private_token"]}
    post_json(
        f"/users/{credentials['profile_id']}/plays",
        {"kink_id": bondage_id, "interest_state": "love", "directions": ["to_me"]},
        headers=headers,
    )
    post_json(
        f"/users/{partner['id']}/plays",
        {"kink_id": bondage_id, "interest_state": "love", "directions": ["by_me"]},
        headers=partner_headers,
    )
    post_json(
        f"/users/{credentials['profile_id']}/partners",
        {"partner_id": partner["id"]},
        headers=headers,
    )
    post_json(
        f"/users/{partner['id']}/partner-requests/accept",
        {"from_user_id": credentials["profile_id"]},
        headers=partner_headers,
    )
    group = post_json(
        f"/users/{credentials['profile_id']}/partner-groups",
        {"name": "Smoke Group", "member_ids": [partner["id"]]},
        headers=headers,
    )
    groups = get_json(f"/users/{credentials['profile_id']}/partner-groups", headers=headers)
    assert_true(any(item["id"] == group["id"] for item in groups["items"]), "created partner group missing from list")
    overlap = get_json(
        f"/users/{credentials['profile_id']}/partner-groups/{group['id']}/overlap",
        headers=headers,
    )
    assert_true(any(item["id"] == bondage_id for item in overlap["direct_matches"]), "group overlap missed compatible play")


def main() -> int:
    try:
        health = get_json("/health")
        assert_true(bool(health.get("ok")), "health check failed")
        credentials = run_browser_smoke()
        run_api_smoke(credentials)
    except (AssertionError, urllib.error.URLError, TimeoutError) as exc:
        print(f"SMOKE FAILED: {exc}", file=sys.stderr)
        return 1
    print("SMOKE OK")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())