generated from john/python-template
zoltan57andCopilot App
4e8c562f92
MED-01 - move remaining blocking work off the event loop: - normalization.py gains normalize_orientation_async; the Pillow decode, transpose, and re-encode now run via asyncio.to_thread. The sync entry point stays for tests and documents that it blocks. - OrientationNormalization.digest_sha256 becomes a stored field computed inside normalize_orientation, which already runs off-loop, instead of a property that hashed page-sized derivative bytes on the caller's thread. - SourceService._write_and_digest_artifact performs the artifact write and its sha256 in a single worker-thread hop; both external-artifact write sites are now dispatched through to_thread. - transcribe_image dispatches load_source_payload and build_prompt_execution through to_thread. MED-04 - replace functools.cache on the engine and session factories with explicit URL-keyed registries. dispose_engine and dispose_session_factory now evict only the requested URL; previously cache_clear() tore down every other database in the process, and dispose_engine would construct an engine for an unknown URL purely to throw it away. New tests/test_engine_registry.py covers distinct engines per URL, targeted eviction, and the unknown-URL no-op. config.py - replace object.__setattr__ in normalize_provider_models with a model_validator(mode="before") over the raw input, so the derived selector is produced by normal construction rather than by mutating a frozen instance. model_copy(update=...) was tried first and rejected: pydantic-settings does not support a top-level validator returning anything other than self when validating via __init__. provider_model is now stripped as well as the tuple entries. models.py - add onupdate to the five updated_at columns and to Job.date_updated, and drop the 10 manual "updated_at = datetime.now(UTC)" assignments across the document, job, people, registry, and source services. Verified DDL-neutral by hashing CreateTable output for every table on both the sqlite and postgresql dialects before and after: identical, so this stays in Phase 6 and Phase 2 does not need re-verification. New tests/services/test_timestamps.py asserts an update through each service advances the timestamp. MED-08 - Job.filename no longer swallows every exception to None. Relationships declare lazy="raise", so the new _loaded_attribute helper inspects load state explicitly and returns None only for genuinely unloaded attributes; real errors now surface. Job.error_detail uses the same helper, which also removes its unguarded read of the lazy="raise" job_sources relationship. Verification: ruff check src tests clean; 288 passed, 4 skipped. Co-authored-by: Copilot App <[email protected]>
54 lines
1.7 KiB
Python
54 lines
1.7 KiB
Python
"""Phase 6 verification for the URL-keyed engine and session-factory registries.
|
|
|
|
Covers [MED-04]: replacing `functools.cache` with an explicit registry so that
|
|
disposing one database's engine cannot silently tear down every other one.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from transcription.db.engine import dispose_engine
|
|
from transcription.db.engine import get_engine
|
|
from transcription.db.session import dispose_session_factory
|
|
from transcription.db.session import get_session_factory
|
|
|
|
URL_A = "sqlite+aiosqlite:///./.registry-test-a.db"
|
|
URL_B = "sqlite+aiosqlite:///./.registry-test-b.db"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_distinct_urls_produce_distinct_engines_and_eviction_is_targeted():
|
|
engine_a = get_engine(URL_A)
|
|
engine_b = get_engine(URL_B)
|
|
|
|
assert engine_a is not engine_b
|
|
assert get_engine(URL_A) is engine_a
|
|
|
|
await dispose_engine(URL_A)
|
|
|
|
assert get_engine(URL_B) is engine_b, "disposing one URL must not evict the others"
|
|
assert get_engine(URL_A) is not engine_a, "the disposed URL must be rebuilt on demand"
|
|
|
|
await dispose_engine(URL_A)
|
|
await dispose_engine(URL_B)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disposing_an_unregistered_url_is_a_noop():
|
|
await dispose_engine("sqlite+aiosqlite:///./.registry-test-never-created.db")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_factory_eviction_is_targeted():
|
|
factory_a = get_session_factory(URL_A)
|
|
factory_b = get_session_factory(URL_B)
|
|
|
|
assert factory_a is not factory_b
|
|
|
|
await dispose_session_factory(URL_A)
|
|
|
|
assert get_session_factory(URL_B) is factory_b
|
|
assert get_session_factory(URL_A) is not factory_a
|
|
|
|
await dispose_session_factory(URL_A)
|
|
await dispose_session_factory(URL_B)
|