jacuzzi/tests/smoke_search_history.py

#!/usr/bin/env python3
"""Searches are kept, and offered back under the box as you type.

Run against a locally running backend:

    backend/main.py &
    .venv/bin/python tests/smoke_search_history.py

The listing is a fixture and records what it was asked for, so "the search
actually ran" is something the test can see rather than infer.
"""
import json
import sys
from playwright.sync_api import sync_playwright

BASE = "http://127.0.0.1:5000/"
SERVER = "https://hottubapp.io"
CHANNEL = "xvideos"

SEED = """([server, channel]) => {
    localStorage.setItem('config', JSON.stringify({ servers: [{ [server]: {} }] }));
    localStorage.setItem('preferences', JSON.stringify({ [server]: { channelId: channel } }));
    localStorage.removeItem('session');
    localStorage.removeItem('searchHistory');
}"""

HISTORY = """() => JSON.parse(localStorage.getItem('searchHistory') || '[]')"""

ROWS = """() => Array.from(document.querySelectorAll('#search-suggest .search-suggest-row'))
    .map((row) => row.dataset.query)"""

OPEN = """() => !document.getElementById('search-suggest').hidden"""

ACTIVE = """() => {
    const row = document.querySelector('#search-suggest .search-suggest-row.is-active');
    return row ? row.dataset.query : null;
}"""


class Checks:
    def __init__(self):
        self.failed = 0

    def ok(self, label, condition, detail=""):
        mark = "PASS" if condition else "FAIL"
        if not condition:
            self.failed += 1
        print(f"  [{mark}] {label}" + (f" -- {detail}" if detail and not condition else ""))


def items(query):
    return [{"id": f"fix-{query}-{i}", "title": f"{query or 'anything'} {i}",
             "url": f"https://fixture.test/v{i}.mp4", "thumb": "", "channel": "test",
             "duration": 1, "isLive": False, "tags": []} for i in range(12)]


def wait_for_query(page, asked, query, ms=4000):
    """The search may go out on the request already in flight rather than a new
    one of its own (the box searches as you type, so there usually is one), so
    what is waited for is the query reaching the server, not a fresh request."""
    for _ in range(ms // 100):
        if query in asked:
            return True
        page.wait_for_timeout(100)
    return query in asked


def type_into(page, text):
    """Types a query the way a reader would, a character at a time."""
    page.click("#search-input")
    page.fill("#search-input", "")
    page.type("#search-input", text, delay=40)


def main():
    c = Checks()
    asked = []

    with sync_playwright() as p:
        browser = p.chromium.launch(args=[
            "--no-sandbox",
            "--disable-dev-shm-usage",
            "--disable-gpu",
            "--renderer-process-limit=1",
            "--js-flags=--max-old-space-size=512",
        ])
        page = browser.new_page(viewport={"width": 1100, "height": 900})
        crashes = []
        page.on("pageerror", lambda e: crashes.append(str(e)))

        # With `hold` on, listing requests are recorded but left unanswered, so
        # the test can act while one is still out.
        hold = {"on": False, "parked": []}

        def answer(route, query):
            route.fulfill(status=200, content_type="application/json", body=json.dumps(
                {"items": items(query), "pageInfo": {"hasNextPage": False}}))

        def listing(route):
            try:
                body = json.loads(route.request.post_data or "{}")
            except Exception:
                body = {}
            query = body.get("query", "")
            asked.append(query)
            if hold["on"]:
                hold["parked"].append((route, query))
                return
            answer(route, query)

        def release():
            hold["on"] = False
            while hold["parked"]:
                route, query = hold["parked"].pop(0)
                answer(route, query)

        page.route("**/api/videos", listing)
        page.goto(BASE, wait_until="domcontentloaded")
        page.evaluate(SEED, [SERVER, CHANNEL])
        page.goto(BASE, wait_until="load")
        page.wait_for_selector(".video-card", timeout=30000)
        page.wait_for_timeout(500)

        print("\nsearching for something")
        type_into(page, "hotel")
        page.wait_for_timeout(1200)
        stored = page.evaluate(HISTORY)
        c.ok("the search was kept", "hotel" in stored, str(stored))
        c.ok("and the typing on the way to it was not",
             stored == ["hotel"], str(stored))
        c.ok("the search itself ran", asked and asked[-1] == "hotel", str(asked[-3:]))

        print("\nand for something else, then reloading")
        type_into(page, "sauna")
        page.wait_for_timeout(1200)
        page.goto(BASE, wait_until="load")
        page.wait_for_selector(".video-card", timeout=30000)
        stored = page.evaluate(HISTORY)
        c.ok("both searches survived the reload",
             stored == ["sauna", "hotel"], str(stored))

        print("\nwith a history to draw on")
        page.evaluate("""() => localStorage.setItem('searchHistory', JSON.stringify([
            'hot tub', 'hotel room', 'shot in the dark', 'sauna', 'steam room',
            'hot spring', 'cold plunge'
        ]))""")
        page.click("#search-input")
        page.wait_for_timeout(300)
        rows = page.evaluate(ROWS)
        c.ok("the box offers the recent ones when empty", page.evaluate(OPEN) is True)
        c.ok("five of them, newest first",
             rows == ['hot tub', 'hotel room', 'shot in the dark', 'sauna', 'steam room'], str(rows))

        print("\ntyping narrows them down")
        page.type("#search-input", "hot", delay=40)
        page.wait_for_timeout(200)
        rows = page.evaluate(ROWS)
        print(f"  (offered: {rows})")
        c.ok("only the ones it appears in are left",
             set(rows) == {'hot tub', 'hotel room', 'hot spring', 'shot in the dark'}, str(rows))
        c.ok("the ones it starts come first",
             rows[:3] == ['hot tub', 'hotel room', 'hot spring'], str(rows))
        c.ok("and the one it is buried in comes last",
             rows[-1] == 'shot in the dark', str(rows))
        c.ok("the part that matched is picked out",
             page.evaluate("""() => {
                 const hit = document.querySelector('#search-suggest .search-suggest-hit');
                 return !!hit && hit.textContent.toLowerCase() === 'hot';
             }"""))

        print("\nwhat is already typed in full is not a suggestion")
        page.fill("#search-input", "")
        page.type("#search-input", "sauna", delay=20)
        page.wait_for_timeout(200)
        rows = page.evaluate(ROWS)
        c.ok("it offers itself back to nobody",
             'sauna' not in rows and page.evaluate(OPEN) is False, str(rows))

        print("\ntaking one with the keyboard")
        page.fill("#search-input", "")
        page.type("#search-input", "hot", delay=20)
        page.wait_for_timeout(200)
        page.keyboard.press("ArrowDown")
        first = page.evaluate(ACTIVE)
        page.keyboard.press("ArrowDown")
        second = page.evaluate(ACTIVE)
        c.ok("the arrows walk the list", first == 'hot tub' and second == 'hotel room',
             f"{first} then {second}")
        asked.clear()
        page.keyboard.press("Enter")
        page.wait_for_timeout(300)
        c.ok("enter takes the one highlighted",
             page.evaluate("() => document.getElementById('search-input').value") == 'hotel room')
        c.ok("and searches for it", wait_for_query(page, asked, 'hotel room'), str(asked))
        c.ok("the list is put away", page.evaluate(OPEN) is False)

        print("\ntaking one with the pointer")
        page.fill("#search-input", "")
        page.click("#search-input")
        page.type("#search-input", "st", delay=20)
        page.wait_for_timeout(250)
        asked.clear()
        rows = page.evaluate(ROWS)
        page.click(f"#search-suggest .search-suggest-row:nth-of-type(1)")
        page.wait_for_timeout(300)
        c.ok("the one clicked is what gets searched for",
             page.evaluate("() => document.getElementById('search-input').value") == rows[0]
             and wait_for_query(page, asked, rows[0]), f"{rows} then {asked}")

        print("\nescape")
        page.fill("#search-input", "")
        page.click("#search-input")
        page.wait_for_timeout(250)
        c.ok("the list is up again", page.evaluate(OPEN) is True)
        page.keyboard.press("Escape")
        page.wait_for_timeout(150)
        c.ok("escape puts it away", page.evaluate(OPEN) is False)

        print("\nchoosing one while the box's own search is still out")
        # The box searches as you type: choose a suggestion at the wrong moment
        # and the search behind it used to be dropped, leaving the results on
        # the query before it.
        page.fill("#search-input", "")
        page.click("#search-input")
        hold["on"] = True
        asked.clear()
        page.type("#search-input", "hot", delay=20)
        page.wait_for_timeout(700)       # long enough for the typed search to go out
        c.ok("the typed search is on the wire", asked == ['hot'], str(asked))
        page.keyboard.press("ArrowDown")
        chosen = page.evaluate(ACTIVE)
        page.keyboard.press("Enter")
        landed = wait_for_query(page, asked, chosen, 3000)
        release()
        page.wait_for_timeout(800)
        c.ok(f"the chosen one ({chosen}) is searched for all the same", landed, str(asked))
        c.ok("and its results are what the grid ends up with",
             page.evaluate("() => (App.state.loadedVideos[0] || {}).id || ''")
             .startswith(f"fix-{chosen}"),
             page.evaluate("() => (App.state.loadedVideos[0] || {}).id || ''"))

        c.ok("nothing threw along the way", not crashes, str(crashes[:2]))
        browser.close()

    print(f"\n{'OK' if not c.failed else 'FAILED'}: {c.failed} check(s) failed")
    return 1 if c.failed else 0


sys.exit(main())