booru-viewer/tests/core/test_concurrency.py
pax bf14466382 Add Phase A test suite for core/ primitives
First regression-test layer for booru-viewer. Pure Python — no Qt, no
mpv, no network, no real filesystem outside tmp_path. Locks in the
security and concurrency invariants from the 54ccc40 + eb58d76 hardening
commits ahead of the upcoming popout state machine refactor (Prompt 3),
which needs a stable baseline to refactor against.

16 tests across five files mirroring the source layout under
booru_viewer/core/:

- tests/core/test_db.py (4):
  - _validate_folder_name rejection rules (.., /foo, \\foo, .hidden,
    ~user, empty) and acceptance categories (unicode, spaces, parens)
  - add_bookmark INSERT OR IGNORE collision returns the existing row
    id (locks in the lastrowid=0 fix)
  - get_bookmarks LIKE escaping (literal cat_ear does not match catear)

- tests/core/test_cache.py (7):
  - _referer_for hostname suffix matching (gelbooru.com / donmai.us
    apex rewrite, both exact-match and subdomain)
  - _referer_for rejects substring attackers
    (imgblahgelbooru.attacker.com does NOT pick up the booru referer)
  - ugoira frame-count and uncompressed-size caps refuse zip bombs
    before any decompression
  - _do_download MAX_DOWNLOAD_BYTES enforced both at the
    Content-Length pre-check AND in the chunk-loop running total
  - _is_valid_media returns True on OSError (no delete + redownload
    loop on transient EBUSY)

- tests/core/test_config.py (2):
  - saved_folder_dir rejects literal .. and ../escape
  - find_library_files walks root + 1 level, filters by
    MEDIA_EXTENSIONS, exact post-id stem match

- tests/core/test_concurrency.py (2):
  - get_app_loop raises RuntimeError before set_app_loop is called
  - run_on_app_loop round-trips a coroutine result from a worker
    thread loop back to the test thread

- tests/core/api/test_base.py (1):
  - BooruClient._shared_client lazy singleton constructor-once under
    10-thread first-call race

Plus tests/conftest.py with fixtures: tmp_db, tmp_library,
reset_app_loop, reset_shared_clients. All fixtures use tmp_path or
reset module-level globals around the test so the suite is parallel-
safe.

pyproject.toml:
- New [project.optional-dependencies] test extra: pytest>=8.0,
  pytest-asyncio>=0.23
- New [tool.pytest.ini_options]: asyncio_mode = "auto",
  testpaths = ["tests"]

README.md:
- Linux install section gains "Run tests" with the
  pip install -e ".[test]" + pytest tests/ invocation

Phase B (post-sweep VideoPlayer regression tests for the seek slider
pin, _pending_mute lazy replay, and volume replay) is deferred to
Prompt 3's state machine work — VideoPlayer cannot be instantiated
without QApplication and a real mpv, which is out of scope for a
unit test suite. Once the state machine carves the pure-Python state
out of VideoPlayer, those tests become trivial against the helper
module.

Suite runs in 0.07s (16 tests). Independent of Qt/mpv/network/ffmpeg.

Test cases for Prompt 3:
- (already covered) — this IS the test suite Prompt 3 builds on top of
2026-04-08 18:50:00 -05:00

63 lines
1.8 KiB
Python

"""Tests for `booru_viewer.core.concurrency` — the persistent-loop handle.
Locks in:
- `get_app_loop` raises a clear RuntimeError if `set_app_loop` was never
called (the production code uses this to bail loudly when async work
is scheduled before the loop thread starts)
- `run_on_app_loop` round-trips a coroutine result from a worker-thread
loop back to the calling thread via `concurrent.futures.Future`
"""
from __future__ import annotations
import asyncio
import threading
import pytest
from booru_viewer.core import concurrency
from booru_viewer.core.concurrency import (
get_app_loop,
run_on_app_loop,
set_app_loop,
)
def test_get_app_loop_raises_before_set(reset_app_loop):
"""Calling `get_app_loop` before `set_app_loop` is a configuration
error — the production code expects a clear RuntimeError so callers
bail loudly instead of silently scheduling work onto a None loop."""
with pytest.raises(RuntimeError, match="not initialized"):
get_app_loop()
def test_run_on_app_loop_round_trips_result(reset_app_loop):
"""Spin up a real asyncio loop in a worker thread, register it via
`set_app_loop`, then from the test (main) thread schedule a coroutine
via `run_on_app_loop` and assert the result comes back through the
`concurrent.futures.Future` interface."""
loop = asyncio.new_event_loop()
ready = threading.Event()
def _run_loop():
asyncio.set_event_loop(loop)
ready.set()
loop.run_forever()
t = threading.Thread(target=_run_loop, daemon=True)
t.start()
ready.wait(timeout=2)
try:
set_app_loop(loop)
async def _produce():
return 42
fut = run_on_app_loop(_produce())
assert fut.result(timeout=2) == 42
finally:
loop.call_soon_threadsafe(loop.stop)
t.join(timeout=2)
loop.close()