scripts/e2e_ui.py

"""Drive the Gradio chat UI end-to-end with Playwright (real conversation)."""

import asyncio
import sys

from playwright.async_api import async_playwright

URL = "http://127.0.0.1:7860"

TURNS = [
    "Hallo, help me energieprijzen vergelijken",
    "9000 Gent",
    "Voor mijn woning",
    "Elektriciteit en gas allebei",
    "Nee, ik ken mijn verbruik niet",
    "Ja, een digitale meter",
    "Enkelvoudig tarief",
    "We wonen met 3 personen",
    "Nee, geen warmtepomp, geen elektrische wagen, geen zonnepanelen en geen thuisbatterij",
    "Ja, dien maar in!",
]


async def last_bot_message(page) -> str:
    return await page.evaluate(
        """() => {
            const msgs = document.querySelectorAll('.bot, [data-testid="bot"]');
            const last = msgs[msgs.length - 1];
            return last ? last.innerText.trim() : '';
        }"""
    )


async def wait_for_reply(page, prev: str, timeout_s: int = 120) -> str:
    for _ in range(timeout_s * 2):
        await asyncio.sleep(0.5)
        cur = await last_bot_message(page)
        if cur and cur != prev:
            busy = await page.evaluate(
                """() => !!document.querySelector('.generating, [class*="pending"]')"""
            )
            if not busy:
                await asyncio.sleep(1.0)
                settled = await last_bot_message(page)
                if settled == cur:
                    return cur
    raise TimeoutError("no reply within timeout")


async def main() -> None:
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        await page.goto(URL)
        box = page.locator('textarea').first
        await box.wait_for(timeout=15_000)
        prev = ""
        for turn in TURNS:
            print(f"\n>>> USER: {turn}")
            await box.fill(turn)
            await box.press("Enter")
            try:
                prev = await wait_for_reply(page, prev)
            except TimeoutError:
                print("!!! TIMEOUT waiting for reply")
                sys.exit(1)
            print(f"<<< AGENT: {prev}")
        await browser.close()


if __name__ == "__main__":
    asyncio.run(main())