test: add Playwright E2E harness for operator GUI with decision-flow spec

Introduce a reusable browser-test harness under tests/operator_gui/:
- conftest.py: shared fixtures (operator_server, browser with graceful skip,
  operator_page with screenshot-on-failure artifacts)
- pages/: OperatorWorkbench Page Object encapsulating selectors and actions
- refactor test_browser_smoke.py onto the fixtures + POM (drops duplicated
  server/browser plumbing)
- test_decision_flow.py: approve/hold/reject E2E against the real server,
  covering the reject-requires-memo guard and decision persistence

Pin playwright in requirements-dev.txt (Chromium bundled offline for the
air-gapped target) and ignore the artifacts/ screenshot dir.

.gitignore

@@ -22,6 +22,7 @@ node_modules/
 # Runtime data, databases, logs, generated artifacts
 data/
 outputs/
+artifacts/
 *.sqlite3
 *.sqlite3-journal
 *.log

requirements-dev.txt

@@ -3,3 +3,8 @@
 pytest==8.4.2
 pytest-cov==6.0.0
 pytest-timeout==2.4.0
+# Operator-GUI browser (E2E) tests. The Chromium binary is NOT pip-installed:
+# on the build/staging host run `playwright install chromium`, then bundle the
+# browser cache (PLAYWRIGHT_BROWSERS_PATH) into the offline artifact for the
+# air-gapped target. Browser tests skip cleanly when Chromium is absent.
+playwright==1.49.1

tests/operator_gui/__init__.py

@@ -0,0 +1 @@
+"""Operator-GUI test package (enables relative imports of shared page objects)."""

tests/operator_gui/conftest.py

@@ -0,0 +1,105 @@
+"""Shared fixtures for operator-GUI tests.
+
+The browser fixtures lazily import Playwright (via ``importorskip``) and skip
+when Chromium is unavailable, so the non-browser static tests in this package
+still collect and run on a host without Playwright installed.
+"""
+
+from __future__ import annotations
+
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+from threading import Thread
+from typing import Any, Iterator
+
+import pytest
+
+
+ROOT = Path(__file__).resolve().parents[2]
+if str(ROOT / "src") not in sys.path:
+    sys.path.insert(0, str(ROOT / "src"))
+
+APP_DIR = ROOT / "web" / "operator-gui"
+ARTIFACT_DIR = ROOT / "artifacts" / "operator-gui-e2e"
+DEFAULT_VIEWPORT = {"width": 1280, "height": 900}
+
+
+@dataclass
+class OperatorServer:
+    """A running operator server plus the paths a spec may need to assert on."""
+
+    base_url: str
+    image_root: Path
+    store: Any
+
+
+@pytest.hookimpl(hookwrapper=True)
+def pytest_runtest_makereport(item, call):  # noqa: ANN001 - pytest hook signature
+    """Stash each phase's report so fixtures can screenshot on failure."""
+    outcome = yield
+    setattr(item, f"rep_{outcome.get_result().when}", outcome.get_result())
+
+
+@pytest.fixture
+def operator_server(tmp_path: Path) -> Iterator[OperatorServer]:
+    """Build and serve the operator app against a fresh, isolated SQLite store."""
+    from rights_filter.server.http_app import build_server
+    from rights_filter.server.image_store import LocalSubmissionImageStore
+    from rights_filter.server.sqlite_store import CopyrighterStore
+
+    image_root = tmp_path / "submissions"
+    image_root.mkdir()
+    store = CopyrighterStore(tmp_path / "copyrighter.sqlite3")
+    store.initialize()
+    server = build_server(
+        host="127.0.0.1",
+        port=0,
+        store=store,
+        image_store=LocalSubmissionImageStore(image_root),
+        static_dir=APP_DIR,
+    )
+    thread = Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield OperatorServer(
+            base_url=f"http://127.0.0.1:{server.server_port}",
+            image_root=image_root,
+            store=store,
+        )
+    finally:
+        server.shutdown()
+        thread.join(timeout=5)
+
+
+@pytest.fixture
+def browser() -> Iterator[Any]:
+    """Launch headless Chromium, skipping cleanly when it is unavailable."""
+    playwright_api = pytest.importorskip("playwright.sync_api")
+    with playwright_api.sync_playwright() as pw:
+        try:
+            launched = pw.chromium.launch(headless=True)
+        except Exception as exc:  # missing browser binaries, sandbox, etc.
+            pytest.skip(f"Playwright Chromium is not available: {exc}")
+        try:
+            yield launched
+        finally:
+            launched.close()
+
+
+@pytest.fixture
+def operator_page(browser: Any, request: pytest.FixtureRequest) -> Iterator[Any]:
+    """A fresh page at the default viewport; screenshots to artifacts on failure."""
+    page = browser.new_page(viewport=DEFAULT_VIEWPORT)
+    try:
+        yield page
+    finally:
+        report = getattr(request.node, "rep_call", None)
+        if report is not None and report.failed:
+            ARTIFACT_DIR.mkdir(parents=True, exist_ok=True)
+            safe_name = request.node.name.replace("/", "_").replace("::", "_")[:120]
+            try:
+                page.screenshot(path=str(ARTIFACT_DIR / f"{safe_name}.png"), full_page=True)
+            except Exception:
+                pass  # screenshot is best-effort diagnostics, never fail teardown
+        page.close()

tests/operator_gui/pages/__init__.py

@@ -0,0 +1,5 @@
+"""Page objects for operator-GUI browser (Playwright) tests."""
+
+from .workbench_page import OperatorWorkbench
+
+__all__ = ["OperatorWorkbench"]

tests/operator_gui/pages/workbench_page.py

@@ -0,0 +1,151 @@
+"""Page Object Model for the operator workbench single-page app.
+
+Encapsulates the operator-GUI selectors and interactions so browser specs read
+as user intent ("select case", "use suggested query") instead of raw locators.
+Mirrors the existing app structure in ``web/operator-gui/index.html``.
+
+All locator accessors return Playwright ``Locator`` objects, which auto-wait;
+prefer them over arbitrary timeouts (see the e2e-testing skill guidance).
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+
+# Status string the app shows after a suggested query is loaded but not yet run.
+SUGGESTED_QUERY_LOADED_STATUS = "추천 쿼리를 입력했습니다. 실행 버튼을 눌러 검색하세요."
+
+# Client-side guard message when a reject/correct decision is missing its memo.
+MEMO_REQUIRED_ERROR = "반려 또는 보정 결정에는 메모가 필요합니다."
+
+
+class OperatorWorkbench:
+    """Drives the operator GUI for browser-level (E2E) assertions."""
+
+    def __init__(self, page: Any, base_url: str) -> None:
+        self.page = page
+        self.base_url = base_url.rstrip("/")
+        self.console_errors: list[str] = []
+        page.on("console", lambda message: self.console_errors.append(f"console:{message.type}:{message.text}"))
+        page.on("pageerror", lambda error: self.console_errors.append(f"pageerror:{error}"))
+
+    # -- Locators ---------------------------------------------------------
+    @property
+    def queue_body(self) -> Any:
+        return self.page.locator("#queue-body")
+
+    @property
+    def workbench_view(self) -> Any:
+        return self.page.locator("#workbench-view")
+
+    @property
+    def case_title(self) -> Any:
+        return self.page.locator("#case-title")
+
+    @property
+    def manual_query(self) -> Any:
+        return self.page.locator("#manual-query")
+
+    @property
+    def manual_query_status(self) -> Any:
+        return self.page.locator("#manual-query-status")
+
+    @property
+    def submission_image_name(self) -> Any:
+        return self.page.locator("#submission-image-name")
+
+    @property
+    def submission_import_status(self) -> Any:
+        return self.page.locator("#submission-import-status")
+
+    @property
+    def recommendation_box(self) -> Any:
+        """Selected-case summary; includes the "현재 운영 결정: <label>" line."""
+        return self.page.locator("#recommendation-box")
+
+    @property
+    def decision_memo(self) -> Any:
+        return self.page.locator("#decision-memo")
+
+    @property
+    def memo_error(self) -> Any:
+        return self.page.locator("#memo-error")
+
+    def panel(self, name: str) -> Any:
+        """Workbench panel by ``data-workbench-panel`` (e.g. ``queries``, ``evidence``)."""
+        return self.page.locator(f'[data-workbench-panel="{name}"]')
+
+    def case_row(self, case_id: str) -> Any:
+        return self.page.locator(f'#queue-body [data-select-case="{case_id}"]')
+
+    # -- Navigation / setup ----------------------------------------------
+    def mock_bootstrap(self, payload: dict) -> "OperatorWorkbench":
+        """Stub ``/api/bootstrap`` so a spec controls the queue state.
+
+        Must be called before :meth:`goto` so the route is registered before the
+        app issues its first request.
+        """
+        self.page.route(
+            "**/api/bootstrap",
+            lambda route: route.fulfill(
+                status=200,
+                content_type="application/json",
+                body=json.dumps(payload, ensure_ascii=False),
+            ),
+        )
+        return self
+
+    def goto(self) -> "OperatorWorkbench":
+        self.page.goto(self.base_url + "/")
+        return self
+
+    # -- Actions ----------------------------------------------------------
+    def wait_for_case(self, case_id: str, **kwargs: Any) -> "OperatorWorkbench":
+        self.page.wait_for_selector(f'#queue-body [data-select-case="{case_id}"]', **kwargs)
+        return self
+
+    def select_case(self, case_id: str) -> "OperatorWorkbench":
+        self.page.get_by_role("button", name=case_id).click()
+        return self
+
+    def use_suggested_query(self, label: str) -> "OperatorWorkbench":
+        """Click a suggested-query chip, which fills the manual-query field."""
+        self.page.get_by_role("button", name=label).click()
+        return self
+
+    def upload_submission(self, file_path: str) -> "OperatorWorkbench":
+        self.page.set_input_files("#submission-image", str(file_path))
+        return self
+
+    def confirm_upload(self) -> "OperatorWorkbench":
+        self.page.get_by_role("button", name="사진 넣기").click()
+        return self
+
+    def upload_and_open(self, file_path: str, case_id: str, timeout: int = 10000) -> "OperatorWorkbench":
+        """Upload an image and wait until its new case is queued and the workbench opens."""
+        self.upload_submission(file_path)
+        self.confirm_upload()
+        self.wait_for_case(case_id, state="attached", timeout=timeout)
+        self.workbench_view.wait_for(state="visible", timeout=timeout)
+        return self
+
+    def set_decision_memo(self, text: str) -> "OperatorWorkbench":
+        self.decision_memo.fill(text)
+        return self
+
+    def decide(self, decision: str) -> "OperatorWorkbench":
+        """Click a decision button by its ``data-decision`` value (approved/held/rejected)."""
+        self.page.locator(f'[data-decision="{decision}"]').click()
+        return self
+
+    def wait_for_decision_label(self, label: str, timeout: int = 10000) -> "OperatorWorkbench":
+        """Wait until the recommendation box reflects a persisted decision label."""
+        self.page.wait_for_function(
+            "(label) => document.getElementById('recommendation-box')"
+            "?.innerText.includes('현재 운영 결정: ' + label)",
+            arg=label,
+            timeout=timeout,
+        )
+        return self

tests/operator_gui/test_browser_smoke.py

@@ -1,33 +1,9 @@
 from __future__ import annotations
 
-import json
-from pathlib import Path
-import sys
-from threading import Thread
-
 import pytest
 
-
+from .pages import OperatorWorkbench
-ROOT = Path(__file__).resolve().parents[2]
+from .pages.workbench_page import SUGGESTED_QUERY_LOADED_STATUS
-sys.path.insert(0, str(ROOT / "src"))
-APP_DIR = ROOT / "web" / "operator-gui"
-
-from rights_filter.server.http_app import build_server
-from rights_filter.server.image_store import LocalSubmissionImageStore
-from rights_filter.server.sqlite_store import CopyrighterStore
-
-
-def _start(server):
-    thread = Thread(target=server.serve_forever, daemon=True)
-    thread.start()
-    return thread
-
-
-def _browser_or_skip(playwright):
-    try:
-        return playwright.chromium.launch(headless=True)
-    except Exception as exc:
-        pytest.skip(f"Playwright Chromium is not available: {exc}")
 
 
 def _bootstrap_payload():
@@ -99,93 +75,42 @@ def _bootstrap_payload():
     }
 
 
-def test_browser_smoke_suggested_query_fills_manual_query_without_running_search(tmp_path: Path):
+def test_browser_smoke_suggested_query_fills_manual_query_without_running_search(operator_server, operator_page):
-    playwright = pytest.importorskip("playwright.sync_api")
+    workbench = OperatorWorkbench(operator_page, operator_server.base_url)
-    image_root = tmp_path / "submissions"
+    workbench.mock_bootstrap(_bootstrap_payload()).goto()
-    image_root.mkdir()
-    store = CopyrighterStore(tmp_path / "copyrighter.sqlite3")
-    store.initialize()
-    server = build_server(
-        host="127.0.0.1",
-        port=0,
-        store=store,
-        image_store=LocalSubmissionImageStore(image_root),
-        static_dir=APP_DIR,
-    )
-    _start(server)
-    base = f"http://127.0.0.1:{server.server_port}"
 
     try:
-        with playwright.sync_playwright() as pw:
+        workbench.wait_for_case("SUB-SMOKE1", timeout=5000)
-            browser = _browser_or_skip(pw)
+    except Exception as exc:
-            page = browser.new_page(viewport={"width": 1280, "height": 900})
+        body = workbench.page.locator("body").inner_text()[:1000]
-            browser_errors = []
+        pytest.fail(f"{exc}\nerrors={workbench.console_errors}\nbody={body}")
-            page.on("console", lambda message: browser_errors.append(f"console:{message.type}:{message.text}"))
-            page.on("pageerror", lambda error: browser_errors.append(f"pageerror:{error}"))
-            page.route(
-                "**/api/bootstrap",
-                lambda route: route.fulfill(
-                    status=200,
-                    content_type="application/json",
-                    body=json.dumps(_bootstrap_payload(), ensure_ascii=False),
-                ),
-            )
-            page.goto(base + "/")
-            try:
-                page.wait_for_selector('#queue-body [data-select-case="SUB-SMOKE1"]', timeout=5000)
-            except Exception as exc:
-                pytest.fail(f"{exc}\nerrors={browser_errors}\nbody={page.locator('body').inner_text()[:1000]}")
-            page.get_by_role("button", name="SUB-SMOKE1").click()
-            page.get_by_role("button", name="Smoke sample 저작권").click()
 
-            assert page.locator('[data-workbench-panel="queries"]').is_visible()
+    workbench.select_case("SUB-SMOKE1")
-            assert page.locator("#manual-query").input_value() == "Smoke sample 저작권"
+    workbench.use_suggested_query("Smoke sample 저작권")
-            assert page.locator("#manual-query-status").inner_text() == "추천 쿼리를 입력했습니다. 실행 버튼을 눌러 검색하세요.", browser_errors
 
-            browser.close()
+    assert workbench.panel("queries").is_visible()
-    finally:
+    assert workbench.manual_query.input_value() == "Smoke sample 저작권"
-        server.shutdown()
+    assert workbench.manual_query_status.inner_text() == SUGGESTED_QUERY_LOADED_STATUS, workbench.console_errors
 
 
-def test_browser_uploads_image_and_selects_new_submission(tmp_path: Path):
+def test_browser_uploads_image_and_selects_new_submission(operator_server, operator_page, tmp_path):
-    playwright = pytest.importorskip("playwright.sync_api")
-    image_root = tmp_path / "submissions"
-    image_root.mkdir()
     upload_file = tmp_path / "smoke upload.svg"
     upload_file.write_text("<svg xmlns='http://www.w3.org/2000/svg' width='80' height='60'></svg>", encoding="utf-8")
-    store = CopyrighterStore(tmp_path / "copyrighter.sqlite3")
+
-    store.initialize()
+    workbench = OperatorWorkbench(operator_page, operator_server.base_url)
-    server = build_server(
+    workbench.goto()
-        host="127.0.0.1",
+
-        port=0,
+    workbench.upload_submission(str(upload_file))
-        store=store,
+    assert workbench.submission_image_name.inner_text() == "smoke upload.svg"
-        image_store=LocalSubmissionImageStore(image_root),
+
-        static_dir=APP_DIR,
+    workbench.confirm_upload()
+    workbench.wait_for_case("smoke-upload", state="attached", timeout=10000)
+    workbench.workbench_view.wait_for(state="visible", timeout=10000)
+
+    assert (operator_server.image_root / "images" / "smoke-upload.svg").exists()
+    assert (
+        "smoke-upload 사진이 추가되었습니다. 새 심사 건으로 바로 선택했습니다."
+        in workbench.submission_import_status.inner_text()
     )
-    _start(server)
+    assert workbench.case_title.inner_text() == "smoke-upload · smoke-upload"
-    base = f"http://127.0.0.1:{server.server_port}"
+    assert not workbench.console_errors
-
-    try:
-        with playwright.sync_playwright() as pw:
-            browser = _browser_or_skip(pw)
-            page = browser.new_page(viewport={"width": 1280, "height": 900})
-            browser_errors = []
-            page.on("console", lambda message: browser_errors.append(f"console:{message.type}:{message.text}"))
-            page.on("pageerror", lambda error: browser_errors.append(f"pageerror:{error}"))
-
-            page.goto(base + "/")
-            page.set_input_files("#submission-image", str(upload_file))
-            assert page.locator("#submission-image-name").inner_text() == "smoke upload.svg"
-
-            page.get_by_role("button", name="사진 넣기").click()
-            page.wait_for_selector('#queue-body [data-select-case="smoke-upload"]', state="attached", timeout=10000)
-            page.wait_for_selector("#workbench-view", state="visible", timeout=10000)
-
-            assert (image_root / "images" / "smoke-upload.svg").exists()
-            assert "smoke-upload 사진이 추가되었습니다. 새 심사 건으로 바로 선택했습니다." in page.locator("#submission-import-status").inner_text()
-            assert page.locator("#case-title").inner_text() == "smoke-upload · smoke-upload"
-            assert not browser_errors
-
-            browser.close()
-    finally:
-        server.shutdown()

tests/operator_gui/test_decision_flow.py

@@ -0,0 +1,72 @@
+"""Operator decision flow (approve / hold / reject) end-to-end.
+
+Runs against the real server (no bootstrap stub): each test uploads an image to
+seed a genuine submission, then exercises the decision pane and asserts the
+decision both renders and persists through ``refreshFromApi``.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from .pages import OperatorWorkbench
+from .pages.workbench_page import MEMO_REQUIRED_ERROR
+
+
+_SVG = "<svg xmlns='http://www.w3.org/2000/svg' width='80' height='60'></svg>"
+
+
+def _open_seeded_case(workbench: OperatorWorkbench, tmp_path: Path, case_id: str = "decide-me") -> str:
+    upload_file = tmp_path / f"{case_id}.svg"
+    upload_file.write_text(_SVG, encoding="utf-8")
+    workbench.goto().upload_and_open(str(upload_file), case_id)
+    # A freshly uploaded submission starts unreviewed.
+    assert "현재 운영 결정: 미심사" in workbench.recommendation_box.inner_text()
+    return case_id
+
+
+def test_reject_without_memo_is_blocked_and_not_persisted(operator_server, operator_page, tmp_path):
+    workbench = OperatorWorkbench(operator_page, operator_server.base_url)
+    _open_seeded_case(workbench, tmp_path)
+
+    workbench.decide("rejected")  # 반려 requires a memo
+
+    assert workbench.memo_error.inner_text() == MEMO_REQUIRED_ERROR
+    # The guard fires client-side, so the decision is never sent or persisted.
+    assert "현재 운영 결정: 미심사" in workbench.recommendation_box.inner_text()
+
+
+def test_approve_persists_without_memo(operator_server, operator_page, tmp_path):
+    workbench = OperatorWorkbench(operator_page, operator_server.base_url)
+    _open_seeded_case(workbench, tmp_path)
+
+    workbench.decide("approved")  # 승인 needs no memo
+    workbench.wait_for_decision_label("승인")
+
+    assert workbench.memo_error.inner_text() == ""
+    assert not workbench.console_errors
+
+
+def test_hold_persists_without_memo(operator_server, operator_page, tmp_path):
+    workbench = OperatorWorkbench(operator_page, operator_server.base_url)
+    _open_seeded_case(workbench, tmp_path)
+
+    workbench.decide("held")  # 보류 needs no memo
+    workbench.wait_for_decision_label("보류")
+
+    assert workbench.memo_error.inner_text() == ""
+    assert not workbench.console_errors
+
+
+def test_reject_with_memo_persists_and_clears_memo(operator_server, operator_page, tmp_path):
+    workbench = OperatorWorkbench(operator_page, operator_server.base_url)
+    _open_seeded_case(workbench, tmp_path)
+
+    workbench.set_decision_memo("저작권 침해 소지")
+    workbench.decide("rejected")
+    workbench.wait_for_decision_label("반려")
+
+    assert workbench.memo_error.inner_text() == ""
+    # The memo field is cleared after a successful decision.
+    assert workbench.decision_memo.input_value() == ""
+    assert not workbench.console_errors