diff --git a/cueweaver/adapters/sqlite_term_maps.py b/cueweaver/adapters/sqlite_term_maps.py index 1c04650..0fff1df 100644 --- a/cueweaver/adapters/sqlite_term_maps.py +++ b/cueweaver/adapters/sqlite_term_maps.py @@ -35,7 +35,7 @@ def __init__(self, database: SqliteDatabase) -> None: def list(self) -> list[TermMapSummary]: try: - with self._database.session() as session: + with self._database.read_session() as session: rows = session.execute( select(TermMapRow, func.count(TermMapEntryRow.position)) .outerjoin( @@ -53,7 +53,7 @@ def list(self) -> list[TermMapSummary]: def get(self, term_map_id: str) -> TermMapDetail: try: - with self._database.session() as session: + with self._database.read_session() as session: row = _require_row(session, term_map_id) entries = session.scalars( select(TermMapEntryRow) @@ -78,8 +78,7 @@ def get(self, term_map_id: str) -> TermMapDetail: def create(self, name: str, content: Mapping[str, str]) -> TermMapSummary: timestamp = _utc_timestamp() try: - with self._database.session() as session: - session.connection().exec_driver_sql("BEGIN IMMEDIATE") + with self._database.write_transaction(immediate=True) as session: sequence = session.scalar(select(func.max(TermMapRow.sequence))) row = TermMapRow( id=_new_id(), @@ -91,7 +90,6 @@ def create(self, name: str, content: Mapping[str, str]) -> TermMapSummary: session.add(row) _replace_entries(session, row.id, content) count = _entry_count(session, row.id) - session.commit() return _summary(row, count) except ( DatabaseOpenError, @@ -103,13 +101,12 @@ def create(self, name: str, content: Mapping[str, str]) -> TermMapSummary: def rename(self, term_map_id: str, name: str) -> TermMapSummary: try: - with self._database.session() as session: + with self._database.write_transaction() as session: row = _require_row(session, term_map_id) row.name = name row.name_folded = name.casefold() row.updated_at = _utc_timestamp() count = _entry_count(session, term_map_id) - session.commit() return _summary(row, int(count or 0)) except ServiceError: raise @@ -123,12 +120,11 @@ def rename(self, term_map_id: str, name: str) -> TermMapSummary: def replace(self, term_map_id: str, content: Mapping[str, str]) -> TermMapSummary: try: - with self._database.session() as session: + with self._database.write_transaction() as session: row = _require_row(session, term_map_id) _replace_entries(session, term_map_id, content) row.updated_at = _utc_timestamp() count = _entry_count(session, term_map_id) - session.commit() return _summary(row, count) except ServiceError: raise @@ -143,7 +139,7 @@ def replace(self, term_map_id: str, content: Mapping[str, str]) -> TermMapSummar def delete(self, term_map_id: str, name: str) -> TermMapSummary: try: - with self._database.session() as session: + with self._database.write_transaction() as session: row = _require_row(session, term_map_id) if name != row.name: raise ServiceError( @@ -154,7 +150,6 @@ def delete(self, term_map_id: str, name: str) -> TermMapSummary: count = _entry_count(session, term_map_id) summary = _summary(row, int(count or 0)) session.delete(row) - session.commit() return summary except ServiceError: raise @@ -176,7 +171,7 @@ def __init__(self, database: SqliteDatabase) -> None: def snapshot_bindings(self) -> dict[str, str]: try: - with self._database.session() as session: + with self._database.read_session() as session: rows = session.scalars( select(DirectoryTermMapBindingRow).order_by( DirectoryTermMapBindingRow.directory @@ -195,7 +190,7 @@ def bind( term_map_id: str, ) -> None: try: - with self._database.session() as session: + with self._database.write_transaction() as session: row = session.get(DirectoryTermMapBindingRow, directory) if row is None: session.add( @@ -205,7 +200,6 @@ def bind( ) else: row.term_map_id = term_map_id - session.commit() except ServiceError: raise except (DatabaseOpenError, DatabasePathError) as error: @@ -225,13 +219,12 @@ def bind( def remove(self, directory: str) -> None: try: - with self._database.session() as session: + with self._database.write_transaction() as session: session.execute( delete(DirectoryTermMapBindingRow).where( DirectoryTermMapBindingRow.directory == directory ) ) - session.commit() except (DatabaseOpenError, DatabasePathError) as error: raise ServiceError( "directory_term_maps_unavailable", diff --git a/cueweaver/application/database.py b/cueweaver/application/database.py index dc6c823..300d11a 100644 --- a/cueweaver/application/database.py +++ b/cueweaver/application/database.py @@ -125,7 +125,7 @@ class DatabaseOpenError(sqlite3.Error): class SqliteDatabase: - """Run migrations once and provide short-lived ORM sessions.""" + """Run migrations once and provide explicit ORM read/write scopes.""" def __init__(self, path: Path) -> None: self.path = path @@ -139,12 +139,25 @@ def initialize(self) -> None: self._initialize() @contextmanager - def session(self) -> Iterator[Session]: + def read_session(self) -> Iterator[Session]: self.initialize() assert self._session_factory is not None session = self._session_factory() try: yield session + finally: + session.close() + + @contextmanager + def write_transaction(self, *, immediate: bool = False) -> Iterator[Session]: + self.initialize() + assert self._session_factory is not None + session = self._session_factory() + try: + if immediate: + session.connection().exec_driver_sql("BEGIN IMMEDIATE") + yield session + session.commit() except Exception: session.rollback() raise diff --git a/cueweaver/application/jobs/store.py b/cueweaver/application/jobs/store.py index 3947fe9..8e3978e 100644 --- a/cueweaver/application/jobs/store.py +++ b/cueweaver/application/jobs/store.py @@ -45,7 +45,7 @@ def __init__(self, database: SqliteDatabase) -> None: def load(self) -> list[JobRecord]: try: - with self._database.session() as session: + with self._database.read_session() as session: rows = session.scalars( select(JobRow).order_by( JobRow.queue_sequence, JobRow.created_at, JobRow.id @@ -63,11 +63,10 @@ def write(self, record: JobRecord) -> None: def remove(self, job_id: str) -> None: _require_valid_job_id(job_id) try: - with self._database.session() as session: + with self._database.write_transaction() as session: row = session.get(JobRow, job_id) if row is not None: session.delete(row) - session.commit() except (sqlite3.Error, SQLAlchemyError) as error: raise ServiceError( "database_unavailable", "Job record could not be deleted" @@ -76,9 +75,8 @@ def remove(self, job_id: str) -> None: def _upsert(self, record: JobRecord) -> None: prepared = self._prepare_record(record) try: - with self._database.session() as session: + with self._database.write_transaction() as session: _upsert_row(session, prepared) - session.commit() except (sqlite3.Error, SQLAlchemyError) as error: raise ServiceError( "database_unavailable", "Job record could not be persisted" diff --git a/tests/test_database.py b/tests/test_database.py index bcc5e5a..2013300 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -7,7 +7,13 @@ from alembic.config import Config from sqlalchemy import URL, create_engine -from cueweaver.application.database import SqliteDatabase, _migration_config +from cueweaver.application.database import ( + JobRow, + JobStatusHistoryRow, + JobTermMapSnapshotRow, + SqliteDatabase, + _migration_config, +) from cueweaver.application.jobs.store import SqliteJobRecordStore @@ -42,10 +48,98 @@ def test_sqlite_database_bootstraps_the_application_schema(tmp_path: Path): row[1] for row in connection.execute("PRAGMA table_info(jobs)") }.isdisjoint({"record_json", "content_json"}) - with database.session() as session: + with database.read_session() as session: assert session.connection().exec_driver_sql("PRAGMA foreign_keys").scalar() == 1 +def test_write_transaction_commits_a_multi_table_change(tmp_path: Path): + database = SqliteDatabase(tmp_path / "cueweaver.sqlite3") + + with database.write_transaction() as session: + _add_job_rows(session) + + with database.read_session() as session: + assert session.get(JobRow, "job-1") is not None + assert session.get(JobStatusHistoryRow, {"job_id": "job-1", "sequence": 0}) + assert session.get(JobTermMapSnapshotRow, {"job_id": "job-1", "position": 0}) + + +def test_write_transaction_rolls_back_when_the_scope_fails(tmp_path: Path): + database = SqliteDatabase(tmp_path / "cueweaver.sqlite3") + + with ( + pytest.raises(RuntimeError, match="abort"), + database.write_transaction() as session, + ): + _add_job_rows(session) + raise RuntimeError("abort") + + with database.read_session() as session: + assert session.get(JobRow, "job-1") is None + assert ( + session.get(JobStatusHistoryRow, {"job_id": "job-1", "sequence": 0}) is None + ) + assert ( + session.get(JobTermMapSnapshotRow, {"job_id": "job-1", "position": 0}) + is None + ) + + +def test_sqlite_database_can_close_and_reinitialize_between_scopes(tmp_path: Path): + database = SqliteDatabase(tmp_path / "cueweaver.sqlite3") + + with database.write_transaction() as session: + _add_job_rows(session) + database.close() + + with database.read_session() as session: + assert session.get(JobRow, "job-1") is not None + database.close() + + +def _job_row() -> JobRow: + return JobRow( + id="job-1", + schema_version=1, + status="Queued", + attempt=1, + created_at="2026-08-24T00:00:00Z", + queue_sequence=0, + media_path="Movie.mkv", + subtitle_path="Movie.en.srt", + target_language_code="zh-Hans", + term_map_mode="none", + output_path="Movie.zh-Hans.srt", + source_format="srt", + dynamic_terminology_enabled=True, + subtitle_terminology_filter_enabled=True, + output_suffix="zh-Hans", + output_conflict_policy="skip", + ) + + +def _add_job_rows(session) -> None: + session.add(_job_row()) + session.add( + JobStatusHistoryRow( + job_id="job-1", + sequence=0, + status="Queued", + attempt=1, + started_at="2026-08-24T00:00:00Z", + ) + ) + session.add( + JobTermMapSnapshotRow( + job_id="job-1", + position=0, + source="Captain", + source_folded="captain", + target="队长", + ) + ) + + def test_migration_discards_issue_193_application_data(tmp_path: Path): path = tmp_path / "cueweaver.sqlite3" record = { diff --git a/tests/test_jobs.py b/tests/test_jobs.py index fdd4806..a220a45 100644 --- a/tests/test_jobs.py +++ b/tests/test_jobs.py @@ -217,6 +217,45 @@ def test_sqlite_record_store_persists_records_and_uses_a_transactional_database( assert store.load() == [] +def test_sqlite_job_write_rolls_back_related_rows_on_database_failure(tmp_path: Path): + database_path = tmp_path / "cueweaver.sqlite3" + database = SqliteDatabase(database_path) + database.initialize() + with sqlite3.connect(database_path) as connection: + connection.execute( + """ + CREATE TRIGGER fail_job_snapshot_insert + BEFORE INSERT ON job_term_map_snapshots + BEGIN + SELECT RAISE(ABORT, 'forced snapshot failure'); + END + """ + ) + + record = persisted_job_record("atomic-job") + request = record["request"] + assert isinstance(request, dict) + request["term_map_mode"] = "selected" + request["term_map"] = { + "id": "map-1", + "name": "Characters", + "content": {"Captain": "队长"}, + } + store = SqliteJobRecordStore(database) + + with pytest.raises(ServiceError, match="could not be persisted"): + store.write(record) + + with sqlite3.connect(database_path) as connection: + assert connection.execute("SELECT COUNT(*) FROM jobs").fetchone() == (0,) + assert connection.execute( + "SELECT COUNT(*) FROM job_status_history" + ).fetchone() == (0,) + assert connection.execute( + "SELECT COUNT(*) FROM job_term_map_snapshots" + ).fetchone() == (0,) + + def test_sqlite_record_store_rejects_a_future_schema_version(tmp_path: Path): store = SqliteJobRecordStore(SqliteDatabase(tmp_path / "cueweaver.sqlite3")) store.write(persisted_job_record("future-schema")) diff --git a/tests/test_term_maps.py b/tests/test_term_maps.py index 6ad8d22..973d1fe 100644 --- a/tests/test_term_maps.py +++ b/tests/test_term_maps.py @@ -6,6 +6,10 @@ from fastapi.testclient import TestClient from test_term_map_helpers import make_client +from cueweaver.adapters.sqlite_term_maps import SqliteTermMapStore +from cueweaver.application.database import SqliteDatabase +from cueweaver.application.errors import ServiceError + def create_term_map(client: TestClient, name: str = "Characters") -> dict[str, object]: return client.post( @@ -210,6 +214,51 @@ def test_term_map_listing_preserves_creation_order(tmp_path: Path): ] == ["First", "Second", "Third"] +def test_concurrent_term_map_creation_allocates_each_sequence_once(tmp_path: Path): + database = SqliteDatabase(tmp_path / "cueweaver.sqlite3") + store = SqliteTermMapStore(database) + + with ThreadPoolExecutor(max_workers=4) as executor: + created = list( + executor.map( + lambda index: store.create(f"Map {index}", {"a": "b"}), + range(8), + ) + ) + + assert len({summary.id for summary in created}) == 8 + with sqlite3.connect(tmp_path / "cueweaver.sqlite3") as connection: + assert [ + row[0] + for row in connection.execute( + "SELECT sequence FROM term_maps ORDER BY sequence" + ) + ] == list(range(8)) + + +def test_term_map_replacement_rolls_back_entries_and_metadata_on_failure( + tmp_path: Path, +): + database_path = tmp_path / "cueweaver.sqlite3" + store = SqliteTermMapStore(SqliteDatabase(database_path)) + created = store.create("Characters", {"Captain": "队长"}) + with sqlite3.connect(database_path) as connection: + connection.execute( + """ + CREATE TRIGGER fail_term_map_entry_insert + BEFORE INSERT ON term_map_entries + BEGIN + SELECT RAISE(ABORT, 'forced entry failure'); + END + """ + ) + + with pytest.raises(ServiceError, match="cannot be saved"): + store.replace(created.id, {"Captain": "舰长"}) + + assert store.get(created.id).content == {"Captain": "队长"} + + def test_unknown_term_map_api_path_remains_a_structured_not_found(tmp_path: Path): response = make_client(tmp_path).post("/api/term-maps/map-1/unknown")