from __future__ import annotations import json import os import threading import time import urllib.parse import urllib.request from dataclasses import asdict, dataclass from typing import Callable from playwright.sync_api import Browser, Page, expect, sync_playwright BASE_URL = os.environ.get("KINK_AUDIT_BASE_URL", "http://127.0.0.1:8012").rstrip("/") DEFAULT_SCENARIOS = ["onboarding", "starter", "settings", "couple"] _POLL_WAIT = threading.Event() def _e2e_onboarding_timeout_ms() -> int: """HF cold start + first ``POST /users`` can block on catalog init; allow long UI waits.""" return int(os.environ.get("KINK_E2E_ONBOARDING_TIMEOUT_MS", "120000") or "120000") @dataclass class Credentials: user_id: str private_token: str @dataclass class StageResult: name: str ok: bool details: dict[str, object] def _json_request(path: str, *, method: str = "GET", payload: dict | None = None, headers: dict[str, str] | None = None, timeout: int = 60) -> dict: request_headers: dict[str, str] = {} if payload is not None: request_headers["Content-Type"] = "application/json; charset=utf-8" if headers: request_headers.update(headers) request = urllib.request.Request( f"{BASE_URL}{path}", data=None if payload is None else json.dumps(payload).encode("utf-8"), headers=request_headers, method=method, ) with urllib.request.urlopen(request, timeout=timeout) as response: body = response.read().decode("utf-8") return json.loads(body) if body else {} def get_json(path: str, headers: dict[str, str] | None = None, timeout: int = 60) -> dict: return _json_request(path, method="GET", headers=headers, timeout=timeout) def post_json(path: str, payload: dict, headers: dict[str, str] | None = None, timeout: int = 60) -> dict: return _json_request(path, method="POST", payload=payload, headers=headers, timeout=timeout) def delete_json(path: str, headers: dict[str, str] | None = None, timeout: int = 60) -> dict: return _json_request(path, method="DELETE", headers=headers, timeout=timeout) def auth_headers(creds: Credentials) -> dict[str, str]: return {"x-private-token": creds.private_token} def create_user() -> Credentials: payload = post_json("/users", {}) return Credentials(user_id=payload["id"], private_token=payload["private_token"]) def get_user(creds: Credentials) -> dict: return get_json(f"/users/{creds.user_id}", headers=auth_headers(creds)) def list_groups(creds: Credentials) -> list[dict]: return get_json(f"/users/{creds.user_id}/partner-groups", headers=auth_headers(creds)).get("items", []) def save_play(creds: Credentials, kink_id: str, interest_state: str, directions: list[str]) -> dict: return post_json( f"/users/{creds.user_id}/plays", {"kink_id": kink_id, "interest_state": interest_state, "directions": directions}, headers=auth_headers(creds), ) def save_scenario(creds: Credentials, parent_kink_id: str, scenario_kink_id: str, interest_state: str, directions: list[str]) -> dict: return post_json( f"/users/{creds.user_id}/scenario-preferences", { "parent_kink_id": parent_kink_id, "scenario_kink_id": scenario_kink_id, "interest_state": interest_state, "directions": directions, }, headers=auth_headers(creds), ) def send_partner_request(creds: Credentials, partner_id: str) -> dict: return post_json( f"/users/{creds.user_id}/partners", {"partner_id": partner_id}, headers=auth_headers(creds), ) def accept_partner_request(creds: Credentials, from_user_id: str) -> dict: return post_json( f"/users/{creds.user_id}/partner-requests/accept", {"from_user_id": from_user_id}, headers=auth_headers(creds), ) def toggle_share(creds: Credentials, group_id: str, share: bool) -> dict: return post_json( f"/users/{creds.user_id}/partner-groups/{urllib.parse.quote(group_id)}/share-toggle", {"share": share}, headers=auth_headers(creds), ) def find_kink_id(query: str, expected_name: str | None = None) -> str: items = get_json(f"/search?q={urllib.parse.quote(query)}&limit=10").get("items", []) if expected_name is None and items: return items[0]["kink"]["id"] expected = str(expected_name or "").strip().lower() for item in items: if item["kink"]["name"].strip().lower() == expected: return item["kink"]["id"] raise AssertionError(f"Could not find kink for query={query!r} expected_name={expected_name!r}") def find_first_exact_name(options: list[tuple[str, str]]) -> tuple[str, str]: for query, expected_name in options: try: return find_kink_id(query, expected_name), expected_name except AssertionError: continue raise AssertionError(f"Could not find any exact kink match from: {[name for _, name in options]}") def find_first_exact_name_not_in(options: list[tuple[str, str]], excluded_ids: set[str]) -> tuple[str, str]: for query, expected_name in options: try: kink_id = find_kink_id(query, expected_name) except AssertionError: continue if kink_id not in excluded_ids: return kink_id, expected_name raise AssertionError(f"Could not find any exact kink match outside excluded ids from: {[name for _, name in options]}") def ensure_exact_options_absent(creds: Credentials, options: list[tuple[str, str]]) -> None: for query, expected_name in options: try: kink_id = find_kink_id(query, expected_name) except AssertionError: continue ensure_play_absent(creds, kink_id) def find_parent_with_scenarios() -> tuple[str, str, str]: candidates = [("anal", "Anal"), ("touching", "Touching"), ("oral sex", "Oral Sex"), ("oral", "Oral")] for query, expected_name in candidates: parent_id = find_kink_id(query, expected_name) items = get_json(f"/kinks/{urllib.parse.quote(parent_id)}/scenarios?limit=8").get("items", []) if items: first = items[0] return parent_id, expected_name, first["scenario"]["id"] raise AssertionError("Could not find a parent kink with scenarios") def find_parent_with_scenario_pair() -> tuple[str, str, str, str]: candidates = [("anal", "Anal"), ("touching", "Touching"), ("oral sex", "Oral Sex"), ("oral", "Oral")] for query, expected_name in candidates: parent_id = find_kink_id(query, expected_name) items = get_json(f"/kinks/{urllib.parse.quote(parent_id)}/scenarios?limit=12").get("items", []) scenario_ids: list[str] = [] for item in items: scenario_id = item.get("scenario", {}).get("id") if scenario_id and scenario_id not in scenario_ids: scenario_ids.append(scenario_id) if len(scenario_ids) >= 2: return parent_id, expected_name, scenario_ids[0], scenario_ids[1] raise AssertionError("Could not find a parent kink with at least two scenarios") def seed_user_with_recommendations(creds: Credentials, target_count: int) -> list[str]: user = get_user(creds) seen = set(user.get("plays", {})) guard = 0 while len(seen) < target_count: guard += 1 if guard > 40: raise AssertionError(f"Could not seed {target_count} likes for {creds.user_id}; stalled at {len(seen)}") recs = get_json(f"/users/{creds.user_id}/recommendations?limit=24", headers=auth_headers(creds)).get("items", []) new_ids = [item["kink"]["id"] for item in recs if item["kink"]["id"] not in seen] if not new_ids: raise AssertionError(f"Recommendations exhausted while seeding {creds.user_id} at {len(seen)} plays") for kink_id in new_ids[: min(8, target_count - len(seen))]: save_play(creds, kink_id, "like", ["together"]) seen.add(kink_id) return sorted(seen) def ensure_play_absent(creds: Credentials, kink_id: str) -> None: if kink_id not in get_user(creds).get("plays", {}): return delete_json(f"/users/{creds.user_id}/plays/{urllib.parse.quote(kink_id)}", headers=auth_headers(creds)) def open_app(page: Page, *, path: str = "") -> None: page.goto(f"{BASE_URL}{path}", wait_until="load", timeout=120_000) def _run_in_fresh_context(browser: Browser, fn: Callable[[Page], dict[str, object]]) -> dict[str, object]: """Each scenario gets an isolated storage / cache origin (Playwright default context is shared).""" context = browser.new_context( viewport={"width": 1280, "height": 720}, extra_http_headers={"Cache-Control": "no-cache"}, ) page = context.new_page() try: return fn(page) finally: context.close() def create_profile_in_browser(page: Page, *, path: str = "") -> Credentials: open_app(page, path=path) t_onboard = _e2e_onboarding_timeout_ms() create_btn = page.get_by_test_id("onboarding-create-profile") expect(create_btn).to_be_visible(timeout=min(30000, t_onboard)) create_btn.click() expect(page.get_by_test_id("onboarding-setup")).to_be_visible(timeout=t_onboard) user_id = page.get_by_test_id("onboarding-created-user-id").inner_text().strip() reveal_btn = page.get_by_test_id("onboarding-reveal-token") if reveal_btn.count(): reveal_btn.click() expect(page.get_by_test_id("onboarding-created-private-token")).to_be_visible(timeout=15000) creds = Credentials( user_id=user_id, private_token=page.get_by_test_id("onboarding-created-private-token").inner_text().strip(), ) page.get_by_test_id("onboarding-start-starter").click() expect(page.get_by_test_id("starter-gate")).to_be_visible(timeout=15000) return creds def login_existing_user(page: Page, creds: Credentials) -> None: open_app(page) if page.get_by_test_id("onboarding-have-profile").count(): page.get_by_test_id("onboarding-have-profile").click() expect(page.get_by_test_id("onboarding-login")).to_be_visible(timeout=15000) page.get_by_test_id("onboarding-login-user-id").fill(creds.user_id) page.get_by_test_id("onboarding-login-private-token").fill(creds.private_token) page.get_by_test_id("onboarding-open-profile").click() def finish_starter_gate(page: Page, *, total_actions: int = 12) -> None: # Single long wait here (avoid duplicating a long ``starter-gate`` expect before this helper — that pairing # caused Chromium headless ``locator.click`` to hang on the first ``Like`` in practice). expect(page.get_by_test_id("starter-gate")).to_be_visible(timeout=120_000) if page.get_by_text("Could not load recommendations").count(): snippet = page.locator("body").inner_text()[:800] raise AssertionError(f"Discover recommendations error in browser. Body snippet:\n{snippet}") starter = page.locator('[data-testid="starter-discover-view"]') expect(starter).to_be_visible(timeout=120_000) expect(starter.get_by_test_id("discover-saved-count")).to_have_text("0 saved", timeout=120_000) expect(starter.locator('[data-testid="discover-direction-strip"]')).to_be_visible(timeout=120_000) bar = page.locator('[data-testid="discover-action-bar"]') expect(bar).to_be_visible(timeout=120_000) expect(bar.locator("button")).to_have_count(5, timeout=30_000) expect(bar.locator("button").nth(0)).to_have_attribute("data-testid", "discover-rate-love", timeout=10_000) expect(bar.locator("button").nth(1)).to_have_attribute("data-testid", "discover-rate-like", timeout=10_000) # Always tap Like on the current top card (re-mounts each advance). Mixing Skip was flaky under headless transitions. _expect_saved = ( "1 saved", "2 saved", "3 saved", "4 saved", "5 saved", "6 saved", "7 saved", "8 saved", "9 saved", "10 saved", "11 saved", "12 saved", ) if total_actions != 12: raise AssertionError("finish_starter_gate only supports total_actions=12 (starter gate contract)") for saved_label in _expect_saved: like_btn = page.get_by_test_id("discover-rate-like") expect(like_btn).to_be_attached(timeout=30_000) with page.expect_response( lambda res: res.request.method == "POST" and "/plays" in res.url, timeout=120_000, ) as play_resp: like_btn.click(force=True, timeout=120_000) resp = play_resp.value if not resp.ok: raise AssertionError(f"POST /plays not ok: status={resp.status} url={resp.url!r}") # After the 12th save the UI can swap ``starter-discover-view`` for ``discover-view`` immediately; # scope saved count to the page so the counter stays findable across that transition. expect(page.get_by_test_id("discover-saved-count")).to_have_text(saved_label, timeout=60_000) # Leaving the starter gate swaps ``starter-discover-view`` for main ``discover-view`` (nav mounts at 1024px+). expect(page.get_by_test_id("discover-view")).to_be_visible(timeout=90_000) expect(page.get_by_test_id("nav-tab-discover")).to_be_visible(timeout=30_000) def open_together(page: Page) -> None: page.get_by_test_id("nav-tab-together").click() expect(page.get_by_test_id("together-view")).to_be_visible(timeout=15000) def open_scenarios(page: Page) -> None: page.get_by_test_id("nav-tab-scenarios").click() expect(page.get_by_test_id("scenarios-view")).to_be_visible(timeout=15000) def open_settings(page: Page) -> None: page.get_by_test_id("nav-tab-settings").click() expect(page.get_by_test_id("settings-view")).to_be_visible(timeout=15000) def group_id_for_pair(owner: Credentials, partner_id: str) -> str: groups = list_groups(owner) for group in groups: participants = set(group.get("participant_ids", [])) if {owner.user_id, partner_id}.issubset(participants): return group["id"] raise AssertionError(f"Could not find partner group for {owner.user_id} and {partner_id}") def wait_for(predicate: Callable[[], bool], *, timeout_s: float = 20.0, step_s: float = 0.5, label: str = "condition") -> None: deadline = time.time() + timeout_s while time.time() < deadline: if predicate(): return _POLL_WAIT.wait(timeout=step_s) raise AssertionError(f"Timed out waiting for {label}") def render_summary(results: list[StageResult]) -> str: total = len(results) passed = sum(1 for result in results if result.ok) lines = [f"{passed}/{total} stages passed"] for result in results: status = "OK" if result.ok else "FAIL" detail = result.details.get("summary") or result.details.get("error") or "" lines.append(f"- {status} {result.name}: {detail}".rstrip(": ")) return "\n".join(lines) def _run_stage(name: str, fn: Callable[[], dict[str, object]]) -> StageResult: try: details = fn() return StageResult(name=name, ok=True, details=details) except Exception as exc: # noqa: BLE001 - report full scenario failure return StageResult(name=name, ok=False, details={"error": str(exc)}) def _seed_browser_session(page: Page, creds: Credentials) -> None: """Open the app with an existing profile (same as returning visitor with saved localStorage).""" open_app(page) page.evaluate( """([uid, tok]) => { localStorage.clear(); try { sessionStorage.clear(); } catch (e) {} localStorage.setItem("kink_auth", JSON.stringify({ userId: uid, token: tok })); }""", [creds.user_id, creds.private_token], ) page.reload(wait_until="load", timeout=120_000) def scenario_onboarding_create_button(page: Page) -> dict[str, object]: """Minimal UX: welcome → Create profile → credential setup (covers POST /users + DOM).""" open_app(page) page.get_by_test_id("onboarding-create-profile").click() expect(page.get_by_test_id("onboarding-setup")).to_be_visible(timeout=_e2e_onboarding_timeout_ms()) expect(page.get_by_test_id("onboarding-created-user-id")).to_be_visible(timeout=30_000) return {"summary": "Create profile reaches credential + role setup screen"} def scenario_starter_gate(page: Page) -> dict[str, object]: # Server-side user avoids fragile cold-start timing on POST /users vs first deck paint; starter + relogin is still all-UI. creds = create_user() fresh = get_user(creds) if len(fresh.get("plays") or {}): raise AssertionError(f"Expected new user with no plays before browser session, got {len(fresh['plays'])}") rec_items = get_json(f"/users/{creds.user_id}/recommendations?limit=8", headers=auth_headers(creds)).get("items", []) if len(rec_items) < 1: raise AssertionError("New user has no recommendations before browser session (seed/API contract)") # After 12 starter saves the catalog can be fully rated; ``GET /recommendations`` may return no rows. discover_source = rec_items[0]["discovery_source"] _seed_browser_session(page, creds) play_posts: list[str] = [] page.on( "request", lambda req: play_posts.append(req.url) if req.method == "POST" and "/plays" in req.url else None, ) try: finish_starter_gate(page) except Exception as exc: u_dbg = get_user(creds) n_dbg = len(u_dbg.get("plays") or {}) keys_dbg = list((u_dbg.get("plays") or {}).keys()) raise AssertionError( f"{exc}; after failure server play_count={n_dbg} play_keys={keys_dbg!r} " f"browser_post_plays_count={len(play_posts)}" ) from exc user = get_user(creds) saved_count = len(user.get("plays", {})) if saved_count < 12: raise AssertionError(f"starter gate unlocked before starter picks were durably saved (saved_count={saved_count})") # Full auth UX: leave app and log back in with the same credentials (guards logout/login + localStorage). expect(page.get_by_test_id("discover-view")).to_be_visible(timeout=15000) page.get_by_test_id("app-logout").click() expect(page.get_by_test_id("onboarding-welcome")).to_be_visible(timeout=15000) login_existing_user(page, creds) expect(page.get_by_test_id("discover-view")).to_be_visible(timeout=15000) user_after = get_user(creds) saved_after = len(user_after.get("plays", {})) if saved_after < 12: raise AssertionError(f"plays missing after re-login (saved_after={saved_after})") return { "summary": "create → starter → discover → logout → login → discover with durable plays", "user_id": creds.user_id, "saved_count": saved_count, "saved_after_relogin": saved_after, "first_play_directions": next(iter(user.get("plays", {}).values()))["directions"], "first_discovery_source": discover_source, } def scenario_settings_controls(page: Page) -> dict[str, object]: creds = create_user() seed_user_with_recommendations(creds, 12) login_existing_user(page, creds) expect(page.get_by_test_id("discover-view")).to_be_visible(timeout=15000) open_settings(page) expect(page.get_by_test_id("settings-export-data")).to_be_visible() expect(page.get_by_test_id("settings-delete-account")).to_be_visible() return { "summary": "settings exposes export and delete controls", "user_id": creds.user_id, } def scenario_couple_audit(browser: Browser) -> dict[str, object]: left = create_user() seed_user_with_recommendations(left, 50) checkpoints: list[dict[str, object]] = [ { "stage": "left_seeded", "expected": "left user can enter the app with a mature enough profile for partner comparison", "actual": {"user_id": left.user_id, "play_count": len(get_user(left).get("plays", {}))}, "summary": "left user seeded past 50 likes", } ] left_context = browser.new_context( viewport={"width": 1280, "height": 720}, extra_http_headers={"Cache-Control": "no-cache"}, ) right_context = browser.new_context( viewport={"width": 1280, "height": 720}, extra_http_headers={"Cache-Control": "no-cache"}, ) left_page = left_context.new_page() right_page = right_context.new_page() try: login_existing_user(left_page, left) expect(left_page.get_by_test_id("discover-view")).to_be_visible(timeout=15000) open_together(left_page) invite_url = left_page.get_by_test_id("together-share-link").locator("code").inner_text().strip() right = create_profile_in_browser(right_page, path=f"/?link={left.user_id}") finish_starter_gate(right_page) wait_for( lambda: left.user_id in get_user(right).get("partners", []) or right.user_id in get_user(left).get("incoming_partner_requests", []), label="right-side invite handoff", ) checkpoints.append( { "stage": "right_onboarded", "expected": "invite link survives onboarding and produces a pending partner connection", "actual": { "user_id": right.user_id, "play_count": len(get_user(right).get("plays", {})), "incoming_for_left": get_user(left).get("incoming_partner_requests", []), }, "summary": "right user completed starter gate from a share link", } ) wait_for( lambda: right.user_id in get_user(left).get("incoming_partner_requests", []), label="left incoming partner request", ) open_together(left_page) expect(left_page.get_by_test_id(f"together-accept-request-{right.user_id}")).to_be_visible(timeout=15000) left_page.get_by_test_id(f"together-accept-request-{right.user_id}").click() wait_for( lambda: right.user_id in get_user(left).get("partners", []), label="linked partners", ) checkpoints.append( { "stage": "connection_accepted", "expected": "left accepts once and the users become linked partners", "actual": { "left_partners": get_user(left).get("partners", []), "right_partners": get_user(right).get("partners", []), }, "summary": "partner request accepted and mirrored link created", } ) open_together(right_page) expect(right_page.get_by_test_id("together-group-picker")).to_be_visible(timeout=25000) seed_user_with_recommendations(right, 50) related_options = [ ("blow jobs", "Blow Jobs"), ("cocksucking", "Cocksucking"), ] private_options = [ ("foot worship", "Foot Worship"), ("lingerie", "Lingerie"), ] ensure_exact_options_absent(left, related_options + private_options) left_play_ids = set(get_user(left).get("plays", {})) direct_id = find_kink_id("bondage", "Bondage") left_only_related = find_kink_id("oral sex", "Oral Sex") right_only_related, right_related_name = find_first_exact_name_not_in(related_options, left_play_ids) right_private_only, private_only_name = find_first_exact_name_not_in(private_options, left_play_ids) ensure_play_absent(right, left_only_related) ensure_play_absent(left, right_only_related) ensure_play_absent(left, right_private_only) save_play(left, direct_id, "like", ["to_me"]) save_play(right, direct_id, "like", ["by_me"]) save_play(left, left_only_related, "like", ["together"]) save_play(right, right_only_related, "like", ["together"]) save_play(right, right_private_only, "love", ["together"]) parent_id, parent_name, scenario_id, related_scenario_id = find_parent_with_scenario_pair() save_play(left, parent_id, "like", ["together"]) save_play(right, parent_id, "like", ["together"]) save_scenario(left, parent_id, scenario_id, "like", ["together"]) save_scenario(right, parent_id, scenario_id, "like", ["together"]) save_scenario(right, parent_id, related_scenario_id, "like", ["together"]) group_id = group_id_for_pair(left, right.user_id) checkpoints.append( { "stage": "group_ready", "expected": "linked partners have a shared comparison group", "actual": {"group_id": group_id, "groups_for_left": [group["id"] for group in list_groups(left)]}, "summary": "shared partner group resolved after acceptance", } ) open_together(right_page) privacy_toggle = right_page.get_by_test_id("together-privacy-toggle") if not privacy_toggle.is_checked(): privacy_toggle.click() wait_for( lambda: right.user_id in next((g for g in list_groups(left) if g["id"] == group_id), {}).get("sharing_member_ids", []), label="share toggle propagation", ) right_page.reload(wait_until="domcontentloaded") open_together(right_page) expect(right_page.get_by_test_id("together-privacy-toggle")).to_be_checked(timeout=15000) checkpoints.append( { "stage": "privacy_enabled", "expected": "their-list sharing toggles on from the right user", "actual": next((g for g in list_groups(left) if g["id"] == group_id), {}), "summary": "privacy toggle exposed the partner list", } ) left_page.reload(wait_until="domcontentloaded") open_together(left_page) wait_for( lambda: any( item.get("name") == "Bondage" for item in get_json( f"/users/{left.user_id}/partner-groups/{urllib.parse.quote(group_id)}/overlap", headers=auth_headers(left), ).get("direct_matches", []) ), timeout_s=30.0, label="shared overlap readiness", ) expect(left_page.get_by_test_id(f"together-group-chip-{group_id}")).to_be_visible(timeout=15000) left_page.get_by_test_id(f"together-group-chip-{group_id}").click() expect(left_page.get_by_test_id("together-shared-panel")).to_be_visible(timeout=15000) expect(left_page.get_by_test_id("together-shared-direct-section")).to_be_visible(timeout=15000) expect(left_page.get_by_test_id("together-shared-scenarios-section")).to_be_visible(timeout=15000) expect(left_page.get_by_test_id(f"together-shared-scenario-{scenario_id}")).to_be_visible(timeout=15000) checkpoints.append( { "stage": "shared_visible", "expected": "Together > Shared shows direct overlaps and shared scenarios", "actual": { "direct_item": "Bondage", "scenario_parent": parent_name, "scenario_id": scenario_id, }, "summary": "shared direct and scenario matches rendered", } ) left_page.get_by_test_id(f"together-shared-scenario-{scenario_id}").click() expect(left_page.get_by_test_id("scenarios-view")).to_be_visible(timeout=15000) expect(left_page.get_by_test_id(f"scenario-parent-{parent_id}")).to_be_visible(timeout=15000) expect(left_page.get_by_test_id(f"scenario-row-{scenario_id}")).to_be_visible(timeout=15000) checkpoints.append( { "stage": "scenarios_view_visible", "expected": "clicking a shared scenario opens the Scenarios tab focused to the matching base kink", "actual": {"parent_id": parent_id, "scenario_id": scenario_id}, "summary": "shared scenario deep-linked into the Scenarios tab", } ) save_scenario(right, parent_id, scenario_id, "curious", ["together"]) wait_for( lambda: all( item.get("scenario_kink", {}).get("id") != scenario_id for item in get_json( f"/users/{left.user_id}/partner-groups/{urllib.parse.quote(group_id)}/overlap", headers=auth_headers(left), ).get("scenario_matches", []) ), timeout_s=30.0, label="shared scenario downgrade propagation", ) save_scenario(right, parent_id, scenario_id, "like", ["together"]) wait_for( lambda: any( item.get("scenario_kink", {}).get("id") == scenario_id for item in get_json( f"/users/{left.user_id}/partner-groups/{urllib.parse.quote(group_id)}/overlap", headers=auth_headers(left), ).get("scenario_matches", []) ), timeout_s=30.0, label="shared scenario restore propagation", ) checkpoints.append( { "stage": "scenario_strength_gate", "expected": "shared scenarios disappear when one partner downgrades to Maybe and return on Like", "actual": {"scenario_id": scenario_id}, "summary": "shared scenario respects love/like-only matching", } ) open_together(left_page) left_page.get_by_test_id("together-tab-explore").click() expect(left_page.get_by_test_id("together-explore-panel")).to_be_visible(timeout=15000) expect(left_page.get_by_test_id("together-explore-related-section")).to_be_visible(timeout=15000) expect(left_page.get_by_test_id("together-explore-scenarios-section")).to_be_visible(timeout=15000) expect(left_page.get_by_test_id(f"together-explore-scenario-{related_scenario_id}")).to_be_visible(timeout=15000) checkpoints.append( { "stage": "explore_visible", "expected": "Together > Worth Exploring shows related but not identical likes and connected scenarios", "actual": { "left_name": "Oral Sex", "right_name": right_related_name, "related_scenario_id": related_scenario_id, }, "summary": "related-match and connected-scenario sections rendered", } ) left_page.get_by_test_id("together-tab-theirs").click() expect(left_page.get_by_test_id("together-their-list-panel")).to_be_visible(timeout=15000) expect(left_page.get_by_text(private_only_name)).to_be_visible(timeout=15000) checkpoints.append( { "stage": "their_list_visible", "expected": "Together > Their List shows only after sharing is enabled", "actual": {"visible_partner_item": private_only_name}, "summary": "their-list tab showed the partner-only item", } ) save_play(left, direct_id, "hard_no", []) wait_for( lambda: all( item.get("id") != direct_id for item in get_json( f"/users/{left.user_id}/partner-groups/{urllib.parse.quote(group_id)}/overlap", headers=auth_headers(left), ).get("direct_matches", []) ), timeout_s=30.0, label="shared removal propagation", ) save_play(left, direct_id, "like", ["to_me"]) wait_for( lambda: any( item.get("id") == direct_id for item in get_json( f"/users/{left.user_id}/partner-groups/{urllib.parse.quote(group_id)}/overlap", headers=auth_headers(left), ).get("direct_matches", []) ), timeout_s=30.0, label="shared restore propagation", ) delete_json(f"/users/{left.user_id}/plays/{urllib.parse.quote(direct_id)}", headers=auth_headers(left)) wait_for( lambda: all( item.get("id") != direct_id for item in get_json( f"/users/{left.user_id}/partner-groups/{urllib.parse.quote(group_id)}/overlap", headers=auth_headers(left), ).get("direct_matches", []) ), timeout_s=30.0, label="shared delete propagation", ) checkpoints.append( { "stage": "post_link_mutations", "expected": "changing, restoring, and deleting a shared play updates Together deterministically", "actual": {"mutated_kink": "Bondage"}, "summary": "post-link add/change/remove flow updated the shared list", } ) return { "summary": "share-link, link acceptance, shared matches, related matches, scenarios, privacy, and post-link mutations all behaved as expected", "left_user_id": left.user_id, "right_user_id": right.user_id, "invite_url": invite_url, "group_id": group_id, "left_play_count": len(get_user(left).get("plays", {})), "right_play_count": len(get_user(right).get("plays", {})), "related_right_name": right_related_name, "their_list_private_name": private_only_name, "scenario_parent": parent_name, "scenario_id": scenario_id, "checkpoints": checkpoints, } finally: left_context.close() right_context.close() def scenario_couple_adversarial(browser: Browser) -> dict[str, object]: """Couple flow with many kinks, conflicts, and the Same-Plays / My-Plays similar-tentative-add path. Adds adversarial coverage on top of ``scenario_couple_audit``: * both partners seeded past 80 likes (recommender stress + bigger overlap pool) * curated love/love overlap of ~9 known-name kinks → asserted in Together > Shared * one love(left)/hard_no(right) conflict → must be ABSENT from Same Plays * clicking a Same Plays chip lands on the My Plays inspector for that kink * clicking a 'Similar' chip in My Plays tentatively adds the similar kink as ``curious`` with the parent's directions, then opens it in the inspector * ``visibleSimilarItems`` invariant: no scenario kinks in the Similar chip set """ overlap_targets = [ ("kissing", "Kissing"), ("massage", "Massage"), ("oral sex", "Oral Sex"), ("anal", "Anal"), ("spanking", "Spanking"), ("blindfolds", "Blindfolds"), ("role play", "Role Play"), ("hair pulling", "Hair Pulling"), ("cuddling", "Cuddling"), ] conflict_candidates = [("biting", "Biting"), ("hickeys", "Hickeys"), ("tickling", "Tickling")] right_only_options = [("foot worship", "Foot Worship"), ("lingerie", "Lingerie"), ("rope bondage", "Rope Bondage")] left = create_user() right = create_user() seed_user_with_recommendations(left, 80) seed_user_with_recommendations(right, 80) overlap_pairs: list[tuple[str, str]] = [] for query, expected in overlap_targets: try: kink_id = find_kink_id(query, expected) except AssertionError: continue save_play(left, kink_id, "love", ["together"]) save_play(right, kink_id, "love", ["together"]) overlap_pairs.append((kink_id, expected)) if len(overlap_pairs) < 5: raise AssertionError(f"Could not seed enough overlap kinks: only {len(overlap_pairs)} matched") overlap_id_set = {kid for kid, _ in overlap_pairs} conflict_pair: tuple[str, str] | None = None for query, expected in conflict_candidates: try: kink_id = find_kink_id(query, expected) except AssertionError: continue if kink_id in overlap_id_set: continue save_play(left, kink_id, "love", ["together"]) save_play(right, kink_id, "hard_no", []) conflict_pair = (kink_id, expected) break if conflict_pair is None: raise AssertionError("Could not seed a love/hard_no conflict pair (no candidate matched the catalog)") right_only_pair: tuple[str, str] | None = None for query, expected in right_only_options: try: kink_id = find_kink_id(query, expected) except AssertionError: continue if kink_id in overlap_id_set or kink_id == conflict_pair[0]: continue ensure_play_absent(left, kink_id) save_play(right, kink_id, "love", ["together"]) right_only_pair = (kink_id, expected) break if right_only_pair is None: raise AssertionError("Could not seed a right-only kink for Their List coverage") parent_id, parent_name, scenario_a, scenario_b = find_parent_with_scenario_pair() save_play(left, parent_id, "like", ["together"]) save_play(right, parent_id, "like", ["together"]) save_scenario(left, parent_id, scenario_a, "like", ["together"]) save_scenario(right, parent_id, scenario_a, "like", ["together"]) save_scenario(right, parent_id, scenario_b, "like", ["together"]) send_partner_request(right, left.user_id) wait_for( lambda: right.user_id in get_user(left).get("incoming_partner_requests", []), label="left incoming partner request", ) accept_partner_request(left, right.user_id) wait_for( lambda: right.user_id in get_user(left).get("partners", []), label="linked partners", ) group_id = group_id_for_pair(left, right.user_id) toggle_share(right, group_id, True) wait_for( lambda: right.user_id in next((g for g in list_groups(left) if g["id"] == group_id), {}).get("sharing_member_ids", []), label="right sharing toggle", ) overlap_path = f"/users/{left.user_id}/partner-groups/{urllib.parse.quote(group_id)}/overlap" wait_for( lambda: {item.get("id") for item in get_json(overlap_path, headers=auth_headers(left)).get("direct_matches", [])} >= overlap_id_set, timeout_s=45.0, label="all curated overlap kinks visible in Same Plays", ) overlap_payload = get_json(overlap_path, headers=auth_headers(left)) direct_ids = {item.get("id") for item in overlap_payload.get("direct_matches", [])} missing_overlap = sorted(overlap_id_set - direct_ids) if missing_overlap: raise AssertionError(f"Same Plays missing curated kinks: {missing_overlap}") if conflict_pair[0] in direct_ids: raise AssertionError(f"Conflict (love/hard_no) kink {conflict_pair[1]!r} leaked into Same Plays") scenario_match_ids = { item.get("scenario_kink", {}).get("id") for item in overlap_payload.get("scenario_matches", []) } if scenario_a not in scenario_match_ids: raise AssertionError("Shared scenario A missing from scenario_matches") checkpoints: list[dict[str, object]] = [ { "stage": "seed_and_overlap", "expected": "both users seeded past 80 likes; curated overlap appears in Same Plays; love/hard_no conflict excluded", "actual": { "left_play_count": len(get_user(left).get("plays", {})), "right_play_count": len(get_user(right).get("plays", {})), "overlap_count_api": len(direct_ids), "curated_overlap": [name for _, name in overlap_pairs], "conflict_excluded": conflict_pair[1], "shared_scenario": scenario_a, }, "summary": "API overlap matches curated love/love and excludes love/hard_no conflict", } ] left_context = browser.new_context( viewport={"width": 1280, "height": 720}, extra_http_headers={"Cache-Control": "no-cache"}, ) page = left_context.new_page() try: login_existing_user(page, left) expect(page.get_by_test_id("discover-view")).to_be_visible(timeout=15000) open_together(page) expect(page.get_by_test_id(f"together-group-chip-{group_id}")).to_be_visible(timeout=15000) page.get_by_test_id(f"together-group-chip-{group_id}").click() expect(page.get_by_test_id("together-shared-panel")).to_be_visible(timeout=15000) expect(page.get_by_test_id("together-shared-direct-section")).to_be_visible(timeout=15000) for kink_id, name in overlap_pairs: expect(page.get_by_test_id(f"together-shared-direct-{kink_id}")).to_be_visible(timeout=20000) if page.get_by_test_id(f"together-shared-direct-{conflict_pair[0]}").count(): raise AssertionError(f"Conflict chip rendered in Same Plays for {conflict_pair[1]!r}") expect(page.get_by_test_id(f"together-shared-scenario-{scenario_a}")).to_be_visible(timeout=15000) checkpoints.append( { "stage": "shared_panel_render", "expected": "every curated overlap kink renders as a Same Plays chip; conflict chip absent; shared scenario chip visible", "actual": { "overlap_rendered": [name for _, name in overlap_pairs], "conflict_absent": conflict_pair[1], "scenario_visible": scenario_a, }, "summary": "Same Plays UI matches API overlap; love/hard_no conflict not shown", } ) focus_kink_id, focus_kink_name = overlap_pairs[0] page.get_by_test_id(f"together-shared-direct-{focus_kink_id}").click() expect(page.get_by_test_id("together-view")).to_be_visible(timeout=15000) inline_inspector = page.get_by_test_id(f"together-shared-direct-inspector-{focus_kink_id}") expect(inline_inspector).to_be_visible(timeout=15000) expect(inline_inspector.locator(".plays-inspector-title")).to_have_text(focus_kink_name, timeout=15000) if page.get_by_test_id("plays-view").count(): raise AssertionError("Same Plays chip should expand inline, not route to plays-view") checkpoints.append( { "stage": "same_plays_click_expands_inline", "expected": "clicking a Same Plays chip expands the inspector inline inside Together → Shared (no view switch)", "actual": {"focused_kink_id": focus_kink_id, "focused_kink_name": focus_kink_name}, "summary": "Same Plays chip expanded inline with the inspector body; stayed on Together", } ) similar_chips = inline_inspector.locator(".chips button.chip") try: similar_chips.first.wait_for(state="visible", timeout=20000) except Exception as exc: raise AssertionError(f"No Similar chips appeared inline for {focus_kink_name!r}: {exc}") from exc chip_count = similar_chips.count() chip_labels = [similar_chips.nth(i).inner_text().strip() for i in range(chip_count)] first_similar_label = chip_labels[0] plays_before = set(get_user(left).get("plays", {}).keys()) with page.expect_response( lambda res: res.request.method == "POST" and "/plays" in res.url, timeout=45000, ) as similar_resp: similar_chips.first.click() body = similar_resp.value if not body.ok: raise AssertionError(f"Tentative add POST /plays failed: status={body.status}") wait_for( lambda: bool(set(get_user(left).get("plays", {}).keys()) - plays_before), timeout_s=15.0, label="tentative add visible in user plays", ) plays_after = get_user(left).get("plays", {}) new_ids = sorted(set(plays_after.keys()) - plays_before) if len(new_ids) != 1: raise AssertionError(f"Expected exactly one new play after similar click, got {new_ids}") new_kink_id = new_ids[0] new_play = plays_after[new_kink_id] if new_play.get("interest_state") != "curious": raise AssertionError(f"Tentative add wrong state: {new_play.get('interest_state')!r}") if "together" not in (new_play.get("directions") or []): raise AssertionError(f"Tentative add did not inherit parent direction 'together': {new_play.get('directions')}") new_kink_detail = get_json(f"/kinks/{urllib.parse.quote(new_kink_id)}") if new_kink_detail.get("is_scenario"): raise AssertionError("Similar chip surfaced a scenario kink (visibleSimilarItems should filter is_scenario)") # The chip is now expanded on the new kink; expect title to update inline. new_inline_inspector = page.get_by_test_id(f"together-shared-direct-inspector-{new_kink_id}") expect(new_inline_inspector).to_be_visible(timeout=15000) expect(new_inline_inspector.locator(".plays-inspector-title")).to_have_text( new_kink_detail.get("name", first_similar_label), timeout=15000 ) expect(new_inline_inspector.get_by_test_id("plays-inspector-tentative")).to_be_visible(timeout=10000) scenario_chips = [label for label in chip_labels if label.lower().startswith("scenario:")] if scenario_chips: raise AssertionError(f"Similar chip set contained scenario-prefixed entries: {scenario_chips}") checkpoints.append( { "stage": "shared_inline_similar_tentative_add", "expected": "clicking a Similar chip on the inline Same-Plays inspector POSTs /plays as 'curious' with parent directions, re-inspects the new kink inline, and shows a Tentative badge", "actual": { "parent_kink_id": focus_kink_id, "parent_kink_name": focus_kink_name, "similar_chip_count": chip_count, "tentative_kink_id": new_kink_id, "tentative_kink_name": new_kink_detail.get("name"), "tentative_state": new_play.get("interest_state"), "tentative_directions": new_play.get("directions"), "similar_is_scenario": bool(new_kink_detail.get("is_scenario")), "chip_labels_sample": chip_labels[:6], }, "summary": "Similar-chip click tentatively saved as curious/together inline and showed the Tentative badge", } ) # Undo the tentative add via the new Undo button; the kink should leave the user's plays. undo_btn = new_inline_inspector.get_by_test_id("plays-inspector-undo") expect(undo_btn).to_be_visible(timeout=10000) with page.expect_response( lambda res: res.request.method == "DELETE" and f"/plays/{urllib.parse.quote(new_kink_id)}" in res.url, timeout=45000, ) as delete_resp: undo_btn.click() if not delete_resp.value.ok: raise AssertionError(f"Undo DELETE /plays failed: status={delete_resp.value.status}") wait_for( lambda: new_kink_id not in get_user(left).get("plays", {}), timeout_s=15.0, label="undo removes tentative play", ) checkpoints.append( { "stage": "shared_inline_undo_tentative", "expected": "Undo on a tentative add removes the play row via DELETE /plays", "actual": {"undone_kink_id": new_kink_id, "undone_kink_name": new_kink_detail.get("name")}, "summary": "Undo button removed the tentative play; user state matches pre-click", } ) open_together(page) page.get_by_test_id("together-tab-explore").click() expect(page.get_by_test_id("together-explore-panel")).to_be_visible(timeout=15000) page.get_by_test_id("together-tab-theirs").click() expect(page.get_by_test_id("together-their-list-panel")).to_be_visible(timeout=15000) expect(page.get_by_text(right_only_pair[1], exact=True).first).to_be_visible(timeout=15000) checkpoints.append( { "stage": "explore_and_theirs_render", "expected": "Worth Exploring + Their List populate; right-only kink visible after sharing", "actual": {"right_only_visible": right_only_pair[1]}, "summary": "Explore and Their List subtabs both render and surface the partner-only item", } ) return { "summary": "adversarial couple flow: many likes, curated overlap, conflict suppression, Same-Plays inline expand, Similar tentative-add + Undo, and Worth Exploring / Their List all behaved as expected", "left_user_id": left.user_id, "right_user_id": right.user_id, "group_id": group_id, "left_play_count_final": len(plays_after), "right_play_count_final": len(get_user(right).get("plays", {})), "overlap_kinks": [name for _, name in overlap_pairs], "conflict_kink": conflict_pair[1], "right_only_kink": right_only_pair[1], "shared_scenario_id": scenario_a, "tentative_kink": new_kink_detail.get("name"), "tentative_kink_id": new_kink_id, "checkpoints": checkpoints, } finally: left_context.close() # Maps a `direction_shape` to a substring expected to appear in the discover-direction-strip. # Mutual hides pills entirely and shows a caption. Other shapes render shape-specific labels. _SHAPE_FINGERPRINTS: dict[str, list[str]] = { "mutual": ["This is mutual by default"], "wearable": ["I wear it", "Partner wears it"], "tool_wielded": ["I use it on them", "Used on me"], "tool_worn": ["I wear one", "Partner wears one"], "state": ["I have it", "Partner has it"], # role uses dynamic role names — match the literal "I'm " prefix from the SHAPE_LABELS dict. "role": ["I'm ", "We switch"], } def scenario_discover_shape_relabel(page: Page) -> dict[str, object]: """Verify the direction-shape classifier surfaces correct pill labels in Discover. Walk the Discover deck rating cards `curious` until we land on a card whose direction strip shows non-action pills (or the mutual caption). Assert the rendered text matches one of the known shape fingerprints. Caps at ~30 advances so the recommendation order randomness can't make the scenario hang. """ creds = create_user() seed_user_with_recommendations(creds, 12) login_existing_user(page, creds) expect(page.get_by_test_id("discover-view")).to_be_visible(timeout=15_000) matched_shape: str | None = None matched_strip_text: str | None = None matched_kink_name: str | None = None advances = 0 max_advances = 30 while advances < max_advances: # Wait for the direction strip to be present (rendered after detail loads). strip = page.locator('[data-testid="discover-direction-strip"]').first try: strip.wait_for(state="attached", timeout=8_000) except Exception: break text = (strip.inner_text() or "").strip() for shape, fingerprints in _SHAPE_FINGERPRINTS.items(): if all(fp.lower() in text.lower() for fp in fingerprints): matched_shape = shape matched_strip_text = text title = page.locator(".tinder-card .card-title").first try: matched_kink_name = title.inner_text() except Exception: matched_kink_name = None break if matched_shape: break # Advance: rate `curious` and let the deck progress. rate_btn = page.locator('[data-testid="discover-rate-curious"]').first if rate_btn.count() == 0: break rate_btn.click() page.wait_for_timeout(450) advances += 1 if matched_shape is None: raise AssertionError( f"discover_shape_relabel: walked {advances} cards without seeing a non-action shape; " f"the catalog should hit at least one mutual/wearable/tool/role/state card." ) return { "summary": f"discover renders shape-specific direction labels (matched shape={matched_shape!r})", "user_id": creds.user_id, "matched_shape": matched_shape, "matched_kink_name": matched_kink_name, "matched_strip_text": (matched_strip_text or "").split("\n")[0][:120], "advances": advances, } SCENARIOS: dict[str, Callable[[Browser], dict[str, object]]] = { "onboarding": lambda browser: _run_in_fresh_context(browser, scenario_onboarding_create_button), "starter": lambda browser: _run_in_fresh_context(browser, scenario_starter_gate), "settings": lambda browser: _run_in_fresh_context(browser, scenario_settings_controls), "couple": scenario_couple_audit, "couple_adversarial": scenario_couple_adversarial, "discover_shape_relabel": lambda browser: _run_in_fresh_context(browser, scenario_discover_shape_relabel), } def normalize_scenarios(names: list[str]) -> list[str]: tokens: list[str] = [] for name in names: tokens.extend(part.strip() for part in name.split(",") if part.strip()) if not tokens or tokens == ["all"]: return DEFAULT_SCENARIOS.copy() out: list[str] = [] for name in tokens: if name not in SCENARIOS: raise AssertionError(f"Unknown scenario: {name}") if name not in out: out.append(name) return out def run_selected_scenarios(selected: list[str], *, headless: bool = True) -> list[StageResult]: results: list[StageResult] = [] with sync_playwright() as playwright: launch_kwargs: dict[str, object] = { "headless": headless, # Shared HTTP cache across fresh contexts caused empty starter decks while urllib saw recs (Chromium). "args": ["--no-sandbox", "--disable-http-cache"], } channel = os.environ.get("KINK_PLAYWRIGHT_CHANNEL", "").strip() if channel: launch_kwargs["channel"] = channel browser = playwright.chromium.launch(**launch_kwargs) try: for name in normalize_scenarios(selected): results.append(_run_stage(name, lambda scenario_name=name: SCENARIOS[scenario_name](browser))) finally: browser.close() return results def results_to_json(results: list[StageResult]) -> dict[str, object]: return { "ok": all(result.ok for result in results), "results": [asdict(result) for result in results], "summary": render_summary(results), }