patrick
55e00ba960
Reads metadata.db and copies the books into a library — from a zip uploaded on the library settings page, or from a path with `litestar calibre-import`. The source is never touched, and re-running only picks up what is new. Also names the formats mimetypes does not know: a Calibre library is full of MOBI and AZW3, and a null content type used to fail the book endpoint.
393 lines
13 KiB
Python
393 lines
13 KiB
Python
import zipfile
|
|
from datetime import date
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from chitai.services.calibre import (
|
|
CalibreLibrary,
|
|
CalibreLibraryError,
|
|
extract_calibre_archive,
|
|
format_series_index,
|
|
parse_date,
|
|
strip_html,
|
|
unescape_author,
|
|
)
|
|
|
|
from tests.calibre_fixtures import UNDEFINED_DATE, CalibreFixture
|
|
|
|
|
|
EPUB = Path("tests/data_files/Metamorphosis - Franz Kafka.epub")
|
|
PDF = Path("tests/data_files/Calculus Made Easy - Silvanus Thompson.pdf")
|
|
|
|
|
|
@pytest.fixture(name="library_root")
|
|
def fx_library_root(tmp_path: Path) -> Path:
|
|
"""A small Calibre library covering the rows that are easy to read wrongly."""
|
|
fixture = CalibreFixture(tmp_path / "Calibre Library")
|
|
fixture.add_pages_table()
|
|
|
|
fixture.add_book(
|
|
1,
|
|
"The Metamorphosis",
|
|
authors=["Franz Kafka"],
|
|
pubdate="1915-10-15 00:00:00+00:00",
|
|
tags=["Fiction", "Absurdist"],
|
|
publisher="Kurt Wolff Verlag",
|
|
languages=["deu", "eng"],
|
|
comment="<p>A travelling salesman.</p><p>He wakes up <i>changed</i>.</p>",
|
|
identifiers={"isbn": "978-0-486-29030-0", "amazon": "B01N5IB20Q"},
|
|
uuid="11111111-2222-3333-4444-555555555555",
|
|
pages=201,
|
|
cover=True,
|
|
formats={"EPUB": EPUB},
|
|
)
|
|
|
|
# Volume seven of a series, and no publication date — the two values most likely to
|
|
# be carried through verbatim when they should not be.
|
|
fixture.add_book(
|
|
2,
|
|
"Persepolis Rising",
|
|
# Calibre escapes the comma and nothing else, so the space after it is stored
|
|
# as-is: `Corey, Jr.` is written `Corey| Jr.`.
|
|
authors=["Corey| Jr., James S. A."],
|
|
series="The Expanse",
|
|
series_index=7.0,
|
|
formats={"EPUB": EPUB},
|
|
)
|
|
|
|
# A novella between two novels: a fractional position is real and must survive.
|
|
fixture.add_book(
|
|
3, "Strange Dogs", series="The Expanse", series_index=6.5, formats={"PDF": PDF}
|
|
)
|
|
|
|
# Every row Calibre will happily hold and Chitai cannot use: no files at all.
|
|
fixture.add_book(4, "Metadata Only")
|
|
|
|
# A catalogue row whose file is not on disk.
|
|
fixture.add_book(5, "Lost Book")
|
|
fixture.add_missing_format(5, "EPUB", "Lost Book - Unknown")
|
|
|
|
return fixture.commit()
|
|
|
|
|
|
async def test_reads_a_book_whole(library_root: Path) -> None:
|
|
async with CalibreLibrary(library_root) as library:
|
|
assert await library.count() == 5
|
|
books = await library.books()
|
|
|
|
book = books[0]
|
|
|
|
assert book.calibre_id == 1
|
|
assert book.title == "The Metamorphosis"
|
|
assert book.authors == ["Franz Kafka"]
|
|
assert book.published_date == date(1915, 10, 15)
|
|
assert book.tags == ["Absurdist", "Fiction"]
|
|
assert book.publisher == "Kurt Wolff Verlag"
|
|
assert book.pages == 201
|
|
assert book.uuid == "11111111-2222-3333-4444-555555555555"
|
|
|
|
# One language, and the one Calibre put first.
|
|
assert book.language == "deu"
|
|
|
|
# Reported as Calibre wrote them: folding `amazon` onto `asin` is the importer's
|
|
# job, not the reader's.
|
|
assert book.identifiers == {"isbn": "978-0-486-29030-0", "amazon": "B01N5IB20Q"}
|
|
|
|
assert book.cover is not None
|
|
assert book.cover.is_file()
|
|
|
|
assert len(book.files) == 1
|
|
assert book.files[0].format == "EPUB"
|
|
assert book.files[0].path.is_file()
|
|
# The stem is Calibre's, truncated and sanitised — never the title.
|
|
assert book.files[0].path.name != f"{book.title}.epub"
|
|
|
|
|
|
async def test_the_undefined_date_is_not_a_date(library_root: Path) -> None:
|
|
"""`0101-01-01` parses fine, which is exactly the problem."""
|
|
async with CalibreLibrary(library_root) as library:
|
|
books = {book.calibre_id: book for book in await library.books()}
|
|
|
|
assert books[2].published_date is None
|
|
|
|
|
|
async def test_series_position_is_a_plain_string(library_root: Path) -> None:
|
|
async with CalibreLibrary(library_root) as library:
|
|
books = {book.calibre_id: book for book in await library.books()}
|
|
|
|
assert books[2].series == "The Expanse"
|
|
assert books[2].series_position == "7"
|
|
|
|
assert books[3].series_position == "6.5"
|
|
|
|
# `series_index` defaults to 1.0 for every book, so a position without a series
|
|
# would invent a volume one out of nothing.
|
|
assert books[1].series is None
|
|
assert books[1].series_position is None
|
|
|
|
|
|
async def test_author_commas_are_unescaped(library_root: Path) -> None:
|
|
async with CalibreLibrary(library_root) as library:
|
|
books = {book.calibre_id: book for book in await library.books()}
|
|
|
|
assert books[2].authors == ["Corey, Jr., James S. A."]
|
|
|
|
|
|
async def test_comments_come_back_as_text(library_root: Path) -> None:
|
|
async with CalibreLibrary(library_root) as library:
|
|
books = {book.calibre_id: book for book in await library.books()}
|
|
|
|
assert books[1].description == "A travelling salesman.\nHe wakes up changed."
|
|
assert books[2].description is None
|
|
|
|
|
|
async def test_files_are_reported_whether_or_not_they_exist(library_root: Path) -> None:
|
|
"""
|
|
The reader says what the catalogue says. Whether the bytes are there is a question
|
|
for whoever is about to copy them, which stats them anyway.
|
|
"""
|
|
async with CalibreLibrary(library_root) as library:
|
|
books = {book.calibre_id: book for book in await library.books()}
|
|
|
|
assert books[4].files == []
|
|
|
|
assert len(books[5].files) == 1
|
|
assert not books[5].files[0].path.exists()
|
|
|
|
|
|
async def test_a_library_without_the_pages_table_still_reads(tmp_path: Path) -> None:
|
|
"""`books_pages_link` is recent; an older library simply does not have it."""
|
|
fixture = CalibreFixture(tmp_path / "Old Library")
|
|
fixture.add_book(1, "Old Book", formats={"EPUB": EPUB})
|
|
root = fixture.commit()
|
|
|
|
async with CalibreLibrary(root) as library:
|
|
books = await library.books()
|
|
|
|
assert books[0].pages is None
|
|
|
|
|
|
async def test_the_original_is_never_opened(library_root: Path) -> None:
|
|
"""
|
|
The catalogue is copied before it is read, and the copy goes away afterwards.
|
|
|
|
Calibre may be running and writing; this is what keeps a live library out of it.
|
|
"""
|
|
before = (library_root / "metadata.db").read_bytes()
|
|
|
|
library = CalibreLibrary(library_root)
|
|
await library.open()
|
|
workspace = library._workspace
|
|
|
|
assert workspace is not None and (workspace / "metadata.db").is_file()
|
|
|
|
await library.close()
|
|
|
|
assert not workspace.exists()
|
|
assert (library_root / "metadata.db").read_bytes() == before
|
|
|
|
|
|
async def test_closing_twice_is_harmless(library_root: Path) -> None:
|
|
library = CalibreLibrary(library_root)
|
|
await library.open()
|
|
await library.close()
|
|
await library.close()
|
|
|
|
|
|
async def test_a_directory_that_is_not_a_calibre_library(tmp_path: Path) -> None:
|
|
with pytest.raises(CalibreLibraryError, match="not a Calibre library"):
|
|
await CalibreLibrary(tmp_path).open()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("stored", "expected"),
|
|
[
|
|
("2017-12-04 04:00:00+00:00", date(2017, 12, 4)),
|
|
("2001-07-02 00:00:00+00:00", date(2001, 7, 2)),
|
|
("1999-01-31", date(1999, 1, 31)),
|
|
# Calibre's sentinel, and anything else implausibly early.
|
|
(UNDEFINED_DATE, None),
|
|
("0101-01-01", None),
|
|
(None, None),
|
|
("", None),
|
|
("not a date", None),
|
|
],
|
|
)
|
|
def test_parse_date(stored: str | None, expected: date | None) -> None:
|
|
assert parse_date(stored) == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("index", "expected"),
|
|
[
|
|
(7.0, "7"),
|
|
(1.0, "1"),
|
|
(6.5, "6.5"),
|
|
(0.0, "0"),
|
|
(12.25, "12.25"),
|
|
(None, None),
|
|
],
|
|
)
|
|
def test_format_series_index(index: float | None, expected: str | None) -> None:
|
|
assert format_series_index(index) == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("stored", "expected"),
|
|
[
|
|
("Doyle| Sir Arthur Conan", "Doyle, Sir Arthur Conan"),
|
|
("Franz Kafka", "Franz Kafka"),
|
|
(" Herman Melville ", "Herman Melville"),
|
|
],
|
|
)
|
|
def test_unescape_author(stored: str, expected: str) -> None:
|
|
assert unescape_author(stored) == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("html", "expected"),
|
|
[
|
|
("<p>One.</p><p>Two.</p>", "One.\nTwo."),
|
|
("Plain text", "Plain text"),
|
|
("<div>A<br>B</div>", "A\nB"),
|
|
("<p>Café & bar</p>", "Café & bar"),
|
|
("<ul><li>One</li><li>Two</li></ul>", "One\nTwo"),
|
|
# Markup carrying no text at all is nothing, not an empty description.
|
|
("<p></p>", None),
|
|
("", None),
|
|
(None, None),
|
|
],
|
|
)
|
|
def test_strip_html(html: str | None, expected: str | None) -> None:
|
|
assert strip_html(html) == expected
|
|
|
|
|
|
class TestArchives:
|
|
"""A Calibre library that arrives zipped rather than as a path."""
|
|
|
|
def zipped(self, root: Path, into: Path, prefix: str = "") -> Path:
|
|
"""Zip a directory the way a file manager would."""
|
|
archive = into / "library.zip"
|
|
|
|
with zipfile.ZipFile(archive, "w") as writing:
|
|
for path in sorted(root.rglob("*")):
|
|
if path.is_file():
|
|
writing.write(path, f"{prefix}{path.relative_to(root)}")
|
|
|
|
return archive
|
|
|
|
async def test_a_library_zipped_at_its_root(
|
|
self, library_root: Path, tmp_path: Path
|
|
) -> None:
|
|
archive = self.zipped(library_root, tmp_path)
|
|
destination = tmp_path / "unpacked"
|
|
destination.mkdir()
|
|
|
|
catalogue = await extract_calibre_archive(archive, destination)
|
|
|
|
assert catalogue == destination
|
|
async with CalibreLibrary(catalogue) as library:
|
|
assert await library.count() == 5
|
|
|
|
async def test_a_library_zipped_inside_a_folder(
|
|
self, library_root: Path, tmp_path: Path
|
|
) -> None:
|
|
"""Zipping the folder itself is at least as common as zipping its contents."""
|
|
archive = self.zipped(library_root, tmp_path, prefix="Calibre Library/")
|
|
destination = tmp_path / "unpacked"
|
|
destination.mkdir()
|
|
|
|
catalogue = await extract_calibre_archive(archive, destination)
|
|
|
|
assert catalogue == destination / "Calibre Library"
|
|
async with CalibreLibrary(catalogue) as library:
|
|
assert await library.count() == 5
|
|
|
|
async def test_an_entry_pointing_outside_the_archive_is_refused(
|
|
self, tmp_path: Path
|
|
) -> None:
|
|
"""
|
|
Zip slip. `ZipFile.extract` sanitises names itself, but relying on that silently
|
|
is how the next person to change the extraction call reintroduces it.
|
|
"""
|
|
archive = tmp_path / "hostile.zip"
|
|
|
|
with zipfile.ZipFile(archive, "w") as writing:
|
|
writing.writestr("metadata.db", "not really")
|
|
writing.writestr("../../escaped.txt", "gotcha")
|
|
|
|
destination = tmp_path / "unpacked"
|
|
destination.mkdir()
|
|
|
|
with pytest.raises(CalibreLibraryError, match="outside itself"):
|
|
await extract_calibre_archive(archive, destination)
|
|
|
|
assert not (tmp_path.parent / "escaped.txt").exists()
|
|
|
|
async def test_something_that_is_not_a_zip(self, tmp_path: Path) -> None:
|
|
archive = tmp_path / "not.zip"
|
|
archive.write_bytes(b"PK-ish, but no")
|
|
|
|
destination = tmp_path / "unpacked"
|
|
destination.mkdir()
|
|
|
|
with pytest.raises(CalibreLibraryError, match="not a zip file"):
|
|
await extract_calibre_archive(archive, destination)
|
|
|
|
async def test_an_archive_with_no_catalogue(self, tmp_path: Path) -> None:
|
|
archive = tmp_path / "books.zip"
|
|
|
|
with zipfile.ZipFile(archive, "w") as writing:
|
|
writing.writestr("Some Book.epub", "content")
|
|
|
|
destination = tmp_path / "unpacked"
|
|
destination.mkdir()
|
|
|
|
with pytest.raises(CalibreLibraryError, match="no metadata.db"):
|
|
await extract_calibre_archive(archive, destination)
|
|
|
|
# Refused before anything was written.
|
|
assert list(destination.iterdir()) == []
|
|
|
|
async def test_a_catalogue_buried_too_deep(self, tmp_path: Path) -> None:
|
|
"""Somebody's whole backup tree is not a library, however much it contains one."""
|
|
archive = tmp_path / "backup.zip"
|
|
|
|
with zipfile.ZipFile(archive, "w") as writing:
|
|
writing.writestr("backups/2026/january/library/metadata.db", "not really")
|
|
|
|
destination = tmp_path / "unpacked"
|
|
destination.mkdir()
|
|
|
|
with pytest.raises(CalibreLibraryError, match="within 3 levels"):
|
|
await extract_calibre_archive(archive, destination)
|
|
|
|
async def test_an_archive_too_big_for_the_disk(
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""
|
|
Checked before writing, not discovered part-way through.
|
|
|
|
A full disk takes the whole application down, and the size is in the archive
|
|
already.
|
|
"""
|
|
archive = tmp_path / "huge.zip"
|
|
|
|
with zipfile.ZipFile(archive, "w") as writing:
|
|
writing.writestr("metadata.db", "not really")
|
|
|
|
destination = tmp_path / "unpacked"
|
|
destination.mkdir()
|
|
|
|
monkeypatch.setattr(
|
|
"chitai.services.calibre.shutil.disk_usage",
|
|
lambda _path: SimpleNamespace(total=1024, used=1024, free=0),
|
|
)
|
|
|
|
with pytest.raises(CalibreLibraryError, match="only"):
|
|
await extract_calibre_archive(archive, destination)
|
|
|
|
assert list(destination.iterdir()) == []
|