bbchops/transcription
1
0
Fork 0
generated from john/python-template
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

Continue GC code review: Pydantic

Browse Source
This commit is contained in:
Jim Lancaster
2026-08-12 01:35:50 -05:00
parent 888a8c380a
commit 1e8d8572d4
18 changed files with 578 additions and 289 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/api/test_v4_documents.py
+32
View File
@@ -117,6 +117,25 @@ def test_set_document_type_by_code_updates_canonical_fields(tmp_path):
assert payload["document_type_code"] == "record"
def test_document_type_payload_requires_exactly_one_selector(tmp_path):
with _v4_api_client(tmp_path, db_filename="api-doc-type-validation.db") as (client, db_url):
document_id, _ = _seed_document_and_person(db_url=db_url)
missing = client.put(f"/api/v4/documents/{document_id}/type", json={})
conflicting = client.put(
f"/api/v4/documents/{document_id}/type",
json={"document_type_id": str(UUID(int=1)), "document_type_code": "record"},
)
unexpected = client.put(
f"/api/v4/documents/{document_id}/type",
json={"document_type_code": "record", "ignored": True},
)
assert missing.status_code == 422
assert conflicting.status_code == 422
assert unexpected.status_code == 422
def test_document_people_role_aware_write_read_and_delete(tmp_path):
with _v4_api_client(tmp_path, db_filename="api-links.db") as (client, db_url):
document_id, person_id = _seed_document_and_person(db_url=db_url)
@@ -155,6 +174,19 @@ def test_document_people_role_aware_write_read_and_delete(tmp_path):
assert list_after_delete.json()["links"] == []
def test_document_person_link_defaults_to_author_when_role_is_omitted(tmp_path):
with _v4_api_client(tmp_path, db_filename="api-default-role.db") as (client, db_url):
document_id, person_id = _seed_document_and_person(db_url=db_url)
response = client.post(
f"/api/v4/documents/{document_id}/people",
json={"person_id": str(person_id)},
)
assert response.status_code == 200
assert response.json()["role_code"] == "author"
def test_duplicate_document_person_link_returns_conflict_envelope(tmp_path):
with _v4_api_client(tmp_path, db_filename="api-dup.db") as (client, db_url):
document_id, person_id = _seed_document_and_person(db_url=db_url)
tests/integration/test_pipeline_flow.py
+6 -1
View File
@@ -9,6 +9,8 @@ from transcription.config import Settings
from transcription.db.models import Document
from transcription.db.models import JobSourceStatus
from transcription.db.models import JobStatus
from transcription.providers.base import ProviderUsage
from transcription.providers.base import TranscriptionMetadata
from transcription.providers.base import TranscriptionResult
from transcription.services import ServiceBundle
from transcription.services.store import create_document_job
@@ -84,7 +86,10 @@ class TestPipelineSuccessFlow:
provider="openrouter",
model="test-model",
prompt_name="transcribe_document.md",
ai_metadata={"finish_reason": "stop", "usage": {"total_tokens": 42}},
metadata=TranscriptionMetadata(
finish_reason="stop",
usage=ProviderUsage(total_tokens=42),
),
raw_api_response={"id": "resp_123", "choices": [{"message": {"content": "Pipeline transcript"}}]},
)
tests/providers/test_openrouter.py
+23 -1
View File
@@ -122,7 +122,7 @@ class TestOpenRouterProviderTranscribe:
assert result.provider == "openrouter"
assert result.prompt_name is None
assert result.model == "vendor/model-b"
assert result.ai_metadata == {
assert result.metadata_payload() == {
"finish_reason": "stop",
"usage": {"input_tokens": 10, "output_tokens": 25, "total_tokens": 35},
}
@@ -178,3 +178,25 @@ class TestOpenRouterProviderTranscribe:
image_bytes=b"img-bytes",
mime_type="image/png",
)
@pytest.mark.asyncio
async def test_ignores_invalid_token_metadata_without_discarding_transcript(self):
response = {
"model": "vendor/model-a",
"choices": [{"message": {"content": "Transcript text"}}],
"usage": {"prompt_tokens": -1},
}
provider = OpenRouterTranscriptionProvider(
settings=Settings(openrouter_api_key="test-key"),
client=_FakeClient(response=response),
)
result = await provider.transcribe(
prompt_text="Prompt body",
image_bytes=b"img-bytes",
mime_type="image/png",
)
assert result.text == "Transcript text"
assert result.metadata_payload() is None
assert result.raw_api_response == response
tests/services/test_document_service.py
+7 -6
View File
@@ -7,6 +7,7 @@ from uuid import uuid4
import pytest
from sqlmodel import select
from transcription.config import Settings
from transcription.db.models import Document
from transcription.db.models import DocumentPerson
from transcription.db.models import DocumentPersonRole
@@ -98,8 +99,8 @@ async def test_delete_document_blocks_when_dependencies_exist(default_session_fa
@pytest.mark.asyncio
async def test_delete_document_succeeds_when_unlinked(default_session_factory, tmp_path):
service = DocumentService(session_factory=default_session_factory)
service.settings.upload_dir = tmp_path
settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
service = DocumentService(session_factory=default_session_factory, settings=settings)
document = await service.create_document(
Document(
@@ -123,9 +124,9 @@ async def test_delete_document_succeeds_when_unlinked(default_session_factory, t
@pytest.mark.asyncio
async def test_delete_document_removes_person_links(default_session_factory, tmp_path):
service = DocumentService(session_factory=default_session_factory)
settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
service = DocumentService(session_factory=default_session_factory, settings=settings)
people_service = PeopleService(session_factory=default_session_factory)
service.settings.upload_dir = tmp_path
document = await service.create_document(
Document(
@@ -163,8 +164,8 @@ async def test_delete_document_removes_person_links(default_session_factory, tmp
@pytest.mark.asyncio
async def test_delete_document_removes_populated_storage_tree(default_session_factory, tmp_path):
service = DocumentService(session_factory=default_session_factory)
service.settings.upload_dir = tmp_path
settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
service = DocumentService(session_factory=default_session_factory, settings=settings)
document = await service.create_document(
Document(
tests/services/test_transcription_service.py
+5 -4
View File
@@ -4,6 +4,7 @@ from uuid import uuid4
import pytest
from transcription.config import Settings
from transcription.db.models import Document
from transcription.db.models import Job
from transcription.db.models import JobSource
@@ -102,8 +103,8 @@ class TestSourceServiceRevisionUpsert:
):
documents = DocumentService(session_factory=default_session_factory)
jobs = JobService(session_factory=default_session_factory)
transcriptions = SourceService(session_factory=default_session_factory)
transcriptions.settings.upload_dir = tmp_path
settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
transcriptions = SourceService(session_factory=default_session_factory, settings=settings)
document = Document(id=uuid4(), name="delete-source-success")
await documents.create_document(document=document)
@@ -174,8 +175,8 @@ class TestSourceServiceRevisionUpsert:
@pytest.mark.asyncio
async def test_delete_unlinked_source_succeeds(self, default_session_factory, tmp_path):
documents = DocumentService(session_factory=default_session_factory)
transcriptions = SourceService(session_factory=default_session_factory)
transcriptions.settings.upload_dir = tmp_path
settings = Settings(openrouter_api_key="test-key", upload_dir=tmp_path)
transcriptions = SourceService(session_factory=default_session_factory, settings=settings)
document = Document(id=uuid4(), name="delete-unlinked-source")
await documents.create_document(document=document)
tests/test_config.py
+25 -2
View File
@@ -24,7 +24,7 @@ class TestSettingsLoading:
"""Settings constructs when OPENROUTER_API_KEY is provided."""
monkeypatch.setenv("OPENROUTER_API_KEY", "test-key-xyz")
settings = Settings()
assert settings.openrouter_api_key == "test-key-xyz"
assert settings.openrouter_api_key.get_secret_value() == "test-key-xyz"
def test_requires_api_key(self, monkeypatch):
"""Settings raises ValidationError when OPENROUTER_API_KEY is missing."""
@@ -52,7 +52,7 @@ class TestSettingsLoading:
]
)
assert settings.openrouter_api_key == "test-key"
assert settings.openrouter_api_key.get_secret_value() == "test-key"
assert settings.port == 8123
assert settings.reload is True
@@ -77,6 +77,29 @@ class TestProviderSettings:
assert settings.openrouter_http_referer is None
assert settings.openrouter_app_title is None
@pytest.mark.parametrize(
("field", "value"),
[
("transcription_temperature", -0.1),
("transcription_temperature", 2.1),
("transcription_top_p", -0.1),
("transcription_top_p", 1.1),
],
)
def test_rejects_sampling_values_outside_provider_ranges(self, field, value):
with pytest.raises(ValidationError):
_make_settings(**{field: value})
def test_rejects_prompt_paths_outside_prompt_directory(self):
with pytest.raises(ValidationError):
_make_settings(default_prompt_name="../secret.md")
def test_settings_are_immutable_runtime_snapshots(self):
settings = _make_settings()
with pytest.raises(ValidationError):
settings.port = 9000
def test_provider_model_accepts_env_default(self, monkeypatch):
"""provider_model is sourced when provided through environment configuration."""
monkeypatch.setenv("PROVIDER_MODEL", "google/gemini-2.5-flash")
tests/test_prompts.py
+52
View File
@@ -1,7 +1,17 @@
"""Tests for prompt artifacts in prompts/."""
import hashlib
from pathlib import Path
import pytest
from pydantic import ValidationError
from transcription.config import Settings
from transcription.services.sources import PromptExecution
from transcription.services.sources import PromptLoadError
from transcription.services.sources import build_prompt_execution
from transcription.services.sources import load_prompt_text
PROMPT_PATH = Path("prompts/transcribe_document.md")
@@ -39,3 +49,45 @@ class TestPromptArtifact:
text = _prompt_text().lower()
assert "[deleted:" in text
assert "[inserted:" in text
class TestPromptConfiguration:
def test_builds_validated_immutable_prompt_provenance(self, tmp_path):
prompt_text = "Transcribe this document verbatim."
(tmp_path / "custom.md").write_text(prompt_text, encoding="utf-8")
settings = Settings(
openrouter_api_key="test-key",
prompt_dir=tmp_path,
default_prompt_name="custom.md",
transcription_temperature=0.2,
transcription_top_p=0.9,
)
execution = build_prompt_execution(settings=settings)
assert execution.prompt_hash == hashlib.sha256(prompt_text.encode()).hexdigest()
assert execution.temperature == 0.2
assert execution.top_p == 0.9
with pytest.raises(ValidationError):
execution.prompt_name = "changed.md"
def test_rejects_prompt_path_traversal_even_with_direct_loader_call(self, tmp_path):
outside_prompt = tmp_path / "outside.md"
prompt_dir = tmp_path / "prompts"
prompt_dir.mkdir()
outside_prompt.write_text("secret", encoding="utf-8")
settings = Settings(openrouter_api_key="test-key", prompt_dir=prompt_dir)
with pytest.raises(PromptLoadError):
load_prompt_text(prompt_name="../outside.md", settings=settings)
def test_prompt_execution_rejects_invalid_provenance_hash(self):
with pytest.raises(ValidationError):
PromptExecution(
prompt_name="prompt.md",
prompt_hash="not-a-sha256",
system_prompt=None,
user_prompt="text",
temperature=None,
top_p=None,
)