patrick
bd8d68b9ba
file.path is relative to book.path, so the archive matched nothing on disk and came out empty. Also namespace entries per book to stop filename collisions, and stream the zip instead of buffering it whole.
205 lines
7.5 KiB
Python
205 lines
7.5 KiB
Python
"""Tests for BookService"""
|
|
|
|
import zipfile
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import aiofiles.os as aios
|
|
|
|
from chitai.schemas import BookCreate
|
|
from chitai.services import BookService
|
|
from chitai.database import models as m
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestBookServiceCRUD:
|
|
"""Test CRUD operation for libraries."""
|
|
|
|
async def test_update_book(
|
|
self, books_service: BookService, test_library: m.Library
|
|
) -> None:
|
|
book_data = BookCreate(
|
|
library_id=1,
|
|
title="Fellowship of the Ring",
|
|
authors=["J.R.R Tolkien"],
|
|
tags=["Fantasy"],
|
|
identifiers={"isbn-13": "9780261102354"},
|
|
pages=427,
|
|
)
|
|
|
|
book = await books_service.to_model_on_create(book_data.model_dump())
|
|
assert isinstance(book, m.Book)
|
|
|
|
# Add path manually as it won't be generated (not using the create function, but manually inserting into db)
|
|
book.path = f"{test_library.root_path}/J.R.R Tolkien/The Fellowship of the Ring"
|
|
await aios.makedirs(book.path) # type: ignore[arg-type]
|
|
|
|
books_service.repository.session.add(book)
|
|
await books_service.repository.session.commit()
|
|
await books_service.repository.session.refresh(book)
|
|
|
|
await books_service.update_book(
|
|
book.id,
|
|
{
|
|
"title": "The Fellowship of the Ring",
|
|
"identifiers": {"isbn-10": "9780261102354"},
|
|
"edition": 3,
|
|
"publisher": "Tolkien Estate",
|
|
"series": "The Lord of the Rings",
|
|
"series_position": "1",
|
|
"tags": ["Fantasy", "Adventure"],
|
|
},
|
|
test_library,
|
|
)
|
|
|
|
updated_book = await books_service.get(book.id)
|
|
|
|
# Assert updated information is correct
|
|
|
|
assert updated_book.title == "The Fellowship of the Ring"
|
|
assert (
|
|
updated_book.path
|
|
== f"{test_library.root_path}/J.R.R Tolkien/The Lord of the Rings/01 - The Fellowship of the Ring"
|
|
)
|
|
|
|
assert len(updated_book.identifiers)
|
|
assert updated_book.identifiers[0].value == "9780261102354"
|
|
|
|
assert updated_book.edition == 3
|
|
assert updated_book.publisher is not None
|
|
assert updated_book.publisher.name == "Tolkien Estate"
|
|
|
|
assert len(updated_book.tags) == 2
|
|
|
|
async def test_update_book_reuses_existing_links(
|
|
self, books_service: BookService, test_library: m.Library
|
|
) -> None:
|
|
"""Re-submitting a relationship a book already has must not duplicate its link.
|
|
|
|
The link tables are unique on (book_id, tag_id) / (book_id, author_id), and
|
|
identifiers on (name, book_id), so replacing a collection wholesale used to
|
|
insert a row that collided with the one it was replacing.
|
|
"""
|
|
book_data = BookCreate(
|
|
library_id=1,
|
|
title="The Two Towers",
|
|
authors=["J.R.R Tolkien"],
|
|
tags=["Fantasy"],
|
|
identifiers={"isbn-13": "9780261102358"},
|
|
pages=352,
|
|
)
|
|
|
|
book = await books_service.to_model_on_create(book_data.model_dump())
|
|
assert isinstance(book, m.Book)
|
|
|
|
# Matches what the default template generates, so these updates move nothing.
|
|
book.path = f"{test_library.root_path}/J.R.R Tolkien/The Two Towers"
|
|
await aios.makedirs(book.path) # type: ignore[arg-type]
|
|
|
|
books_service.repository.session.add(book)
|
|
await books_service.repository.session.commit()
|
|
await books_service.repository.session.refresh(book)
|
|
|
|
# Through the service, so the link collections come back eagerly loaded.
|
|
created_book = await books_service.get(book.id)
|
|
original_tag_link_id = created_book.tag_links[0].id
|
|
|
|
# Every collection resubmitted unchanged.
|
|
await books_service.update_book(
|
|
book.id,
|
|
{
|
|
"authors": ["J.R.R Tolkien"],
|
|
"tags": ["Fantasy"],
|
|
"identifiers": {"isbn-13": "9780261102358"},
|
|
},
|
|
test_library,
|
|
)
|
|
|
|
updated_book = await books_service.get(book.id)
|
|
assert [tag.name for tag in updated_book.tags] == ["Fantasy"]
|
|
assert [author.name for author in updated_book.authors] == ["J.R.R Tolkien"]
|
|
assert len(updated_book.identifiers) == 1
|
|
# The existing link is reused, not deleted and reinserted.
|
|
assert updated_book.tag_links[0].id == original_tag_link_id
|
|
|
|
# Keeping one tag while adding another.
|
|
await books_service.update_book(
|
|
book.id, {"tags": ["Fantasy", "Adventure"]}, test_library
|
|
)
|
|
updated_book = await books_service.get(book.id)
|
|
assert [tag.name for tag in updated_book.tags] == ["Fantasy", "Adventure"]
|
|
|
|
# Dropping one while keeping the other.
|
|
await books_service.update_book(book.id, {"tags": ["Adventure"]}, test_library)
|
|
updated_book = await books_service.get(book.id)
|
|
assert [tag.name for tag in updated_book.tags] == ["Adventure"]
|
|
|
|
# A name the book already carries has its value updated in place.
|
|
await books_service.update_book(
|
|
book.id, {"identifiers": {"isbn-13": "9780261102999"}}, test_library
|
|
)
|
|
updated_book = await books_service.get(book.id)
|
|
assert len(updated_book.identifiers) == 1
|
|
assert updated_book.identifiers[0].value == "9780261102999"
|
|
|
|
async def test_get_files_zips_every_book_file(
|
|
self, books_service: BookService, test_library: m.Library
|
|
) -> None:
|
|
"""The multi-book download must contain the real bytes of each file.
|
|
|
|
`file.path` holds a bare filename relative to `book.path`, so resolving it on
|
|
its own matched nothing on disk and yielded a valid but empty archive.
|
|
"""
|
|
contents = {
|
|
"Dracula.epub": b"epub-payload-" * 512,
|
|
"Dracula.pdf": b"pdf-payload-" * 512,
|
|
}
|
|
|
|
book_data = BookCreate(
|
|
library_id=test_library.id,
|
|
title="Dracula",
|
|
authors=["Bram Stoker"],
|
|
pages=418,
|
|
)
|
|
book = await books_service.to_model_on_create(book_data.model_dump())
|
|
assert isinstance(book, m.Book)
|
|
|
|
book.path = f"{test_library.root_path}/Bram Stoker/Dracula"
|
|
await aios.makedirs(book.path) # type: ignore[arg-type]
|
|
|
|
for name, payload in contents.items():
|
|
Path(book.path, name).write_bytes(payload)
|
|
book.files.append(
|
|
m.FileMetadata(
|
|
path=name,
|
|
size=len(payload),
|
|
hash=f"hash-{name}",
|
|
content_type=None,
|
|
)
|
|
)
|
|
|
|
books_service.repository.session.add(book)
|
|
await books_service.repository.session.commit()
|
|
|
|
archive_bytes = b"".join(
|
|
[
|
|
chunk
|
|
async for chunk in books_service.get_files([book.id], test_library.id)
|
|
]
|
|
)
|
|
|
|
with zipfile.ZipFile(BytesIO(archive_bytes)) as archive:
|
|
assert archive.testzip() is None
|
|
|
|
names = archive.namelist()
|
|
assert len(names) == len(contents)
|
|
|
|
# Entries are namespaced by the book's directory, so two books sharing a
|
|
# filename cannot overwrite one another.
|
|
assert {Path(name).parent.name for name in names} == {"Dracula"}
|
|
|
|
for name, payload in contents.items():
|
|
entry = next(n for n in names if Path(n).name == name)
|
|
assert archive.read(entry) == payload
|