pax
bf14466382
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
78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
"""Tests for `booru_viewer.core.api.base` — the lazy `_shared_client`
|
|
singleton on `BooruClient`.
|
|
|
|
Locks in the lock-and-recheck pattern at `base.py:90-108`. Without it,
|
|
two threads racing on first `.client` access would both see
|
|
`_shared_client is None`, both build an `httpx.AsyncClient`, and one of
|
|
them would leak (overwritten without aclose).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
import pytest
|
|
|
|
from booru_viewer.core.api.base import BooruClient
|
|
|
|
|
|
class _StubClient(BooruClient):
|
|
"""Concrete subclass so we can instantiate `BooruClient` for the test
|
|
— the base class has abstract `search` / `get_post` methods."""
|
|
api_type = "stub"
|
|
|
|
async def search(self, tags="", page=1, limit=40):
|
|
return []
|
|
|
|
async def get_post(self, post_id):
|
|
return None
|
|
|
|
|
|
def test_shared_client_singleton_under_concurrency(reset_shared_clients):
|
|
"""N threads racing on first `.client` access must result in exactly
|
|
one `httpx.AsyncClient` constructor call. The threading.Lock guards
|
|
the check-and-set so the second-and-later callers re-read the now-set
|
|
`_shared_client` after acquiring the lock instead of building their
|
|
own."""
|
|
constructor_calls = 0
|
|
constructor_lock = threading.Lock()
|
|
|
|
def _fake_async_client(*args, **kwargs):
|
|
nonlocal constructor_calls
|
|
with constructor_lock:
|
|
constructor_calls += 1
|
|
m = MagicMock()
|
|
m.is_closed = False
|
|
return m
|
|
|
|
# Barrier so all threads hit the property at the same moment
|
|
n_threads = 10
|
|
barrier = threading.Barrier(n_threads)
|
|
results = []
|
|
results_lock = threading.Lock()
|
|
|
|
client_instance = _StubClient("http://example.test")
|
|
|
|
def _worker():
|
|
barrier.wait()
|
|
c = client_instance.client
|
|
with results_lock:
|
|
results.append(c)
|
|
|
|
with patch("booru_viewer.core.api.base.httpx.AsyncClient",
|
|
side_effect=_fake_async_client):
|
|
threads = [threading.Thread(target=_worker) for _ in range(n_threads)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join(timeout=5)
|
|
|
|
assert constructor_calls == 1, (
|
|
f"Expected exactly one httpx.AsyncClient construction, "
|
|
f"got {constructor_calls}"
|
|
)
|
|
# All threads got back the same shared instance
|
|
assert len(results) == n_threads
|
|
assert all(r is results[0] for r in results)
|