"""Phase 49 story E2E (Playwright): archive upload sources. Story: ``.agents/user_stories/archive-upload-sources.md`` Run in isolation (DB must be up: ``podman compose up -d db``): uv run pytest tests/e2e/test_archive_upload_sources.py -v --no-cov The story gate for the **archive upload** form on the admin Sources page (``/git-sources.html``, phase 49 — the phase-38 "Add a local directory" form is gone, replaced by this form): an uploaded ``.tar``/``.tar.gz``/ ``.tgz``/``.zip`` is safely unpacked under ``BOR_UPLOAD_DIR//`` (name = filename minus the archive suffix), the ``git_sources`` row is upserted (``kind='local'``, no duplicates), and the source is **scanned in a background task** (phase 64, task 03 — owner-locked A1: the POST answers **202 the moment the archive is safely on disk** — the "Successfully uploaded — " toast fires then and the user may navigate away — while unpack → swap → row upsert → model check → single-source ``import_sources`` with ``prune=True`` + the change-gated overview refresh run server-side behind ``GET /api/git-sources/upload/status``, the phase-32 ``SyncStatus`` pattern with the phase-64 ``current_file``/counts). The result line and the row land from the status ``success`` (same ``UploadOut`` counts, uncompressed in shape) — the real pipeline, against the deterministic mock LLM (no real models, no network beyond the app itself). **Timing fixture (phase 64):** the mock LLM answers instantly, so this module's app boots behind ``tests/e2e/slow_llm.py`` — a delay-injecting reverse proxy in front of it (``SLOW_DELAY_S`` per request). A 2-file scan is 5 LLM requests ≈ 5 × 0.6 s ≈ 3 s: long enough to outlive the UI's 2 s status poll, so the button's literal "Uploading… → Processing… → restored" lifecycle is observable (the "Processing…" tick even carries the live file, A4) instead of racing the mock. The archives are **built in-test** with Python's ``tarfile`` over ``tmp_path`` fixture files carrying markdown sentinels (``ALPHA-…`` / ``BETA-…`` / ``GAMMA-…``) and are always named ``e2e-upload.tar.gz`` — so the source name is ``e2e-upload`` and re-uploading under the same filename exercises the in-place replace (one folder, one row, dropped files pruned from the KB). ``v1`` holds ``alpha.md`` + ``beta.md``; ``v2`` (same basename) modifies ``alpha``, drops ``beta``, adds ``gamma``. Per-module app env (the conftest pattern, module-scoped — as in ``test_git_sources_admin.py`` / ``test_local_directory_sources.py``): ``BOR_UPLOAD_DIR`` points at a scratch dir the suite can inspect from the host (the app runs on the same machine), and ``BOR_GIT_SOURCES`` is forced empty so the dev ``.env``'s fallback URL never renders as an env row on the (initially empty) table. Contract under test: * the **swap** (task 03): the phase-38 local form is gone (count 0); the upload form is in its place with the labeled file input (accept = the four archive extensions), the "Upload & scan" button, and the hint explains unpack/scan + in-place replace; * **upload → 202 + toast → background scan → list** (§7.4 never-stale, phase-64 A1/A2): the button shows "Uploading…" while the POST is in flight (the request is held in the browser via ``page.route`` so the in-flight state is deterministic); at the 202 the "Successfully uploaded — " toast fires (``.toast.is-visible``, ``role="status"``) WHILE the scan is still running, and the button hands over to the scan — "Processing…" (the live-file tick carries the current file, A4) — then restores when the status ``success`` lands: the result line shows the added count; the list gains exactly one row for ``e2e-upload`` with the **Local** badge; ``GET /api/docs`` lists both sentinel files under source ``e2e-upload``; the RAG catalog (``/sources.html``) shows them; * **re-upload, same filename** → in-place replace: the result line shows the prune, the SECOND RUN'S STATUS ``detail`` carries the prune/refresh counts, the list still has exactly ONE ``e2e-upload`` row (no duplicate), the KB shows the changed ``alpha`` + the new ``gamma`` and NOT the dropped ``beta``, and the on-disk folder holds only the new archive's files; * **bad file** → inline 422 (role=alert) naming the accepted formats (UNCHANGED — the name/format/cap gates are inline, pre-202, exactly as before), button restored, the file selection kept, the list unchanged, and a subsequent good upload still works (the form is not wedged); * **anonymous** → the sign-in gate (``#git-sources-gate``) shows, the manager (and thus the upload form) stays hidden, and ``POST /api/git-sources/upload`` is 403 — as is the phase-64 ``GET /api/git-sources/upload/status`` (same wall). Test → story mapping (Playwright Mapping Rule): 1. ``test_form_swapped`` 2. ``test_upload_scans_and_lists`` 3. ``test_reupload_replaces_in_place`` 4. ``test_bad_file_inline_error`` 5. ``test_anonymous_gate`` """ from __future__ import annotations import io import os import re import subprocess import sys import tarfile import time from collections.abc import Iterator from pathlib import Path from typing import Any import pytest from playwright.sync_api import Page, expect from sqlalchemy import text from app.db import SessionLocal from e2e.auth_helpers import login from e2e.conftest import ( ADMIN_PASSWORD, APP_PORT, MOCK_PORT, SESSION_SECRET, USE_REAL_LLM, _wait_http, ) REPO = Path(__file__).resolve().parents[2] APP_URL = f"http://127.0.0.1:{APP_PORT}" #: The slow-LLM proxy's port (the conftest's mock LLM stays on MOCK_PORT). SLOW_PORT = int(os.environ.get("E2E_SLOW_LLM_PORT", "8902")) SLOW_URL = f"http://127.0.0.1:{SLOW_PORT}" #: Per-LLM-request delay on the proxy — a 2-file scan is 5 LLM requests #: (the check_models embed + chat probe, one embed per file, the #: change-gated overview chat) ≈ 5 × 0.6 s ≈ 3 s: the scan outlives the #: UI's 2 s status poll, so the button's "Uploading… → Processing… → #: restored" lifecycle (with the live-file tick, A4) is observable. SLOW_DELAY_S = "0.6" GIT_SOURCES_URL = "/git-sources.html" SOURCES_URL = "/sources.html" #: The archive basename (both versions) — the source/folder name is the #: filename minus the archive suffix (the phase's locked naming rule). SOURCE_NAME = "e2e-upload" #: v1: two sentinel docs. v2 (same filename): alpha CHANGED, beta DROPPED, #: gamma ADDED — the in-place-replace subject. ALPHA_SENTINEL_V1 = "ALPHA-TOKEN-v1-7f31" ALPHA_SENTINEL_V2 = "ALPHA-TOKEN-v2-8b42" BETA_SENTINEL_V1 = "BETA-TOKEN-v1-2c90" GAMMA_SENTINEL_V2 = "GAMMA-TOKEN-v2-5e44" V1_FILES: dict[str, str] = { "alpha.md": ( "# Alpha note\n" "\n" "First version of the alpha note — it changes in v2.\n" f"\nMarker: {ALPHA_SENTINEL_V1}\n" ), "beta.md": ( "# Beta note\n" "\n" "Only present in v1 — v2 drops it (the prune subject).\n" f"\nMarker: {BETA_SENTINEL_V1}\n" ), } V2_FILES: dict[str, str] = { "alpha.md": ( "# Alpha note\n" "\n" "Second version of the alpha note — modified in place.\n" f"\nMarker: {ALPHA_SENTINEL_V2}\n" ), "gamma.md": ( "# Gamma note\n" "\n" "Brand new in v2 — the add subject of the re-upload.\n" f"\nMarker: {GAMMA_SENTINEL_V2}\n" ), } #: The scan runs the full pipeline against the mock LLM (models probe + #: embed batch + per-doc summaries + the change-gated overview) — #: generous, like the sync suites; no client-side hard timeout. UPLOAD_TIMEOUT_MS = 90_000 # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- def _build_targz(path: Path, files: dict[str, str]) -> Path: """A deterministic ``.tar.gz`` (mtime 0) over the given files.""" with tarfile.open(path, "w:gz") as tf: for rel, content in files.items(): data = content.encode("utf-8") info = tarfile.TarInfo(rel) info.size = len(data) info.mtime = 0 tf.addfile(info, io.BytesIO(data)) return path @pytest.fixture(scope="module") def upload_dir(tmp_path_factory: pytest.TempPathFactory) -> Path: """The app's ``BOR_UPLOAD_DIR`` for this suite — a scratch dir the host-side assertions inspect (the app server runs on the same machine). The app creates it on the first upload.""" return tmp_path_factory.mktemp("bor_uploads") / "uploads" @pytest.fixture(scope="module") def tarball_v1(tmp_path_factory: pytest.TempPathFactory) -> Path: """v1 — in its OWN subdirectory so v2 can reuse the same basename (``e2e-upload.tar.gz``): the in-place-replace identity IS the filename, and ``set_input_files`` sends the path's basename.""" root = tmp_path_factory.mktemp("bor_archive_v1") return _build_targz(root / f"{SOURCE_NAME}.tar.gz", V1_FILES) @pytest.fixture(scope="module") def tarball_v2(tmp_path_factory: pytest.TempPathFactory) -> Path: """v2 — same basename as v1 (a different parent dir).""" root = tmp_path_factory.mktemp("bor_archive_v2") return _build_targz(root / f"{SOURCE_NAME}.tar.gz", V2_FILES) @pytest.fixture(scope="module") def slow_llm(mock_llm: int) -> Iterator[int]: """The delay-injecting reverse proxy in front of the mock LLM (tests/e2e/slow_llm.py) — this suite's timing fixture: the phase-64 button lifecycle ("Uploading… → Processing… → restored") needs the 2-file scan to outlive the UI's 2 s status poll (see ``SLOW_DELAY_S``).""" env = dict(os.environ) env.pop("DEBUGPY", None) env["SLOW_LLM_DELAY_S"] = SLOW_DELAY_S env["E2E_MOCK_PORT"] = str(MOCK_PORT) proc = subprocess.Popen( [sys.executable, "-m", "uvicorn", "tests.e2e.slow_llm:app", "--host", "127.0.0.1", "--port", str(SLOW_PORT), "--log-level", "warning"], cwd=REPO, env=env, ) try: _wait_http(f"{SLOW_URL}/v1/models") yield SLOW_PORT finally: proc.terminate() try: proc.wait(timeout=10) except subprocess.TimeoutExpired: proc.kill() @pytest.fixture(scope="module") def app_server( mock_llm: int, slow_llm: int, upload_dir: Path, tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[str]: """The real app under test — per-module env: the LLM base URL is the SLOW PROXY in front of the mock (the timing fixture), uploads unpack into a scratch dir and the env git list is forced empty (the dev ``.env``'s ``BOR_GIT_SOURCES`` must not render as env rows on the initially empty table). No sync is triggered here — the upload's own scan is the pipeline under test.""" env = dict(os.environ) env.pop("DEBUGPY", None) env["BOR_ENVIRONMENT"] = "e2e" env["BOR_STATIC_DIR"] = str(REPO / "frontend") env["BOR_LLM_BASE_URL"] = ( "https://aipi.reeseapps.com/v1" if USE_REAL_LLM else f"{SLOW_URL}/v1" ) # Mock-calibrated threshold (conftest pattern) — no chat turn is # ever sent in this suite, but the app boots with the same env shape. env["BOR_RELEVANCE_THRESHOLD"] = "0.30" env.setdefault( "BOR_DATABASE_URL", "postgresql+psycopg://reese:reese@localhost:5432/brain_of_reese", ) # Phase 16: admin auth must be set or create_app() refuses to boot. env["BOR_ADMIN_PASSWORD"] = ADMIN_PASSWORD env["BOR_SESSION_SECRET"] = SESSION_SECRET env["BOR_GIT_SOURCES"] = "" # Phase 49: unpack uploads into the suite's scratch dir (host- # inspectable) and keep the (unused) git checkouts out of the dev # location. env["BOR_UPLOAD_DIR"] = str(upload_dir) env["BOR_SOURCES_DIR"] = str(tmp_path_factory.mktemp("bor_checkouts")) proc = subprocess.Popen( [sys.executable, "-m", "uvicorn", "app.main:app", "--host", "127.0.0.1", "--port", str(APP_PORT), "--log-level", "warning"], cwd=REPO, env=env, ) try: _wait_http(f"{APP_URL}/api/health") yield APP_URL finally: proc.terminate() try: proc.wait(timeout=10) except subprocess.TimeoutExpired: proc.kill() @pytest.fixture(scope="module") def app_url(app_server: str) -> str: return app_server def _truncate_all() -> None: """Fresh registry + KB per test (the E2E isolation pattern): the upload's counts and every ``/api/docs`` assertion must be this test's own doing. The E2E suites share one Postgres, and a leftover git_sources row or document would corrupt the row-count and doc-list assertions (and a leftover document under the same source name would survive the re-upload's single-source prune).""" with SessionLocal() as db: db.execute(text("TRUNCATE chunks, documents, query_log, kb_overview, git_sources")) db.commit() @pytest.fixture(autouse=True) def _clean(db_ready: None) -> Iterator[None]: _truncate_all() yield _truncate_all() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _admin_git_sources_page(page: Page, app_url: str) -> None: """Real form login landing on the git sources page (admin settled: Sign out visible, the manager revealed by the page module).""" login(page, app_url, next=GIT_SOURCES_URL) expect(page).to_have_url(app_url + GIT_SOURCES_URL, timeout=30_000) expect(page.locator("#sign-out-btn")).to_be_visible(timeout=15_000) expect(page.locator("#git-sources-gate")).to_be_hidden() expect(page.locator("#git-sources-content")).to_be_visible() def _docs(page: Page, app_url: str) -> list[tuple[str, str]]: """``GET /api/docs`` as the signed-in page → sorted (source, path) pairs (the admin cookie rides the browser context).""" r = page.request.get(f"{app_url}/api/docs") assert r.status == 200, r.text return sorted((d["source"], d["path"]) for d in r.json()["documents"]) def _upload_via_page(page: Page, archive: Path) -> str: """Pick the archive, submit the form, and wait for the result line (the phase-64 202 path: toast at the 202, then the button's status polling renders the line from the run's ``success``) — returns its text. The failing path is asserted explicitly by the bad-file test, so any non-result outcome here is a test error.""" page.set_input_files("#archive-upload-file", str(archive)) page.click("#archive-upload-btn") result = page.locator("#archive-upload-result") expect(result).to_be_visible(timeout=UPLOAD_TIMEOUT_MS) text = result.text_content() assert text is not None return text def _wait_upload_running(page: Page, app_url: str, timeout_s: float = 15.0) -> dict[str, Any]: """Poll (cookie-authenticated) the upload status endpoint until the run is ``running`` — the phase-64 single source of truth for the background scan (A1).""" deadline = time.monotonic() + timeout_s body: dict[str, Any] = {} while time.monotonic() < deadline: r = page.request.get(f"{app_url}/api/git-sources/upload/status") assert r.status == 200, r.text body = r.json() if body["state"] == "running": return body if body["state"] in ("success", "failed"): raise AssertionError(f"the scan settled too fast to observe: {body}") time.sleep(0.1) raise AssertionError(f"the scan never entered running: {body}") def _hold_upload_request(page: Page, hold_s: float) -> None: """Intercept the upload POST and hold the REQUEST in the browser for ``hold_s`` seconds before letting it reach the server. While it is held, the page's fetch is guaranteed pending — so the §7.4 in-flight state (disabled button, "Uploading…" label) is observable deterministically instead of racing the mock LLM's fast scan.""" def handle(route: Any) -> None: time.sleep(hold_s) route.continue_() page.route("**/api/git-sources/upload", handle) # --------------------------------------------------------------------------- # 1. The swap: local form out, upload form in # --------------------------------------------------------------------------- def test_form_swapped(page: Page, app_url: str, db_ready: None) -> None: """The phase-38 "Add a local directory" form is GONE and the archive upload form stands in its place: visible file input (accept = the four archive extensions), the "Upload & scan" button, and a hint that explains the unpack/scan + in-place-replace semantics.""" page.set_default_timeout(30_000) _admin_git_sources_page(page, app_url) # The phase-38 local form is gone (phase 49 replaced it)… expect(page.locator("#local-source-form")).to_have_count(0) expect(page.locator("#local-source-path")).to_have_count(0) expect(page.locator("#local-source-add")).to_have_count(0) # …and the upload form is in its place, visible with its parts. expect(page.locator("#archive-upload-form")).to_be_visible() file_input = page.locator("#archive-upload-file") expect(file_input).to_be_visible() accept = file_input.get_attribute("accept") or "" for ext in (".tar", ".tar.gz", ".tgz", ".zip"): assert ext in accept, f"accept={accept!r} is missing {ext!r}" btn = page.locator("#archive-upload-btn") expect(btn).to_be_visible() expect(btn).to_be_enabled() expect(btn).to_have_text("Upload & scan") # The error/result lines ship (hidden) with the right roles. assert page.locator("#archive-upload-error").get_attribute("role") == "alert" result = page.locator("#archive-upload-result") assert result.get_attribute("role") == "status" expect(result).to_be_hidden() # The hint explains unpack/scan + in-place replace (task 03). hint = page.locator("#git-sources-hint") expect(hint).to_be_visible() expect(hint).to_contain_text("unpack") expect(hint).to_contain_text("scan") expect(hint).to_contain_text("in place") # --------------------------------------------------------------------------- # 2. Upload → scan → list (the §7.4 in-flight state, the counts, the # Local row, the KB, the RAG catalog) # --------------------------------------------------------------------------- def test_upload_scans_and_lists( page: Page, app_url: str, db_ready: None, tarball_v1: Path, upload_dir: Path ) -> None: """One real upload through the page: while the POST is in flight the button is disabled and reads "Uploading…"; at the 202 the "Successfully uploaded — " toast fires (A2) WHILE the background scan is still running and the button hands over to it — "Processing…" (the 2 s poll tick carries the live file, A4); when the status ``success`` lands the button restores, the result line shows the added count (2), the file input clears, the list gains exactly ONE row for ``e2e-upload`` with the Local badge, ``/api/docs`` lists both sentinel files under the source, and the RAG catalog shows them where the admin expects them.""" page.set_default_timeout(30_000) _admin_git_sources_page(page, app_url) expect(page.locator("#git-sources-tbody tr")).to_have_count(0) btn = page.locator("#archive-upload-btn") result = page.locator("#archive-upload-result") # Hold the upload request in the browser: the in-flight state below # cannot race the receive while it is held. _hold_upload_request(page, hold_s=0.8) page.set_input_files("#archive-upload-file", str(tarball_v1)) btn.click() # In flight (§7.4): disabled + relabeled, no result yet. expect(btn).to_be_disabled() expect(btn).to_have_text("Uploading…") expect(result).to_be_hidden() # The request goes out; the server stores the archive and answers # 202 the moment it is safely on disk (A1) → the toast fires NOW # (A2) — while the scan is still running — and the button hands # over to the scan (bare "Processing…" — A4: no file yet during the # unpack phase). toast = page.locator(".toast") expect(toast).to_have_count(1, timeout=UPLOAD_TIMEOUT_MS) expect(toast).to_have_class(re.compile(r"\bis-visible\b")) assert toast.get_attribute("role") == "status" expect(toast).to_have_text(f"Successfully uploaded — {SOURCE_NAME}") expect(result).to_be_hidden() expect(btn).to_be_disabled() expect(btn).to_have_text("Processing…", timeout=5_000) # The scan is running server-side (the status endpoint is the # single source of truth, A1) — the run the UI's poll tracks. _wait_upload_running(page, app_url) # …and the button's 2 s poll tick renders the live file label # ("Processing… (n/m)", A4). expect(btn).to_have_text( re.compile(rf"Processing… {re.escape(SOURCE_NAME)}/.+\.md"), timeout=UPLOAD_TIMEOUT_MS, ) # The status success lands → the result line (the same UploadOut # counts) + the never-stale restore (input cleared). expect(result).to_be_visible(timeout=UPLOAD_TIMEOUT_MS) expect(result).to_have_text("2 added") expect(btn).to_be_enabled() expect(btn).to_have_text("Upload & scan") expect(page.locator("#archive-upload-file")).to_have_value("") # The list gained exactly one row — for the source, with the Local # badge and the full unpacked path in the mono cell. expect(page.locator("#git-sources-tbody tr")).to_have_count(1, timeout=30_000) row = page.locator("#git-sources-tbody tr", has_text=SOURCE_NAME) expect(row).to_have_count(1) expect(row.locator("span.git-source-kind")).to_have_text("Local") expect(row.locator("td.git-source-url-cell code")).to_have_text( str(upload_dir / SOURCE_NAME) ) # The KB: both sentinel files, under the source name e2e-upload. assert _docs(page, app_url) == [(SOURCE_NAME, "alpha.md"), (SOURCE_NAME, "beta.md")] # The RAG catalog (admin sees it): both docs, under the source. page.goto(app_url + SOURCES_URL) expect(page.locator("#docs-tbody tr")).to_have_count(2) expect(page.locator("#docs-tbody tr", has_text="alpha.md")).to_have_count(1) expect(page.locator("#docs-tbody tr", has_text="beta.md")).to_have_count(1) expect(page.locator("#docs-tbody tr", has_text=SOURCE_NAME)).to_have_count(2) # --------------------------------------------------------------------------- # 3. Re-upload, same filename → in-place replace (no duplicate row, # dropped file pruned, changed/new file indexed) # --------------------------------------------------------------------------- def test_reupload_replaces_in_place( page: Page, app_url: str, db_ready: None, tarball_v1: Path, tarball_v2: Path, upload_dir: Path, ) -> None: """v1 then v2 under the SAME filename (``e2e-upload.tar.gz``): the result line shows the prune, the SECOND RUN'S STATUS ``detail`` carries the prune/refresh counts (phase 64 — the line is rendered from the status success), the list still has exactly ONE ``e2e-upload`` row (the row count for that source is invariant — no duplicate), the KB shows the changed ``alpha`` + the new ``gamma`` and NOT the dropped ``beta``, and the on-disk folder holds only the new archive's files.""" page.set_default_timeout(30_000) _admin_git_sources_page(page, app_url) # Baseline: v1 through the page (202 → "2 added", one row). assert _upload_via_page(page, tarball_v1) == "2 added" expect(page.locator("#git-sources-tbody tr", has_text=SOURCE_NAME)).to_have_count(1) # Re-upload v2 — SAME basename, different parent dir (the file # input's selection is replaced wholesale). assert _upload_via_page(page, tarball_v2) is not None result = page.locator("#archive-upload-result") expect(result).to_have_text(re.compile(r"\d+ pruned")) # The SECOND RUN's status ``detail`` shows the prune/refresh counts # (phase 64: the result line is rendered from this success). r = page.request.get(f"{app_url}/api/git-sources/upload/status") assert r.status == 200, r.text status = r.json() assert status["state"] == "success", status detail = status["detail"] assert detail["source"] == SOURCE_NAME assert detail["files"] == 2 assert detail["added"] == 1 # gamma — new in v2 assert detail["updated"] == 1 # alpha — changed in v2 assert detail["pruned"] == 1 # beta — dropped in v2 # No duplicate: exactly ONE row for that source (and one row total). expect(page.locator("#git-sources-tbody tr", has_text=SOURCE_NAME)).to_have_count(1) expect(page.locator("#git-sources-tbody tr")).to_have_count(1) # The registry agrees: one kind=local row, the unpacked path. r = page.request.get(f"{app_url}/api/git-sources") assert r.status == 200, r.text body = r.json() assert [(s["kind"], s["path"]) for s in body["sources"]] == [ ("local", str(upload_dir / SOURCE_NAME)) ] # The KB: gamma + the CHANGED alpha, NOT the dropped beta. assert _docs(page, app_url) == [(SOURCE_NAME, "alpha.md"), (SOURCE_NAME, "gamma.md")] # …and the indexed alpha is the v2 one (in-place replace, proven in # the KB, not just the filesystem). content = page.request.get( f"{app_url}/api/documents/content?source={SOURCE_NAME}&path=alpha.md" ) assert content.status == 200, content.text assert ALPHA_SENTINEL_V2 in content.json()["content"] assert ALPHA_SENTINEL_V1 not in content.json()["content"] # The on-disk folder holds ONLY v2's files (the swap replaced the # whole folder — no stale v1 file survived). folder = upload_dir / SOURCE_NAME assert {p.name for p in folder.iterdir()} == set(V2_FILES) # --------------------------------------------------------------------------- # 4. Bad file → inline 422; the form is not wedged # --------------------------------------------------------------------------- def test_bad_file_inline_error( page: Page, app_url: str, db_ready: None, tarball_v1: Path, tmp_path: Path ) -> None: """A ``.txt`` through the file input: the role=alert line shows the 422 detail naming the accepted formats, the button restores, the file selection is KEPT (the fix is one re-pick), the list is unchanged — and a subsequent good upload still works (the form is not wedged).""" page.set_default_timeout(30_000) _admin_git_sources_page(page, app_url) expect(page.locator("#git-sources-tbody tr")).to_have_count(0) bad = tmp_path / "notes.txt" bad.write_text("I am not an archive.\n", encoding="utf-8") error = page.locator("#archive-upload-error") btn = page.locator("#archive-upload-btn") page.set_input_files("#archive-upload-file", str(bad)) btn.click() # The 422 detail inline (role=alert), naming the accepted formats. expect(error).to_be_visible(timeout=30_000) assert error.get_attribute("role") == "alert" expect(error).to_contain_text("only .tar, .tar.gz, .tgz or .zip archives are accepted") # Never stale + the selection kept + no result line + list unchanged. expect(btn).to_be_enabled() expect(btn).to_have_text("Upload & scan") # The selection is kept (the fix is one re-pick) — Chromium reports # a fake path (``…/notes.txt``), so assert on the basename. bad_value = page.locator("#archive-upload-file").input_value() assert bad_value.endswith("notes.txt"), bad_value expect(page.locator("#archive-upload-result")).to_be_hidden() expect(page.locator("#git-sources-tbody tr")).to_have_count(0) # The form is not wedged: a good upload right after still works. assert _upload_via_page(page, tarball_v1) == "2 added" expect(error).to_be_hidden() expect(page.locator("#git-sources-tbody tr", has_text=SOURCE_NAME)).to_have_count(1) assert _docs(page, app_url) == [(SOURCE_NAME, "alpha.md"), (SOURCE_NAME, "beta.md")] # --------------------------------------------------------------------------- # 5. Anonymous: the gate, the hidden manager, the 403 # --------------------------------------------------------------------------- def test_anonymous_gate(page: Page, app_url: str, db_ready: None) -> None: """Anonymous on the page: the sign-in gate shows, the manager (and thus the upload form) stays hidden, and the upload route 403s (``require_admin`` — A10).""" page.set_default_timeout(30_000) page.goto(app_url + GIT_SOURCES_URL) expect(page.locator("#sign-in-link")).to_be_visible(timeout=15_000) expect(page.locator("#sign-out-btn")).to_be_hidden() gate = page.locator("#git-sources-gate") expect(gate).to_be_visible() expect(gate).to_contain_text("Sign in to manage the git sources") # The manager is hidden — so is the upload form inside it. expect(page.locator("#git-sources-content")).to_be_hidden() expect(page.locator("#archive-upload-form")).to_be_hidden() # The upload route 403s anonymous callers (require_admin runs before # the multipart body is parsed — the body is a stand-in, the # test_local_directory_sources.py pattern for this route)… r = page.request.post(f"{app_url}/api/git-sources/upload", data={"file": ""}) assert r.status == 403 # …and so does the phase-64 upload STATUS endpoint (same wall). assert page.request.get(f"{app_url}/api/git-sources/upload/status").status == 403