Richie
b8a21a6013
treefmt / nix fmt (pull_request) Successful in 6s
pytest / pytest (pull_request) Failing after 27s
build_systems / build-bob (pull_request) Successful in 47s
build_systems / build-brain (pull_request) Successful in 47s
build_systems / build-leviathan (pull_request) Successful in 52s
build_systems / build-rhapsody-in-green (pull_request) Successful in 58s
build_systems / build-jeeves (pull_request) Successful in 2m28s
127 lines
4.0 KiB
Python
127 lines
4.0 KiB
Python
"""test_audiobook_catalog."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, select
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from python.orm.richie import AudiobookAuthor, AudiobookSeries, RichieBase
|
|
from python.tools.audiobook import catalog
|
|
|
|
|
|
@pytest.fixture
|
|
def audiobook_session():
|
|
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:
|
|
yield session
|
|
engine.dispose()
|
|
|
|
|
|
def test_upsert_catalog_csv_inserts_and_updates_authors_and_series(tmp_path, audiobook_session) -> None:
|
|
audiobook_session.add_all(
|
|
[
|
|
AudiobookAuthor(id=10, name="old_author"),
|
|
AudiobookAuthor(id=11, name="craig_alanson"),
|
|
],
|
|
)
|
|
audiobook_session.commit()
|
|
authors_csv = tmp_path / "authors.csv"
|
|
series_csv = tmp_path / "series.csv"
|
|
authors_csv.write_text(
|
|
"name,id\n"
|
|
"glynn_stewart,\n"
|
|
"craig_alanson,\n"
|
|
"updated_author,10\n",
|
|
encoding="utf-8",
|
|
)
|
|
series_csv.write_text(
|
|
"name,author_name,id\n"
|
|
"starships_mage,glynn_stewart,\n"
|
|
"expeditionary_force,craig_alanson,\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
author_count = catalog.upsert_authors_from_csv(audiobook_session, authors_csv)
|
|
series_count = catalog.upsert_series_from_csv(audiobook_session, series_csv)
|
|
audiobook_session.commit()
|
|
|
|
authors = audiobook_session.scalars(select(AudiobookAuthor).order_by(AudiobookAuthor.id)).all()
|
|
series = audiobook_session.scalars(select(AudiobookSeries).order_by(AudiobookSeries.name)).all()
|
|
assert author_count == 3
|
|
assert series_count == 2
|
|
assert [(author.id, author.name) for author in authors] == [
|
|
(10, "updated_author"),
|
|
(11, "craig_alanson"),
|
|
(12, "glynn_stewart"),
|
|
]
|
|
assert [(row.name, row.author.name) for row in series] == [
|
|
("expeditionary_force", "craig_alanson"),
|
|
("starships_mage", "glynn_stewart"),
|
|
]
|
|
|
|
|
|
def test_upsert_series_csv_updates_series_by_id(tmp_path, audiobook_session) -> None:
|
|
author = AudiobookAuthor(id=1, name="glynn_stewart")
|
|
audiobook_session.add_all(
|
|
[
|
|
author,
|
|
AudiobookSeries(id=7, name="old_series", author=author),
|
|
],
|
|
)
|
|
audiobook_session.commit()
|
|
series_csv = tmp_path / "series.csv"
|
|
series_csv.write_text(
|
|
"name,author_name,id\n"
|
|
"starships_mage,glynn_stewart,7\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
count = catalog.upsert_series_from_csv(audiobook_session, series_csv)
|
|
audiobook_session.commit()
|
|
|
|
series = audiobook_session.get(AudiobookSeries, 7)
|
|
assert count == 1
|
|
assert series.name == "starships_mage"
|
|
assert series.author.name == "glynn_stewart"
|
|
|
|
|
|
def test_upsert_csv_allows_missing_id_column(tmp_path, audiobook_session) -> None:
|
|
authors_csv = tmp_path / "authors.csv"
|
|
series_csv = tmp_path / "series.csv"
|
|
authors_csv.write_text(
|
|
"name\n"
|
|
"glynn_stewart\n",
|
|
encoding="utf-8",
|
|
)
|
|
series_csv.write_text(
|
|
"name,author_name\n"
|
|
"starships_mage,glynn_stewart\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
author_count = catalog.upsert_authors_from_csv(audiobook_session, authors_csv)
|
|
series_count = catalog.upsert_series_from_csv(audiobook_session, series_csv)
|
|
audiobook_session.commit()
|
|
|
|
series = audiobook_session.scalar(select(AudiobookSeries))
|
|
assert author_count == 1
|
|
assert series_count == 1
|
|
assert series.name == "starships_mage"
|
|
assert series.author.name == "glynn_stewart"
|
|
|
|
|
|
def test_upsert_series_csv_rejects_unknown_author(tmp_path, audiobook_session) -> None:
|
|
series_csv = tmp_path / "series.csv"
|
|
series_csv.write_text(
|
|
"name,author_name,id\n"
|
|
"starships_mage,glynn_stewart,\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with pytest.raises(catalog.CatalogImportError) as error:
|
|
catalog.upsert_series_from_csv(audiobook_session, series_csv)
|
|
|
|
assert "author not found: glynn_stewart" in str(error.value)
|