UI tests with Playwright

.github/actions/setup-env/action.yml

@@ -58,6 +58,10 @@ runs:
       shell: bash -el {0}
       run: mamba install --yes --channel conda-forge redis-server
 
+    - name: Install playwright
+      shell: bash -el {0}
+      run: playwright install
+
     - name: Install ragna
       shell: bash -el {0}
       run: |

.github/workflows/test.yml

@@ -68,7 +68,19 @@ jobs:
 
       - name: Run unit tests
         id: tests
-        run: pytest --junit-xml=test-results.xml --durations=25
+        run: |
+          pytest \
+          --junit-xml=test-results.xml \
+          --durations=25 \
+          --video=retain-on-failure  # playwright ui tests
+
+      - name: Upload playwright video
+        if: failure()
+        uses: actions/upload-artifact@v4
+        with:
+          name:
+            playwright-${{ matrix.os }}-${{ matrix.python-version}}-${{ github.run_id }}
+          path: test-results
 
       - name: Surface failing tests
         if: steps.tests.outcome != 'success'

environment-dev.yml

@@ -10,6 +10,7 @@ dependencies:
       - pytest >=6
       - pytest-mock
       - pytest-asyncio
+      - pytest-playwright
       - mypy ==1.10.0
       - pre-commit
       - types-aiofiles

tests/deploy/api/test_batch_endpoints.py

@@ -4,7 +4,7 @@
 from ragna.deploy import Config
 from ragna.deploy._api import app
 
-from .utils import authenticate
+from ..utils import authenticate
 
 
 def test_batch_sequential_upload_equivalence(tmp_local_root):

tests/deploy/api/test_components.py

@@ -7,7 +7,7 @@
 from ragna.deploy import Config
 from ragna.deploy._api import app
 
-from .utils import authenticate
+from ..utils import authenticate
 
 
 @pytest.mark.parametrize("ignore_unavailable_components", [True, False])

tests/deploy/api/test_e2e.py

@@ -1,29 +1,12 @@
 import json
-import time
 
 import pytest
 from fastapi.testclient import TestClient
 
-from ragna.assistants import RagnaDemoAssistant
 from ragna.deploy import Config
 from ragna.deploy._api import app
 
-from .utils import authenticate
-
-
-class TestAssistant(RagnaDemoAssistant):
-    def answer(self, prompt, sources, *, multiple_answer_chunks: bool):
-        # Simulate a "real" assistant through a small delay. See
-        # https://github.com/Quansight/ragna/pull/401#issuecomment-2095851440
-        # for why this is needed.
-        time.sleep(1e-3)
-        content = next(super().answer(prompt, sources))
-
-        if multiple_answer_chunks:
-            for chunk in content.split(" "):
-                yield f"{chunk} "
-        else:
-            yield content
+from ..utils import TestAssistant, authenticate
 
 
 @pytest.mark.parametrize("multiple_answer_chunks", [True, False])

tests/deploy/ui/test_ui.py

@@ -0,0 +1,174 @@
+import socket
+import subprocess
+import sys
+import time
+
+import httpx
+import panel as pn
+import pytest
+from playwright.sync_api import Page, expect
+
+from ragna._utils import timeout_after
+from ragna.deploy import Config
+
+from ..utils import TestAssistant
+
+
+def get_available_port():
+    with socket.socket() as s:
+        s.bind(("", 0))
+        return s.getsockname()[1]
+
+
+@pytest.fixture(scope="session")
+def api_port():
+    return get_available_port()
+
+
+@pytest.fixture(scope="session")
+def ui_port():
+    return get_available_port()
+
+
+@pytest.fixture(scope="session")
+def headed_mode(pytestconfig):
+    return pytestconfig.getoption("headed") or False
+
+
+@pytest.fixture
+def config(
+    tmp_local_root,
+):
+    config = Config(
+        local_root=tmp_local_root,
+        assistants=[TestAssistant],
+        ui=dict(port=get_available_port()),
+        api=dict(port=get_available_port()),
+    )
+    path = tmp_local_root / "ragna.toml"
+    config.to_file(path)
+    return config
+
+
+class Server:
+    def __init__(self, config):
+        self.config = config
+        self.base_url = f"http://{config.ui.hostname}:{config.ui.port}"
+
+    def server_up(self):
+        try:
+            return httpx.get(self.base_url).is_success
+        except httpx.ConnectError:
+            return False
+
+    @timeout_after(60)
+    def start(self):
+        self.proc = subprocess.Popen(
+            [
+                sys.executable,
+                "-m",
+                "ragna",
+                "ui",
+                "--config",
+                self.config.local_root / "ragna.toml",
+                "--start-api",
+                "--ignore-unavailable-components",
+                "--no-open-browser",
+            ],
+            stdout=sys.stdout,
+            stderr=sys.stderr,
+        )
+
+        while not self.server_up():
+            time.sleep(1)
+
+    def stop(self):
+        self.proc.kill()
+        pn.state.kill_all_servers()
+
+
+@pytest.fixture
+def server(config):
+    server = Server(config)
+    try:
+        server.start()
+        yield server
+    finally:
+        server.stop()
+
+
+def test_health(server, page: Page) -> None:
+    health_url = server.base_url + "/health"
+    response = page.goto(health_url)
+    assert response.ok
+
+
+def test_index(server, config, page: Page) -> None:
+    # Index page, no auth
+    index_url = server.base_url
+    page.goto(index_url)
+    expect(page.get_by_role("button", name="Sign In")).to_be_visible()
+
+    # Authorize with no credentials
+    page.get_by_role("button", name="Sign In").click()
+    expect(page.get_by_role("button", name=" New Chat")).to_be_visible()
+
+    # expect auth token to be set
+    cookies = page.context.cookies()
+    assert len(cookies) == 1
+    cookie = cookies[0]
+    assert cookie.get("name") == "auth_token"
+    auth_token = cookie.get("value")
+    assert auth_token is not None
+
+    # New page button
+    new_chat_button = page.get_by_role("button", name=" New Chat")
+    expect(new_chat_button).to_be_visible()
+    new_chat_button.click()
+
+    document_root = config.local_root / "documents"
+    document_root.mkdir()
+    document_name = "test.txt"
+    document_path = document_root / document_name
+    with open(document_path, "w") as file:
+        file.write("!\n")
+
+    # File upload selector
+    with page.expect_file_chooser() as fc_info:
+        page.locator(".fileUpload").click()
+    file_chooser = fc_info.value
+    file_chooser.set_files(document_path)
+
+    # Upload document and expect to see it listed
+    file_list = page.locator(".fileListContainer")
+    expect(file_list.first).to_have_text(str(document_name))
+
+    chat_dialog = page.get_by_role("dialog")
+    expect(chat_dialog).to_be_visible()
+    start_chat_button = page.get_by_role("button", name="Start Conversation")
+    expect(start_chat_button).to_be_visible()
+    time.sleep(0.5)  # hack while waiting for button to be fully clickable
+    start_chat_button.click(delay=5)
+
+    chat_box_row = page.locator(".chat-interface-input-row")
+    expect(chat_box_row).to_be_visible()
+
+    chat_box = chat_box_row.get_by_role("textbox")
+    expect(chat_box).to_be_visible()
+
+    # Document should be in the database
+    chats_url = f"http://{config.api.hostname}:{config.api.port}/chats"
+    chats = httpx.get(
+        chats_url, headers={"Authorization": f"Bearer {auth_token}"}
+    ).json()
+    assert len(chats) == 1
+    chat = chats[0]
+    chat_documents = chat["metadata"]["documents"]
+    assert len(chat_documents) == 1
+    assert chat_documents[0]["name"] == document_name
+
+    chat_box.fill("Tell me about the documents")
+
+    chat_button = chat_box_row.get_by_role("button")
+    expect(chat_button).to_be_visible()
+    chat_button.click()

tests/deploy/utils.py

@@ -0,0 +1,44 @@
+import os
+import time
+
+from fastapi.testclient import TestClient
+
+from ragna.assistants import RagnaDemoAssistant
+from ragna.core._utils import default_user
+
+
+class TestAssistant(RagnaDemoAssistant):
+    def answer(self, prompt, sources, *, multiple_answer_chunks: bool = True):
+        # Simulate a "real" assistant through a small delay. See
+        # https://github.com/Quansight/ragna/pull/401#issuecomment-2095851440
+        # for why this is needed.
+        #
+        # Note: multiple_answer_chunks is given a default value here to satisfy
+        # the tests in deploy/ui/test_ui.py. This can be removed if TestAssistant
+        # is ever removed from that file.
+        time.sleep(1e-3)
+        content = next(super().answer(prompt, sources))
+
+        if multiple_answer_chunks:
+            for chunk in content.split(" "):
+                yield f"{chunk} "
+        else:
+            yield content
+
+
+def authenticate(client: TestClient) -> None:
+    username = default_user()
+    token = (
+        client.post(
+            "/token",
+            data={
+                "username": username,
+                "password": os.environ.get(
+                    "RAGNA_DEMO_AUTHENTICATION_PASSWORD", username
+                ),
+            },
+        )
+        .raise_for_status()
+        .json()
+    )
+    client.headers["Authorization"] = f"Bearer {token}"