"""test_audible_convert.""" from __future__ import annotations import json import subprocess import pytest from sqlalchemy import create_engine, select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session, sessionmaker from python.orm.richie import Audiobook, AudiobookAuthor, AudiobookSeries, RichieBase from python.tools.audiobook import audible_convert, metadata_agent from python.tools.audiobook.metadata_agent import StandardBookMetadata, standard_book_metadata class FakeOllamaResponse: def __init__(self, payload): self._payload = payload def raise_for_status(self): return None def json(self): return self._payload class FakeFfprobeError(RuntimeError): def __str__(self): return "bad ffprobe" @pytest.fixture def audiobook_engine(): engine = create_engine("sqlite+pysqlite:///:memory:", future=True) RichieBase.metadata.create_all(engine) with sessionmaker(bind=engine, expire_on_commit=False, future=True)() as session: session.add_all( [ AudiobookAuthor(id=1, name="glynn_stewart"), AudiobookAuthor(id=2, name="craig_alanson"), AudiobookAuthor(id=4, name="dennis_e_taylor"), AudiobookSeries(id=1, name="starships_mage", author_id=1), AudiobookSeries(id=2, name="black_fleet_trilogy", author_id=1), AudiobookSeries(id=3, name="expeditionary_force", author_id=2), AudiobookSeries(id=4, name="bobiverse", author_id=4), ], ) session.commit() yield engine engine.dispose() def install_fake_ollama(monkeypatch, payloads): calls = [] def fake_post(*args, **kwargs): calls.append((args, kwargs)) return FakeOllamaResponse(payloads.pop(0)) monkeypatch.setattr(metadata_agent.httpx, "post", fake_post) return calls def conversion_config(output_directory, *, dry_run=False, overwrite=False): return audible_convert.ConversionConfig( resolved_output=output_directory, ollama_api_key="test-key", agent_config=metadata_agent.AgentConfig(), engine=create_engine("sqlite+pysqlite:///:memory:"), activation_bytes=None, dry_run=dry_run, overwrite=overwrite, ) def sqlite_engine(): return create_engine("sqlite+pysqlite:///:memory:") def tool_response(name, arguments): return { "message": { "role": "assistant", "content": "", "tool_calls": [{"function": {"name": name, "arguments": arguments}}], }, } def final_response(metadata): return {"message": {"role": "assistant", "content": json.dumps(metadata)}} def fenced_final_response(metadata): return {"message": {"role": "assistant", "content": f"```json\n{json.dumps(metadata)}\n```"}} def test_output_stem_uses_catalog_slugs() -> None: metadata = StandardBookMetadata( author_id=1, author="glynn_stewart", book_id=None, title="title-slug", series_id=1, series="starships_mage", series_index=1, confidence=0.96, needs_review=False, evidence=["test"], ) assert audible_convert.output_stem(metadata) == "glynn_stewart-starships_mage_01-title-slug" def test_output_stem_formats_half_series_index() -> None: metadata = StandardBookMetadata( author_id=1, author="glynn_stewart", book_id=None, title="title-slug", series_id=1, series="starships_mage", series_index=1.5, confidence=0.96, needs_review=False, evidence=["test"], ) assert audible_convert.output_stem(metadata) == "glynn_stewart-starships_mage_01.5-title-slug" @pytest.mark.parametrize( ("metadata", "expected"), [ ( StandardBookMetadata( author_id=1, author="mark_e_cooper", book_id=None, title="merkiaari-wars-series-books-1-3", series_id=1, series="merkiaari_wars", series_index=1, confidence=0.96, needs_review=False, evidence=["test"], ), "mark_e_cooper-merkiaari_wars_01-03-merkiaari-wars-series-books-1-3", ), ( StandardBookMetadata( author_id=1, author="rhett_c_bruno", book_id=None, title="the-circuit-books-1-3", series_id=1, series="the_circuit", series_index=1, confidence=0.96, needs_review=False, evidence=["test"], ), "rhett_c_bruno-the_circuit_01-03-the-circuit-books-1-3", ), ], ) def test_output_stem_formats_omnibus_book_range(metadata, expected) -> None: assert audible_convert.output_stem(metadata) == expected def test_convert_aax_file_runs_ffmpeg(tmp_path, monkeypatch) -> None: """test_convert_aax_file_runs_ffmpeg.""" commands = [] def fake_run_command(arguments, *, capture=False): assert capture is False commands.append(arguments) return subprocess.CompletedProcess(arguments, 0, "", "") source = tmp_path / "book.aax" destination = tmp_path / "book" / "book.m4b" monkeypatch.setattr(audible_convert, "run_command", fake_run_command) audible_convert.convert_aax_file(source, destination, "abc123", overwrite=False) assert commands == [ [ "ffmpeg", "-hide_banner", "-n", "-activation_bytes", "abc123", "-i", str(source), "-map_metadata", "0", "-c", "copy", str(destination), ], ] assert destination.parent.is_dir() def test_run_command_redacts_activation_bytes_in_logs_and_errors(monkeypatch, caplog) -> None: def fake_run(arguments, *, check, capture_output, text): assert check is True assert capture_output is False assert text is True raise subprocess.CalledProcessError(1, arguments) monkeypatch.setattr(audible_convert.subprocess, "run", fake_run) caplog.set_level("DEBUG", audible_convert.__name__) with pytest.raises(audible_convert.CommandExecutionError) as error: audible_convert.run_command(["ffmpeg", "-activation_bytes", "secret-token", "-i", "book.aax"]) assert "secret-token" not in caplog.text assert "secret-token" not in str(error.value) assert "" in caplog.text assert "" in str(error.value) def test_write_agent_log_serializes_metadata_as_json_object(tmp_path) -> None: metadata = StandardBookMetadata( author_id=1, author="glynn_stewart", book_id=None, title="starship-mage", series_id=1, series="starships_mage", series_index=1, confidence=0.95, needs_review=False, evidence=["test"], ) log_file = tmp_path / "agent.jsonl" metadata_agent.write_agent_log(log_file, "final_metadata", metadata=metadata, path=tmp_path) record = json.loads(log_file.read_text(encoding="utf-8")) assert record["event"] == "final_metadata" assert record["metadata"]["author"] == "glynn_stewart" assert record["metadata"]["title"] == "starship-mage" assert record["path"] == str(tmp_path) def test_system_prompt_instructs_agent_to_detect_omnibuses() -> None: prompt = metadata_agent.system_prompt() assert "Detect omnibus or box-set editions" in prompt assert "books-1-3" in prompt assert "Keep series_index as the" in prompt assert "series_index must be a whole number or .5 value" in prompt assert "differ only by underscores" in prompt def test_standard_book_metadata_accepts_valid_tool_output(tmp_path, monkeypatch, audiobook_engine) -> None: install_fake_ollama( monkeypatch, [ tool_response("search_authors", {"query": "Glynn Stewart"}), tool_response("search_series", {"query": "starships_mage"}), final_response( { "author_id": 1, "book_id": None, "title": "starship-mage", "series_id": 1, "series_index": 1, "confidence": 0.95, "evidence": ["filename and catalog match"], }, ), ], ) metadata = standard_book_metadata( "Starship Mage.aax", {"title": "Starship Mage", "artist": "Glynn Stewart"}, audiobook_engine, tmp_path / "agent.jsonl", "test-key", config=metadata_agent.AgentConfig(), ) assert metadata == StandardBookMetadata( author_id=1, author="glynn_stewart", book_id=1, title="starship-mage", series_id=1, series="starships_mage", series_index=1, confidence=0.95, needs_review=False, evidence=["filename and catalog match"], ) records = [ json.loads(line) for line in (tmp_path / "agent.jsonl").read_text(encoding="utf-8").splitlines() ] sent = [record for record in records if record["event"] == "llm_messages_sent"] received = [record for record in records if record["event"] == "llm_message_received"] assert sent[0]["messages"][0]["role"] == "system" assert "Starship Mage" in sent[0]["messages"][1]["content"] assert received[0]["message"]["tool_calls"][0]["function"]["name"] == "search_authors" with Session(audiobook_engine) as session: book = session.get(Audiobook, 1) assert book.title == "starship-mage" assert book.author.name == "glynn_stewart" def test_standard_book_metadata_uses_agent_config(tmp_path, monkeypatch, audiobook_engine) -> None: config = metadata_agent.AgentConfig( model="custom-model", ollama_chat_url="https://ollama.example.test/api/chat", http_timeout_seconds=12, max_agent_turns=1, min_confidence=0.5, tool_names=("search_authors",), ) calls = install_fake_ollama( monkeypatch, [ tool_response("search_authors", {"query": "Glynn Stewart"}), final_response( { "author_id": 1, "book_id": None, "title": "standalone-book", "series_id": None, "series_index": 0, "confidence": 0.5, "evidence": ["custom config"], }, ), ], ) metadata = standard_book_metadata( "Standalone Book.aax", {"title": "Standalone Book", "artist": "Glynn Stewart"}, audiobook_engine, tmp_path / "agent.jsonl", "test-key", config=config, ) first_request_url = calls[0][0][0] first_request_options = calls[0][1] tool_names = [ tool_schema["function"]["name"] for tool_schema in first_request_options["json"]["tools"] ] assert first_request_url == "https://ollama.example.test/api/chat" assert first_request_options["timeout"] == 12 assert first_request_options["json"]["model"] == "custom-model" assert tool_names == ["search_authors"] assert metadata.needs_review is False assert metadata.series == "standalone" def test_standard_book_metadata_retries_invalid_json_then_needs_review( tmp_path, monkeypatch, audiobook_engine, ) -> None: install_fake_ollama( monkeypatch, [ tool_response("search_authors", {"query": "Glynn Stewart"}), tool_response("search_series", {"query": "Starship Mage"}), {"message": {"role": "assistant", "content": "{"}}, {"message": {"role": "assistant", "content": "{"}}, ], ) metadata = standard_book_metadata( "Starship Mage.aax", {"title": "Starship Mage"}, audiobook_engine, tmp_path / "agent.jsonl", "test-key", config=metadata_agent.AgentConfig(), ) assert metadata.needs_review is True assert metadata.confidence == 0 def test_standard_book_metadata_accepts_fenced_final_json( tmp_path, monkeypatch, audiobook_engine, ) -> None: install_fake_ollama( monkeypatch, [ tool_response("search_authors", {"query": "Dennis E. Taylor"}), tool_response("search_series", {"query": "Bobiverse", "author_id": 4}), tool_response("search_books", {"query": "All These Worlds", "author_id": 4, "series_id": 4}), fenced_final_response( { "author_id": 4, "book_id": None, "title": "all-these-worlds", "series_id": 4, "series_index": 3, "confidence": 0.95, "evidence": ["fenced json from model"], }, ), ], ) metadata = standard_book_metadata( "All These Worlds.aax", {"title": "All These Worlds: Bobiverse, Book 3", "artist": "Dennis E. Taylor"}, audiobook_engine, tmp_path / "agent.jsonl", "test-key", config=metadata_agent.AgentConfig(), ) assert metadata.needs_review is False assert metadata.author == "dennis_e_taylor" assert metadata.series == "bobiverse" assert metadata.title == "all-these-worlds" def test_standard_book_metadata_recovers_from_tool_validation_error( tmp_path, monkeypatch, audiobook_engine, ) -> None: install_fake_ollama( monkeypatch, [ tool_response("search_authors", {"query": "Cormac McCarthy"}), tool_response("ensure_author", {"name": "Cormac McCarthy"}), tool_response("ensure_series", {"name": "The Cormac McCarthy Collection", "author_id": 5}), tool_response( "ensure_book", { "title": "The Road", "author_id": 5, "series_id": 5, "series_index": 0, }, ), final_response( { "author_id": 5, "book_id": None, "title": "The Road", "series_id": None, "series_index": 0, "confidence": 0.9, "evidence": ["tool error showed this should be standalone"], }, ), ], ) log_file = tmp_path / "agent.jsonl" metadata = standard_book_metadata( "The Road.aax", {"title": "The Road", "artist": "Cormac McCarthy"}, audiobook_engine, log_file, "test-key", config=metadata_agent.AgentConfig(), ) assert metadata == StandardBookMetadata( author_id=5, author="cormac_mccarthy", book_id=1, title="the-road", series_id=None, series="standalone", series_index=0, confidence=0.9, needs_review=False, evidence=["tool error showed this should be standalone"], ) assert "series books must use a positive series_index" in log_file.read_text(encoding="utf-8") with Session(audiobook_engine) as session: assert session.get(AudiobookSeries, 5) is None book = session.get(Audiobook, 1) assert book.title == "the-road" assert book.series_id is None def test_standard_book_metadata_rejects_unknown_tool(tmp_path, monkeypatch, audiobook_engine) -> None: log_file = tmp_path / "agent.jsonl" install_fake_ollama(monkeypatch, [tool_response("drop_table", {})]) metadata = standard_book_metadata( "Book.aax", {"title": "Book"}, audiobook_engine, log_file, "test-key", config=metadata_agent.AgentConfig(), ) assert metadata.needs_review is True assert "Unknown audiobook metadata tool" in metadata.evidence[0] assert "tool_error" in log_file.read_text(encoding="utf-8") def test_standard_book_metadata_rejects_ids_not_returned_by_tools( tmp_path, monkeypatch, audiobook_engine, ) -> None: install_fake_ollama( monkeypatch, [ tool_response("search_authors", {"query": "Glynn Stewart"}), tool_response("search_series", {"query": "Starship Mage"}), final_response( { "author_id": 2, "book_id": None, "title": "expeditionary-force", "series_id": 1, "series_index": 1, "confidence": 0.99, "evidence": ["bad id"], }, ), final_response( { "author_id": 2, "book_id": None, "title": "expeditionary-force", "series_id": 1, "series_index": 1, "confidence": 0.99, "evidence": ["bad id"], }, ), ], ) metadata = standard_book_metadata( "Book.aax", {"title": "Book"}, audiobook_engine, tmp_path / "agent.jsonl", "test-key", config=metadata_agent.AgentConfig(), ) assert metadata.needs_review is True assert "author_id 2 was not returned" in metadata.evidence[0] def test_standard_book_metadata_rejects_series_for_wrong_author( tmp_path, monkeypatch, audiobook_engine, ) -> None: install_fake_ollama( monkeypatch, [ tool_response("search_authors", {"query": "Glynn Stewart"}), tool_response("search_series", {"query": "expeditionary_force"}), final_response( { "author_id": 1, "book_id": None, "title": "expeditionary-force", "series_id": 3, "series_index": 1, "confidence": 0.99, "evidence": ["wrong author"], }, ), final_response( { "author_id": 1, "book_id": None, "title": "expeditionary-force", "series_id": 3, "series_index": 1, "confidence": 0.99, "evidence": ["wrong author"], }, ), ], ) metadata = standard_book_metadata( "Book.aax", {"title": "Book"}, audiobook_engine, tmp_path / "agent.jsonl", "test-key", config=metadata_agent.AgentConfig(), ) assert metadata.needs_review is True assert "series_id 3 does not belong to author_id 1" in metadata.evidence[0] def test_standard_book_metadata_forces_final_after_empty_book_searches( tmp_path, monkeypatch, audiobook_engine, ) -> None: config = metadata_agent.AgentConfig(max_agent_turns=5) install_fake_ollama( monkeypatch, [ tool_response("search_authors", {"query": "Dennis E. Taylor"}), tool_response("search_series", {"query": "Bobiverse", "author_id": 4}), tool_response("search_books", {"query": "We Are Legion We Are Bob", "author_id": 4, "series_id": 4}), tool_response("search_books", {"query": "we are legion", "author_id": 4}), tool_response("search_books", {"query": "We Are Legion"}), final_response( { "author_id": 4, "book_id": None, "title": "we-are-legion-we-are-bob", "series_id": 4, "series_index": 1, "confidence": 0.95, "evidence": ["author and series tool results; title from ffprobe tags"], }, ), ], ) metadata = standard_book_metadata( "We_Are_Legion_(We_Are_Bob)_Bobiverse_Book_1-LC_128_44100_stereo.aax", { "album": "We Are Legion (We Are Bob): Bobiverse, Book 1", "artist": "Dennis E. Taylor", "title": "We Are Legion (We Are Bob): Bobiverse, Book 1", }, audiobook_engine, tmp_path / "agent.jsonl", "test-key", config=config, ) assert metadata == StandardBookMetadata( author_id=4, author="dennis_e_taylor", book_id=1, title="we-are-legion-we-are-bob", series_id=4, series="bobiverse", series_index=1, confidence=0.95, needs_review=False, evidence=["author and series tool results; title from ffprobe tags"], ) assert '"tools_enabled": false' in (tmp_path / "agent.jsonl").read_text(encoding="utf-8") def test_standard_book_metadata_can_create_missing_catalog_rows( tmp_path, monkeypatch, audiobook_engine, ) -> None: install_fake_ollama( monkeypatch, [ tool_response("search_authors", {"query": "Martha Wells"}), tool_response("ensure_author", {"name": "martha_wells"}), tool_response("search_series", {"query": "Murderbot Diaries", "author_id": 5}), tool_response("ensure_series", {"name": "murderbot_diaries", "author_id": 5}), tool_response("search_books", {"query": "All Systems Red", "author_id": 5, "series_id": 5}), final_response( { "author_id": 5, "book_id": None, "title": "all-systems-red", "series_id": 5, "series_index": 1, "confidence": 0.96, "evidence": ["created missing author and series; title from tags"], }, ), ], ) metadata = standard_book_metadata( "All Systems Red.aax", {"title": "All Systems Red", "artist": "Martha Wells"}, audiobook_engine, tmp_path / "agent.jsonl", "test-key", config=metadata_agent.AgentConfig(), ) assert metadata == StandardBookMetadata( author_id=5, author="martha_wells", book_id=1, title="all-systems-red", series_id=5, series="murderbot_diaries", series_index=1, confidence=0.96, needs_review=False, evidence=["created missing author and series; title from tags"], ) with Session(audiobook_engine) as session: author = session.get(AudiobookAuthor, 5) series = session.get(AudiobookSeries, 5) book = session.get(Audiobook, 1) assert author.name == "martha_wells" assert series.name == "murderbot_diaries" assert series.author_id == author.id assert book.title == "all-systems-red" assert book.author_id == author.id assert book.series_id == series.id def test_standard_book_metadata_accepts_half_series_index(tmp_path, monkeypatch, audiobook_engine) -> None: install_fake_ollama( monkeypatch, [ tool_response("search_series", {"query": "bobiverse", "author_id": 4}), final_response( { "author_id": 4, "book_id": None, "title": "bobiverse-short", "series_id": 4, "series_index": 1.5, "confidence": 0.95, "evidence": ["series novella from tags"], }, ), ], ) metadata = standard_book_metadata( "Bobiverse Short.aax", {"title": "Bobiverse Short", "artist": "Dennis E Taylor"}, audiobook_engine, tmp_path / "agent.jsonl", "test-key", config=metadata_agent.AgentConfig(), ) assert metadata.series_index == 1.5 assert metadata.needs_review is False with Session(audiobook_engine) as session: book = session.get(Audiobook, 1) assert book.series_index == 1.5 def test_standard_book_metadata_reuses_series_with_only_underscore_difference( tmp_path, monkeypatch, audiobook_engine, ) -> None: with Session(audiobook_engine) as session: session.add(AudiobookSeries(id=5, name="starships", author_id=1)) session.commit() install_fake_ollama( monkeypatch, [ tool_response("ensure_series", {"name": "starship_s", "author_id": 1}), final_response( { "author_id": 1, "book_id": None, "title": "starships-short", "series_id": 5, "series_index": 1, "confidence": 0.95, "evidence": ["reused existing series with equivalent slug"], }, ), ], ) metadata = standard_book_metadata( "Starship S Short.aax", {"title": "Starship S Short", "artist": "Glynn Stewart"}, audiobook_engine, tmp_path / "agent.jsonl", "test-key", config=metadata_agent.AgentConfig(), ) assert metadata.series == "starships" with Session(audiobook_engine) as session: series_names = session.scalars( select(AudiobookSeries.name).where(AudiobookSeries.author_id == 1).order_by(AudiobookSeries.name), ).all() assert "starship_s" not in series_names assert series_names == ["black_fleet_trilogy", "starships", "starships_mage"] def test_standard_book_metadata_normalizes_noisy_created_catalog_rows( tmp_path, monkeypatch, audiobook_engine, ) -> None: install_fake_ollama( monkeypatch, [ tool_response("search_authors", {"query": "Charles Lamb"}), tool_response("ensure_author", {"name": "charles-lamb"}), tool_response("search_series", {"query": "AL:ICE Series", "author_id": 5}), tool_response("ensure_series", {"name": "AL:ICE Series", "author_id": 5}), tool_response("search_books", {"query": "AL:ICE Space War", "author_id": 5, "series_id": 5}), final_response( { "author_id": 5, "book_id": None, "title": "AL:ICE Space War", "series_id": 5, "series_index": 4, "confidence": 0.95, "evidence": ["created normalized author and series; title from tags"], }, ), ], ) metadata = standard_book_metadata( "ALICE_Space_War_ALICE_Series_Book_4-LC_64_22050_stereo.aax", { "album": "AL:ICE Space War: AL:ICE Series, Book 4", "artist": "Charles Lamb", "title": "AL:ICE Space War: AL:ICE Series, Book 4", }, audiobook_engine, tmp_path / "agent.jsonl", "test-key", config=metadata_agent.AgentConfig(), ) assert metadata == StandardBookMetadata( author_id=5, author="charles_lamb", book_id=1, title="al-ice-space-war", series_id=5, series="al_ice_series", series_index=4, confidence=0.95, needs_review=False, evidence=["created normalized author and series; title from tags"], ) with Session(audiobook_engine) as session: author = session.get(AudiobookAuthor, 5) series = session.get(AudiobookSeries, 5) book = session.get(Audiobook, 1) assert author.name == "charles_lamb" assert series.name == "al_ice_series" assert series.author_id == author.id assert book.title == "al-ice-space-war" assert book.author_id == author.id assert book.series_id == series.id def test_convert_aax_file_with_agent_success_renames_temp_output(tmp_path, monkeypatch) -> None: source = tmp_path / "book.aax" output_directory = tmp_path / "audiobooks" source.touch() monkeypatch.setattr(audible_convert, "read_metadata", lambda _: {"title": "Starship Mage"}) monkeypatch.setattr( audible_convert, "standard_book_metadata", lambda *_, **__: StandardBookMetadata( author_id=1, author="glynn_stewart", book_id=None, title="starship-mage", series_id=1, series="starships_mage", series_index=1, confidence=0.95, needs_review=False, evidence=["test"], ), ) def fake_convert(_source, destination, _activation_bytes, *, overwrite): assert overwrite is True destination.parent.mkdir(parents=True, exist_ok=True) destination.write_text("converted", encoding="utf-8") monkeypatch.setattr(audible_convert, "convert_aax_file", fake_convert) audible_convert.convert_aax_file_with_agent( source, conversion_config(output_directory), ) expected = output_directory / "glynn_stewart-starships_mage_01-starship-mage" destination = expected / "glynn_stewart-starships_mage_01-starship-mage.m4b" assert destination.read_text(encoding="utf-8") == "converted" assert not list((output_directory / ".audible_convert" / "tmp").glob("*/converted.m4b")) def test_ffprobe_failure_writes_review_without_converting(tmp_path, monkeypatch) -> None: source = tmp_path / "book.aax" output_directory = tmp_path / "audiobooks" source.touch() calls = [] def fake_read_metadata(_source): raise FakeFfprobeError def fake_convert(*args, **kwargs): calls.append((args, kwargs)) monkeypatch.setattr(audible_convert, "read_metadata", fake_read_metadata) monkeypatch.setattr(audible_convert, "convert_aax_file", fake_convert) audible_convert.convert_aax_file_with_agent(source, conversion_config(output_directory)) review_files = list((output_directory / ".audible_convert" / "review").glob("*.json")) assert calls == [] assert len(review_files) == 1 review = json.loads(review_files[0].read_text(encoding="utf-8")) assert review["ffprobe_metadata"] == {} assert review["reason"] == "ffprobe_failed: bad ffprobe" assert review["temp_file"] is None def test_low_confidence_metadata_keeps_temp_output_for_review(tmp_path, monkeypatch) -> None: source = tmp_path / "book.aax" output_directory = tmp_path / "audiobooks" source.touch() monkeypatch.setattr(audible_convert, "read_metadata", lambda _: {"title": "Unknown"}) monkeypatch.setattr( audible_convert, "standard_book_metadata", lambda *_, **__: StandardBookMetadata( author_id=0, author="unknown_author", book_id=None, title="unknown-title", series_id=None, series="standalone", series_index=0, confidence=0.25, needs_review=True, evidence=["unclear"], ), ) def fake_convert(_source, destination, _activation_bytes, *, overwrite): assert overwrite is True destination.parent.mkdir(parents=True, exist_ok=True) destination.write_text("converted", encoding="utf-8") monkeypatch.setattr(audible_convert, "convert_aax_file", fake_convert) audible_convert.convert_aax_file_with_agent( source, conversion_config(output_directory), ) temp_files = list((output_directory / ".audible_convert" / "tmp").glob("*/converted.m4b")) review_files = list((output_directory / ".audible_convert" / "review").glob("*.json")) assert len(temp_files) == 1 assert temp_files[0].read_text(encoding="utf-8") == "converted" assert len(review_files) == 1 def test_existing_destination_skips_rename_and_removes_temp(tmp_path, monkeypatch) -> None: source = tmp_path / "book.aax" output_directory = tmp_path / "audiobooks" source.touch() final_file = ( output_directory / "glynn_stewart-starships_mage_01-starship-mage" / "glynn_stewart-starships_mage_01-starship-mage.m4b" ) final_file.parent.mkdir(parents=True) final_file.write_text("existing", encoding="utf-8") monkeypatch.setattr(audible_convert, "read_metadata", lambda _: {"title": "Starship Mage"}) monkeypatch.setattr( audible_convert, "standard_book_metadata", lambda *_, **__: StandardBookMetadata( author_id=1, author="glynn_stewart", book_id=None, title="starship-mage", series_id=1, series="starships_mage", series_index=1, confidence=0.95, needs_review=False, evidence=["test"], ), ) def fake_convert(_source, destination, _activation_bytes, *, overwrite): assert overwrite is True destination.parent.mkdir(parents=True, exist_ok=True) destination.write_text("converted", encoding="utf-8") monkeypatch.setattr(audible_convert, "convert_aax_file", fake_convert) audible_convert.convert_aax_file_with_agent( source, conversion_config(output_directory), ) assert final_file.read_text(encoding="utf-8") == "existing" assert not list((output_directory / ".audible_convert" / "tmp").glob("*/converted.m4b")) def test_richie_exports_audiobook_models() -> None: from python.orm.richie import Audiobook # noqa: PLC0415 assert Audiobook.__tablename__ == "audiobook" def test_audiobook_title_author_series_is_unique(audiobook_engine) -> None: with Session(audiobook_engine) as session: session.add_all( [ Audiobook(title="duplicate-title", author_id=1, series_id=1, series_index=1), Audiobook(title="duplicate-title", author_id=1, series_id=1, series_index=2), ], ) with pytest.raises(IntegrityError): session.commit() def test_main_dry_run_prints_outputs_without_converting(tmp_path, monkeypatch, capsys) -> None: input_directory = tmp_path / "raw" output_directory = tmp_path / "audiobooks" input_directory.mkdir() source = input_directory / "book.aax" source.touch() monkeypatch.setenv("OLLAMA_API_KEY", "test-key") monkeypatch.setattr( audible_convert, "read_metadata", lambda _: { "artist": "Charles Lamb", "title": "Alice: Alice Series #1", }, ) calls = [] def fake_convert(*args, **kwargs): calls.append((args, kwargs)) monkeypatch.setattr(audible_convert, "convert_aax_file", fake_convert) monkeypatch.setattr( audible_convert, "standard_book_metadata", lambda *_, **__: StandardBookMetadata( author_id=1, author="charles_lamb", book_id=None, title="alice", series_id=1, series="alice", series_index=1, confidence=0.95, needs_review=False, evidence=["test"], ), ) def fake_get_postgres_engine(*, name): assert name == "RICHIE" return create_engine("sqlite+pysqlite:///:memory:") monkeypatch.setattr(audible_convert, "get_postgres_engine", fake_get_postgres_engine) audible_convert.main(input_directory, output_directory, dry_run=True) assert calls == [] assert capsys.readouterr().out == ( f"{source} -> {output_directory / 'charles_lamb-alice_01-alice' / 'charles_lamb-alice_01-alice.m4b'}\n" ) dry_run_file = ( output_directory / ".audible_convert" / "dry-run" / "charles_lamb-alice_01-alice" / "charles_lamb-alice_01-alice.m4b" ) assert dry_run_file.read_text(encoding="utf-8") == ( f"{output_directory / 'charles_lamb-alice_01-alice' / 'charles_lamb-alice_01-alice.m4b'}\n" ) assert (output_directory / ".audible_convert" / "logs").is_dir() def test_main_reads_activation_bytes_from_env(tmp_path, monkeypatch) -> None: input_directory = tmp_path / "raw" output_directory = tmp_path / "audiobooks" input_directory.mkdir() source = input_directory / "book.aax" source.touch() configs = [] def fake_convert(_source, config): configs.append(config) def fake_get_postgres_engine(*, name): assert name == "RICHIE" return sqlite_engine() monkeypatch.setenv("OLLAMA_API_KEY", "test-key") monkeypatch.setenv("AUDIBLE_ACTIVATION_BYTES", "activation-secret") monkeypatch.setattr(audible_convert, "get_postgres_engine", fake_get_postgres_engine) monkeypatch.setattr(audible_convert, "convert_aax_file_with_agent", fake_convert) audible_convert.main(input_directory, output_directory) assert configs == [ audible_convert.ConversionConfig( resolved_output=output_directory, ollama_api_key="test-key", agent_config=configs[0].agent_config, engine=configs[0].engine, activation_bytes="activation-secret", dry_run=False, overwrite=False, ), ]