patrick
d78b21c27f
Incoming files are matched against what is already stored, keyed on the KOReader hash and the file size. Bulk uploads skip and report them, deliberate creates are refused with a 409, the consume directory parks them aside, and allow_duplicates overrides all three. Also: books whose metadata generates a path another book already owns are moved aside, so a forced copy cannot overwrite the original's files.
1099 lines
36 KiB
Python
1099 lines
36 KiB
Python
import pytest
|
|
from httpx import AsyncClient
|
|
from pathlib import Path
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("path_to_book", "library_id", "title", "authors"),
|
|
[
|
|
(
|
|
Path("tests/data_files/Metamorphosis - Franz Kafka.epub"),
|
|
1,
|
|
"Metamorphosis",
|
|
["Franz Kafka"],
|
|
),
|
|
(
|
|
Path("tests/data_files/Moby Dick; Or, The Whale - Herman Melville.epub"),
|
|
1,
|
|
"Moby Dick; Or, The Whale",
|
|
["Herman Melville"],
|
|
),
|
|
(
|
|
Path(
|
|
"tests/data_files/Relativity: The Special and General Theory - Albert Einstein.epub"
|
|
),
|
|
2,
|
|
"Relativity : the Special and General Theory",
|
|
["Albert Einstein"],
|
|
),
|
|
(
|
|
Path(
|
|
"tests/data_files/On The Origin of Species By Means of Natural Selection - Charles Darwin.epub"
|
|
),
|
|
2,
|
|
"On the Origin of Species By Means of Natural Selection / Or, the Preservation of Favoured Races in the Struggle for Life",
|
|
["Charles Darwin"],
|
|
),
|
|
(
|
|
Path("tests/data_files/Calculus Made Easy - Silvanus Thompson.pdf"),
|
|
2,
|
|
"The Project Gutenberg eBook #33283: Calculus Made Easy, 2nd Edition",
|
|
["Silvanus Phillips Thompson"],
|
|
),
|
|
],
|
|
)
|
|
async def test_upload_book_without_data(
|
|
authenticated_client: AsyncClient,
|
|
path_to_book: Path,
|
|
library_id: int,
|
|
title: str,
|
|
authors: list[str],
|
|
) -> None:
|
|
"""Test uploading a book file. Book information should be extracted from file."""
|
|
# Read file contents
|
|
file_content = path_to_book.read_bytes()
|
|
|
|
# Prepare multipart form data
|
|
files = [("files", (path_to_book.name, file_content, "application/epub+zip"))]
|
|
|
|
# The rest of the book data will be parsed from file
|
|
data = {
|
|
"library_id": library_id,
|
|
}
|
|
|
|
response = await authenticated_client.post(
|
|
f"/books?library_id={library_id}",
|
|
files=files,
|
|
data=data,
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
book_data = response.json()
|
|
|
|
assert book_data["id"] is not None
|
|
assert book_data["title"] == title
|
|
assert book_data["library_id"] == library_id
|
|
|
|
# Check if authors were properly parsed
|
|
book_authors = [author["name"] for author in book_data["authors"]]
|
|
assert book_authors == authors
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("path_to_book", "library_id", "title", "authors", "tags"),
|
|
[
|
|
(
|
|
Path("tests/data_files/Metamorphosis - Franz Kafka.epub"),
|
|
1,
|
|
"The Metamorphosis",
|
|
["Franz Kafka"],
|
|
["Psychological fiction"],
|
|
),
|
|
(
|
|
Path("tests/data_files/Moby Dick; Or, The Whale - Herman Melville.epub"),
|
|
1,
|
|
"Moby Dick",
|
|
["Herman Melville"],
|
|
["Classic Literature", "Whaling"],
|
|
),
|
|
(
|
|
Path(
|
|
"tests/data_files/Relativity: The Special and General Theory - Albert Einstein.epub"
|
|
),
|
|
2,
|
|
"Relativity: the Special and General Theory",
|
|
["Albert Einstein"],
|
|
["Physics", "Mathematics"],
|
|
),
|
|
(
|
|
Path(
|
|
"tests/data_files/On The Origin of Species By Means of Natural Selection - Charles Darwin.epub"
|
|
),
|
|
2,
|
|
"On the Origin of Species By Means of Natural Selection",
|
|
["Charles Darwin"],
|
|
["Biology"],
|
|
),
|
|
(
|
|
Path("tests/data_files/Calculus Made Easy - Silvanus Thompson.pdf"),
|
|
2,
|
|
"Calculus Made Easy",
|
|
["Silvanus Thompson"],
|
|
["Mathematics"],
|
|
),
|
|
],
|
|
)
|
|
async def test_upload_book_with_data(
|
|
authenticated_client: AsyncClient,
|
|
path_to_book: Path,
|
|
library_id: int,
|
|
title: str,
|
|
authors: list[str],
|
|
tags: list[str],
|
|
) -> None:
|
|
"""Test uploading a book file with some book information provided."""
|
|
# Read file contents
|
|
file_content = path_to_book.read_bytes()
|
|
|
|
# Prepare multipart form data
|
|
files = [("files", (path_to_book.name, file_content, "application/epub+zip"))]
|
|
|
|
# The rest of the book data will be parsed from file
|
|
data = {"library_id": library_id, "title": title, "authors": authors, "tags": tags}
|
|
|
|
response = await authenticated_client.post(
|
|
f"/books?library_id={library_id}",
|
|
files=files,
|
|
data=data,
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
book_data = response.json()
|
|
|
|
assert book_data["id"] is not None
|
|
assert book_data["title"] == title
|
|
assert book_data["library_id"] == library_id
|
|
|
|
book_authors = [author["name"] for author in book_data["authors"]]
|
|
assert book_authors == authors
|
|
|
|
book_tags = [tag["name"] for tag in book_data["tags"]]
|
|
assert book_tags == tags
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("path_to_book", "library_id"),
|
|
[
|
|
(Path("tests/data_files/Moby Dick; Or, The Whale - Herman Melville.epub"), 1),
|
|
(Path("tests/data_files/Calculus Made Easy - Silvanus Thompson.pdf"), 2),
|
|
],
|
|
)
|
|
async def test_get_book_file(
|
|
authenticated_client: AsyncClient, path_to_book: Path, library_id: int
|
|
) -> None:
|
|
"""Test uploading a book then downloading the book file."""
|
|
|
|
# Read file contents
|
|
file_content = path_to_book.read_bytes()
|
|
|
|
# Prepare multipart form data
|
|
files = [("files", (path_to_book.name, file_content, "application/epub+zip"))]
|
|
|
|
# The rest of the book data will be parsed from file
|
|
data = {
|
|
"library_id": library_id,
|
|
}
|
|
|
|
create_response = await authenticated_client.post(
|
|
f"/books?library_id={library_id}",
|
|
files=files,
|
|
data=data,
|
|
)
|
|
|
|
assert create_response.status_code == 201
|
|
book_data = create_response.json()
|
|
|
|
# Retrieve the book file
|
|
book_id = book_data["id"]
|
|
file_id = book_data["files"][0]["id"]
|
|
file_response = await authenticated_client.get(
|
|
f"/books/download/{book_id}/{file_id}"
|
|
)
|
|
|
|
assert file_response.status_code == 200
|
|
|
|
downloaded_content = file_response.content
|
|
|
|
assert len(downloaded_content) == len(file_content)
|
|
assert downloaded_content == file_content
|
|
|
|
|
|
async def test_get_book_by_id(populated_authenticated_client: AsyncClient) -> None:
|
|
"""Test retrieving a specific book by ID."""
|
|
|
|
# Retrieve the book
|
|
response = await populated_authenticated_client.get(f"/books/1")
|
|
|
|
assert response.status_code == 200
|
|
book_data = response.json()
|
|
assert book_data["id"] == 1
|
|
assert book_data["title"] == "The Adventures of Sherlock Holmes"
|
|
assert len(book_data["authors"]) == 1
|
|
assert book_data["authors"][0]["name"] == "Arthur Conan Doyle"
|
|
assert len(book_data["tags"]) == 1
|
|
assert book_data["tags"][0]["name"] == "Mystery"
|
|
|
|
|
|
async def test_list_books(populated_authenticated_client: AsyncClient) -> None:
|
|
"""Test listing books with pagination and filters."""
|
|
response = await populated_authenticated_client.get(
|
|
"/books?library_id=1&pageSize=10&page=1"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
assert "items" in data # Pagination structure
|
|
assert isinstance(data.get("items"), list)
|
|
assert data["total"] == 3 # There should be 3 books in this library
|
|
|
|
|
|
async def test_list_books_with_tag_filter(
|
|
populated_authenticated_client: AsyncClient,
|
|
) -> None:
|
|
"""Test listing books filtered by tags."""
|
|
|
|
response = await populated_authenticated_client.get("/books?library_id=1&tags=1")
|
|
|
|
assert response.status_code == 200
|
|
|
|
result = response.json()
|
|
assert len(result["items"]) == 2 # Two books with the "Mystery" tag
|
|
|
|
|
|
async def test_list_books_with_author_filter(
|
|
populated_authenticated_client: AsyncClient,
|
|
) -> None:
|
|
"""Test listing books filtered by authors."""
|
|
response = await populated_authenticated_client.get("/books?library_id=1&authors=1")
|
|
|
|
assert response.status_code == 200
|
|
|
|
result = response.json()
|
|
assert len(result["items"]) == 1 # One book by "Arthur Conan Doyle"
|
|
|
|
|
|
async def test_list_books_with_search(
|
|
populated_authenticated_client: AsyncClient,
|
|
) -> None:
|
|
"""Test listing books with title search."""
|
|
response = await populated_authenticated_client.get(
|
|
"/books?library_id=1&searchString=frankenstein"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
result = response.json()
|
|
assert len(result["items"]) == 1 # One matching book
|
|
|
|
|
|
async def test_delete_book_metadata_only(
|
|
populated_authenticated_client: AsyncClient,
|
|
) -> None:
|
|
"""Test deleting a book record without deleting files."""
|
|
|
|
# Delete book without deleting files
|
|
response = await populated_authenticated_client.delete(
|
|
f"/books?book_ids=3&delete_files=false&library_id=1"
|
|
)
|
|
|
|
assert response.status_code == 204
|
|
|
|
# Verify book is deleted
|
|
get_response = await populated_authenticated_client.get(f"/books/3")
|
|
assert get_response.status_code == 404
|
|
|
|
|
|
async def test_delete_book_with_files(
|
|
populated_authenticated_client: AsyncClient,
|
|
) -> None:
|
|
"""Test deleting a book and its associated files."""
|
|
|
|
# Delete book and files
|
|
response = await populated_authenticated_client.delete(
|
|
f"/books?book_ids=3&delete_files=true&library_id=1"
|
|
)
|
|
|
|
assert response.status_code == 204
|
|
|
|
|
|
async def test_delete_specific_book_files(
|
|
populated_authenticated_client: AsyncClient,
|
|
) -> None:
|
|
"""Test deleting specific files from a book."""
|
|
|
|
# Delete specific file
|
|
response = await populated_authenticated_client.delete(
|
|
f"/books/1/files?file_ids=1",
|
|
)
|
|
|
|
assert response.status_code == 204
|
|
|
|
|
|
async def test_update_reading_progress(
|
|
populated_authenticated_client: AsyncClient,
|
|
) -> None:
|
|
"""Test updating reading progress for a book."""
|
|
|
|
# Update progress
|
|
progress_data = {
|
|
"percentage": 0.5,
|
|
}
|
|
|
|
response = await populated_authenticated_client.post(
|
|
f"/books/progress/1",
|
|
json=progress_data,
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
|
|
# Get the book and check if the progress is correct
|
|
response = await populated_authenticated_client.get("/books/1")
|
|
|
|
assert response.status_code == 200
|
|
|
|
book = response.json()
|
|
assert book["progress"]["percentage"] == 0.5
|
|
|
|
|
|
async def test_create_multiple_books_from_directory(
|
|
authenticated_client: AsyncClient,
|
|
) -> None:
|
|
"""Test creating multiple books from a directory of files."""
|
|
path1 = Path("tests/data_files/Metamorphosis - Franz Kafka.epub")
|
|
path2 = Path("tests/data_files/Moby Dick; Or, The Whale - Herman Melville.epub")
|
|
|
|
files = [
|
|
("files", (path1.name, path1.read_bytes(), "application/epub+zip")),
|
|
("files", (path2.name, path2.read_bytes(), "application/epub+zip")),
|
|
]
|
|
|
|
response = await authenticated_client.post(
|
|
"/books/fromFiles?library_id=1",
|
|
files=files,
|
|
data={"library_id": 1},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert len(data["created"]) == 2
|
|
assert data["skipped"] == []
|
|
|
|
|
|
async def test_create_books_from_parent_directory_keeps_embedded_title(
|
|
authenticated_client: AsyncClient,
|
|
) -> None:
|
|
"""A folder name in the upload path must not override the file's own metadata.
|
|
|
|
The browser sends webkitRelativePath, so picking the shelf above a book's folder
|
|
submits one more path component than picking the folder itself. That extra level
|
|
used to make the directory name win over the title inside the EPUB.
|
|
"""
|
|
source = Path("tests/data_files/Metamorphosis - Franz Kafka.epub")
|
|
files = [
|
|
(
|
|
"files",
|
|
(
|
|
"Shelf/Metamorphosis - Franz Kafka/Metamorphosis.epub",
|
|
source.read_bytes(),
|
|
"application/epub+zip",
|
|
),
|
|
)
|
|
]
|
|
|
|
response = await authenticated_client.post(
|
|
"/books/fromFiles?library_id=1", files=files, data={"library_id": 1}
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
|
|
books = response.json()["created"]
|
|
assert len(books) == 1
|
|
assert books[0]["title"] == "Metamorphosis"
|
|
|
|
|
|
async def test_create_books_groups_formats_within_one_folder(
|
|
authenticated_client: AsyncClient,
|
|
) -> None:
|
|
"""Picking a book's own folder yields one book with both formats, not two books."""
|
|
epub = Path("tests/data_files/Metamorphosis - Franz Kafka.epub").read_bytes()
|
|
pdf = Path("tests/data_files/Calculus Made Easy - Silvanus Thompson.pdf").read_bytes()
|
|
|
|
files = [
|
|
("files", ("Metamorphosis/Metamorphosis.epub", epub, "application/epub+zip")),
|
|
("files", ("Metamorphosis/Metamorphosis.pdf", pdf, "application/pdf")),
|
|
]
|
|
|
|
response = await authenticated_client.post(
|
|
"/books/fromFiles?library_id=1", files=files, data={"library_id": 1}
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
|
|
books = response.json()["created"]
|
|
assert len(books) == 1
|
|
assert len(books[0]["files"]) == 2
|
|
|
|
|
|
class TestDuplicateHandling:
|
|
"""A file the library already holds must not be stored a second time."""
|
|
|
|
epub_path = Path("tests/data_files/Metamorphosis - Franz Kafka.epub")
|
|
|
|
def upload(self, name: str | None = None) -> list[tuple[str, tuple]]:
|
|
return [
|
|
(
|
|
"files",
|
|
(
|
|
name or self.epub_path.name,
|
|
self.epub_path.read_bytes(),
|
|
"application/epub+zip",
|
|
),
|
|
)
|
|
]
|
|
|
|
async def test_bulk_upload_reports_skipped_files(
|
|
self, authenticated_client: AsyncClient
|
|
) -> None:
|
|
"""Re-dropping a folder must import what is new and name what was not."""
|
|
first = await authenticated_client.post(
|
|
"/books/fromFiles?library_id=1", files=self.upload()
|
|
)
|
|
assert first.status_code == 201
|
|
created = first.json()["created"][0]
|
|
|
|
second = await authenticated_client.post(
|
|
"/books/fromFiles?library_id=1", files=self.upload()
|
|
)
|
|
|
|
assert second.status_code == 201
|
|
result = second.json()
|
|
assert result["created"] == []
|
|
assert len(result["skipped"]) == 1
|
|
|
|
skipped = result["skipped"][0]
|
|
assert skipped["filename"] == self.epub_path.name
|
|
assert skipped["book_id"] == created["id"]
|
|
assert skipped["book_title"] == created["title"]
|
|
|
|
async def test_bulk_upload_can_be_forced(
|
|
self, authenticated_client: AsyncClient
|
|
) -> None:
|
|
await authenticated_client.post(
|
|
"/books/fromFiles?library_id=1", files=self.upload()
|
|
)
|
|
|
|
response = await authenticated_client.post(
|
|
"/books/fromFiles?library_id=1&allow_duplicates=true", files=self.upload()
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
assert len(response.json()["created"]) == 1
|
|
assert response.json()["skipped"] == []
|
|
|
|
async def test_single_book_create_conflicts(
|
|
self, authenticated_client: AsyncClient
|
|
) -> None:
|
|
"""Naming files deliberately earns a refusal rather than a silent drop."""
|
|
await authenticated_client.post(
|
|
"/books/fromFiles?library_id=1", files=self.upload()
|
|
)
|
|
|
|
response = await authenticated_client.post(
|
|
"/books?library_id=1", files=self.upload(), data={"library_id": 1}
|
|
)
|
|
|
|
assert response.status_code == 409
|
|
assert response.json()["extra"][0]["filename"] == self.epub_path.name
|
|
|
|
forced = await authenticated_client.post(
|
|
"/books?library_id=1&allow_duplicates=true",
|
|
files=self.upload(),
|
|
data={"library_id": 1},
|
|
)
|
|
assert forced.status_code == 201
|
|
|
|
async def test_adding_another_books_file_conflicts(
|
|
self, authenticated_client: AsyncClient
|
|
) -> None:
|
|
created = await authenticated_client.post(
|
|
"/books/fromFiles?library_id=1", files=self.upload()
|
|
)
|
|
book_id = created.json()["created"][0]["id"]
|
|
|
|
other = await authenticated_client.post(
|
|
"/books?library_id=1",
|
|
files=[
|
|
(
|
|
"files",
|
|
(
|
|
"war.epub",
|
|
Path("tests/data_files/The Art of War - Sun Tzu.epub").read_bytes(),
|
|
"application/epub+zip",
|
|
),
|
|
)
|
|
],
|
|
data={"library_id": 1},
|
|
)
|
|
other_id = other.json()["id"]
|
|
|
|
response = await authenticated_client.post(
|
|
f"/books/{book_id}/files",
|
|
files=[
|
|
(
|
|
"files",
|
|
(
|
|
"war.epub",
|
|
Path("tests/data_files/The Art of War - Sun Tzu.epub").read_bytes(),
|
|
"application/epub+zip",
|
|
),
|
|
)
|
|
],
|
|
)
|
|
|
|
assert response.status_code == 409
|
|
assert response.json()["extra"][0]["book_id"] == other_id
|
|
|
|
async def test_resending_a_books_own_file_changes_nothing(
|
|
self, authenticated_client: AsyncClient
|
|
) -> None:
|
|
created = await authenticated_client.post(
|
|
"/books/fromFiles?library_id=1", files=self.upload()
|
|
)
|
|
book_id = created.json()["created"][0]["id"]
|
|
|
|
response = await authenticated_client.post(
|
|
f"/books/{book_id}/files", files=self.upload()
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
assert len(response.json()["files"]) == 1
|
|
|
|
async def test_duplicates_can_be_checked_before_uploading(
|
|
self, authenticated_client: AsyncClient
|
|
) -> None:
|
|
"""The pre-flight check answers from hashes alone, with no file sent."""
|
|
created = await authenticated_client.post(
|
|
"/books/fromFiles?library_id=1", files=self.upload()
|
|
)
|
|
book = created.json()["created"][0]
|
|
stored = book["files"][0]
|
|
|
|
response = await authenticated_client.post(
|
|
"/books/duplicates?library_id=1",
|
|
json=[
|
|
{
|
|
"hash": stored["hash"],
|
|
"size": stored["size"],
|
|
"filename": "local-copy.epub",
|
|
},
|
|
{"hash": stored["hash"], "size": stored["size"] + 1},
|
|
{"hash": "0" * 32, "size": 1234},
|
|
],
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
matches = response.json()
|
|
assert len(matches) == 1
|
|
assert matches[0]["filename"] == "local-copy.epub"
|
|
assert matches[0]["book_id"] == book["id"]
|
|
|
|
|
|
# NOTE: the multi-book ZIP download is covered at the service level, in
|
|
# tests/unit/test_services/test_book_service.py. Driving `/books/download` through
|
|
# AsyncTestClient hangs in fixture teardown: it is the only `Stream` endpoint in the
|
|
# app, and the test transport never sends the `http.disconnect` that Litestar's
|
|
# streaming response waits on, so the app's lifespan shutdown never completes.
|
|
|
|
|
|
# async def test_delete_book_metadata(authenticated_client: AsyncClient) -> None:
|
|
# raise NotImplementedError()
|
|
|
|
|
|
# async def test_delete_book_metadata_and_files(
|
|
# authenticated_client: AsyncClient,
|
|
# ) -> None:
|
|
# raise NotImplementedError()
|
|
|
|
|
|
# async def test_edit_book_metadata(authenticated_client: AsyncClient) -> None:
|
|
# raise NotImplementedError()
|
|
|
|
import pytest
|
|
import aiofiles
|
|
from httpx import AsyncClient
|
|
from pathlib import Path
|
|
from litestar.status_codes import HTTP_400_BAD_REQUEST
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from datetime import date
|
|
|
|
|
|
class TestMetadataUpdates:
|
|
@pytest.mark.parametrize(
|
|
("updated_field", "new_value"),
|
|
[
|
|
("title", "New Title"),
|
|
("subtitle", "Updated Subtitle"),
|
|
("pages", 256),
|
|
("description", "An updated description"),
|
|
("edition", 2),
|
|
("language", "pl"),
|
|
("published_date", "1910-04-05"),
|
|
("series_position", "3"),
|
|
],
|
|
)
|
|
async def test_update_single_scalar_field(
|
|
self,
|
|
populated_authenticated_client: AsyncClient,
|
|
updated_field: str,
|
|
new_value: str,
|
|
) -> None:
|
|
"""Test updating a single field without affecting others."""
|
|
|
|
# Get original book state
|
|
original_response = await populated_authenticated_client.get("/books/1")
|
|
assert original_response.status_code == 200
|
|
original_data = original_response.json()
|
|
|
|
# Update field
|
|
response = await populated_authenticated_client.patch(
|
|
"/books/1",
|
|
json={updated_field: str(new_value)},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
updated_data = response.json()
|
|
|
|
# Verify updated field
|
|
assert updated_data[updated_field] == new_value
|
|
|
|
# Verify all other fields remain unchanged
|
|
for field, original_value in original_data.items():
|
|
if field != updated_field and field not in ["updated_at", "created_at"]:
|
|
assert updated_data[field] == original_value, (
|
|
f"Field {field} was unexpectedly changed"
|
|
)
|
|
|
|
@pytest.mark.parametrize(
|
|
("updated_field", "new_values", "assertion_func"),
|
|
[
|
|
(
|
|
"authors", # Update with new authors
|
|
["New Author 1", "New Author 2"],
|
|
lambda data: {a["name"] for a in data["authors"]}
|
|
== {"New Author 1", "New Author 2"},
|
|
),
|
|
(
|
|
"authors", # Clear authors
|
|
[],
|
|
lambda data: data["authors"] == [],
|
|
),
|
|
(
|
|
"tags", # Update with new tags
|
|
["Tag 1", "Tag 2", "Tag 3"],
|
|
lambda data: {t["name"] for t in data["tags"]}
|
|
== {"Tag 1", "Tag 2", "Tag 3"},
|
|
),
|
|
(
|
|
"tags", # Clear tags
|
|
[],
|
|
lambda data: data["tags"] == [],
|
|
),
|
|
(
|
|
"publisher", # Update with new publisher
|
|
"Updated Publisher",
|
|
lambda data: data["publisher"]["name"] == "Updated Publisher",
|
|
),
|
|
(
|
|
"publisher", # Clear publisher
|
|
None,
|
|
lambda data: data["publisher"] is None,
|
|
),
|
|
(
|
|
"identifiers", # Update with new identifiers
|
|
{"isbn-13": "978-1234567890", "doi": "10.example/id"},
|
|
lambda data: data["identifiers"]
|
|
== {"isbn-13": "978-1234567890", "doi": "10.example/id"},
|
|
),
|
|
(
|
|
"identifiers", # Clear identifiers
|
|
{},
|
|
lambda data: data["identifiers"] == {},
|
|
),
|
|
(
|
|
"series", # Update with new series
|
|
"Updated Series",
|
|
lambda data: data["series"]["title"] == "Updated Series",
|
|
),
|
|
(
|
|
"series", # Clear series
|
|
None,
|
|
lambda data: data["series"] is None,
|
|
),
|
|
],
|
|
)
|
|
async def test_update_relationship_fields(
|
|
self,
|
|
populated_authenticated_client: AsyncClient,
|
|
updated_field: str,
|
|
new_values: list[str],
|
|
assertion_func,
|
|
) -> None:
|
|
"""Test updating relationship fields (authors, tags, etc.)"""
|
|
|
|
# Get original book state
|
|
original_response = await populated_authenticated_client.get("/books/1")
|
|
assert original_response.status_code == 200
|
|
original_data = original_response.json()
|
|
|
|
response = await populated_authenticated_client.patch(
|
|
"/books/1",
|
|
json={updated_field: new_values},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
# Verify updated field
|
|
updated_data = response.json()
|
|
assert assertion_func(updated_data)
|
|
|
|
# Verify all other fields remain unchanged
|
|
for field, original_value in original_data.items():
|
|
if field != updated_field and field not in ["updated_at", "created_at"]:
|
|
assert updated_data[field] == original_value, (
|
|
f"Field {field} was unexpectedly changed"
|
|
)
|
|
|
|
async def test_update_with_invalid_pages_value(
|
|
self, populated_authenticated_client: AsyncClient
|
|
) -> None:
|
|
"""Test updating with invalid pages value (non-integer)."""
|
|
data = {
|
|
"pages": "not_a_number",
|
|
}
|
|
|
|
response = await populated_authenticated_client.patch(
|
|
"/books/1",
|
|
files=data,
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
async def test_update_with_invalid_edition_value(
|
|
self, populated_authenticated_client: AsyncClient
|
|
) -> None:
|
|
"""Test updating with invalid edition value (non-integer)."""
|
|
data = {
|
|
"edition": "invalid_edition",
|
|
}
|
|
|
|
response = await populated_authenticated_client.patch(
|
|
"/books/1",
|
|
files=data,
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
async def test_update_with_invalid_date_format(
|
|
self, populated_authenticated_client: AsyncClient
|
|
) -> None:
|
|
"""Test updating with invalid published_date format."""
|
|
data = {
|
|
"published_date": "invalid-date",
|
|
}
|
|
|
|
response = await populated_authenticated_client.patch(
|
|
"/books/1",
|
|
files=data,
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
async def test_update_with_invalid_identifiers_json(
|
|
self, populated_authenticated_client: AsyncClient
|
|
) -> None:
|
|
"""Test updating with invalid JSON in identifiers."""
|
|
data = {
|
|
"identifiers": "not valid json",
|
|
}
|
|
|
|
response = await populated_authenticated_client.patch(
|
|
"/books/1",
|
|
files=data,
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
async def test_update_multiple_fields_at_once(
|
|
self, populated_authenticated_client: AsyncClient
|
|
) -> None:
|
|
"""Test updating multiple fields simultaneously."""
|
|
data = {
|
|
"title": "New Title",
|
|
"description": "New description",
|
|
"publisher": "New Publisher",
|
|
"pages": 430,
|
|
}
|
|
|
|
response = await populated_authenticated_client.patch(
|
|
"/books/1",
|
|
json=data,
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
book_data = response.json()
|
|
assert book_data["title"] == "New Title"
|
|
assert book_data["description"] == "New description"
|
|
assert book_data["publisher"]["name"] == "New Publisher"
|
|
assert book_data["pages"] == 430
|
|
|
|
async def test_update_nonexistent_book(
|
|
self, populated_authenticated_client: AsyncClient
|
|
) -> None:
|
|
"""Test updating a nonexistent book returns 404."""
|
|
data = {
|
|
"title": "New Title",
|
|
}
|
|
|
|
response = await populated_authenticated_client.patch(
|
|
"/books/99999",
|
|
json=data,
|
|
)
|
|
assert response.status_code == 404
|
|
assert "book does not exist" in response.text
|
|
|
|
@pytest.mark.parametrize(
|
|
("updated_field"),
|
|
[
|
|
("subtitle"),
|
|
("pages"),
|
|
("description"),
|
|
("edition"),
|
|
("language"),
|
|
("published_date"),
|
|
("series_position"),
|
|
],
|
|
)
|
|
async def test_update_clears_optional_field(
|
|
self, populated_authenticated_client: AsyncClient, updated_field: str
|
|
) -> None:
|
|
"""Test that optional fields can be cleared by passing None or empty values."""
|
|
|
|
data = {updated_field: None}
|
|
|
|
response = await populated_authenticated_client.patch(
|
|
"/books/1",
|
|
json=data,
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
result = response.json()
|
|
|
|
assert result[updated_field] == None
|
|
|
|
@pytest.mark.parametrize(
|
|
("updated_field"),
|
|
[
|
|
("title"),
|
|
],
|
|
)
|
|
async def test_update_clears_required_field(
|
|
self, populated_authenticated_client: AsyncClient, updated_field: str
|
|
) -> None:
|
|
"""Test that optional fields can be cleared by passing None or empty values."""
|
|
|
|
data = {updated_field: None}
|
|
|
|
response = await populated_authenticated_client.patch(
|
|
"/books/1",
|
|
json=data,
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
async def test_update_cover_successfully(
|
|
self, populated_authenticated_client: AsyncClient
|
|
) -> None:
|
|
"""Test successfully updating a book's cover image."""
|
|
path = Path("tests/data_files/cover.jpg")
|
|
file_content = path.read_bytes()
|
|
|
|
files = {"cover_image": (path.name, file_content, "image/jpeg")}
|
|
|
|
# Get original book state
|
|
original_response = await populated_authenticated_client.get("/books/1")
|
|
assert original_response.status_code == 200
|
|
original_data = original_response.json()
|
|
|
|
response = await populated_authenticated_client.put(
|
|
"/books/1/cover",
|
|
files=files,
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
book_data = response.json()
|
|
assert book_data["id"] == 1
|
|
|
|
assert book_data["cover_image"] is not None
|
|
assert book_data["cover_image"] != original_data["cover_image"]
|
|
|
|
|
|
class TestFileManagement:
|
|
async def test_add_multiple_files_to_book(
|
|
self, populated_authenticated_client: AsyncClient
|
|
) -> None:
|
|
"""Test adding multiple files to a book in a single request."""
|
|
path1 = Path("tests/data_files/Metamorphosis - Franz Kafka.epub")
|
|
path2 = Path("tests/data_files/Calculus Made Easy - Silvanus Thompson.pdf")
|
|
|
|
file1_content = path1.read_bytes()
|
|
file2_content = path2.read_bytes()
|
|
|
|
files = [
|
|
("files", (path1.name, file1_content, "application/epub+zip")),
|
|
("files", (path2.name, file2_content, "application/pdf")),
|
|
]
|
|
|
|
response = await populated_authenticated_client.post(
|
|
"/books/1/files",
|
|
files=files,
|
|
)
|
|
assert response.status_code == 201
|
|
|
|
book_data = response.json()
|
|
assert len(book_data["files"]) >= 2
|
|
|
|
async def test_add_files_to_nonexistent_book(
|
|
self, populated_authenticated_client: AsyncClient
|
|
) -> None:
|
|
"""Test adding files to a nonexistent book returns 404."""
|
|
path = Path("tests/data_files/Metamorphosis - Franz Kafka.epub")
|
|
file_content = path.read_bytes()
|
|
files = [("files", (path.name, file_content, "application/epub+zip"))]
|
|
|
|
response = await populated_authenticated_client.post(
|
|
"/books/99999/files?library_id=1",
|
|
files=files,
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
async def test_remove_multiple_files_at_once(
|
|
self, populated_authenticated_client: AsyncClient
|
|
) -> None:
|
|
"""Test removing multiple files from a book at once."""
|
|
# First add multiple files
|
|
path1 = Path("tests/data_files/Metamorphosis - Franz Kafka.epub")
|
|
path2 = Path("tests/data_files/Calculus Made Easy - Silvanus Thompson.pdf")
|
|
|
|
files = [
|
|
("files", (path1.name, path1.read_bytes(), "application/epub+zip")),
|
|
("files", (path2.name, path2.read_bytes(), "application/pdf")),
|
|
]
|
|
|
|
add_response = await populated_authenticated_client.post(
|
|
"/books/1/files",
|
|
files=files,
|
|
)
|
|
assert add_response.status_code == 201
|
|
|
|
book_data = add_response.json()
|
|
file_ids = [f["id"] for f in book_data["files"]]
|
|
|
|
# Remove multiple files
|
|
response = await populated_authenticated_client.delete(
|
|
f"/books/1/files?file_ids={file_ids[0]}&file_ids={file_ids[1]}&delete_files=false",
|
|
)
|
|
assert response.status_code == 204
|
|
|
|
# Verify files were removed
|
|
book = await populated_authenticated_client.get("/books/1")
|
|
book_data = book.json()
|
|
assert len(book_data["files"]) < len(file_ids)
|
|
|
|
async def test_remove_zero_files_raises_error(
|
|
self, populated_authenticated_client: AsyncClient
|
|
) -> None:
|
|
"""Test error when trying to remove zero files."""
|
|
response = await populated_authenticated_client.delete(
|
|
"/books/1/files?delete_files=false",
|
|
)
|
|
assert response.status_code == HTTP_400_BAD_REQUEST
|
|
assert "missing required query parameter 'file_ids'" in response.text.lower()
|
|
|
|
async def test_remove_files_from_nonexistent_book(
|
|
self, populated_authenticated_client: AsyncClient
|
|
) -> None:
|
|
"""Test removing files from a nonexistent book returns 404."""
|
|
response = await populated_authenticated_client.delete(
|
|
"/books/99999/files?file_ids=1&delete_files=false&library_id=1",
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
async def test_remove_nonexistent_file_id_from_book(
|
|
self, populated_authenticated_client: AsyncClient
|
|
) -> None:
|
|
"""Test removing a file_id that doesn't exist on that book."""
|
|
# Try to remove a file that doesn't belong to this book
|
|
response = await populated_authenticated_client.delete(
|
|
"/books/1/files?file_ids=99999&delete_files=false",
|
|
)
|
|
# Should succeed (idempotent) or return 404, depending on implementation
|
|
assert response.status_code in [204, 404]
|
|
|
|
async def test_remove_file_with_delete_files_false_keeps_filesystem_file(
|
|
self, populated_authenticated_client: AsyncClient
|
|
) -> None:
|
|
"""Test that removing with delete_files=false keeps file on disk."""
|
|
# Get the book to retrieve file info
|
|
book_response = await populated_authenticated_client.get("/books/1")
|
|
book_data = book_response.json()
|
|
|
|
if not book_data["files"]:
|
|
pytest.skip("Book has no files")
|
|
|
|
file_id = book_data["files"][0]["id"]
|
|
filename = book_data["files"][0].get("path")
|
|
|
|
# Remove file without deleting from filesystem
|
|
response = await populated_authenticated_client.delete(
|
|
f"/books/1/files?file_ids={file_id}&delete_files=false",
|
|
)
|
|
assert response.status_code == 204
|
|
|
|
async def test_remove_file_with_delete_files_true_removes_filesystem_file(
|
|
self, populated_authenticated_client: AsyncClient
|
|
) -> None:
|
|
"""Test that removing with delete_files=true deletes from filesystem."""
|
|
# Add a new file first
|
|
path = Path("tests/data_files/Calculus Made Easy - Silvanus Thompson.pdf")
|
|
file_content = path.read_bytes()
|
|
files = [("files", (path.name, file_content, "application/pdf"))]
|
|
|
|
add_response = await populated_authenticated_client.post(
|
|
"/books/1/files",
|
|
files=files,
|
|
)
|
|
assert add_response.status_code == 201
|
|
|
|
book_data = add_response.json()
|
|
file_id = book_data["files"][-1]["id"]
|
|
file_path = book_data["files"][-1].get("path")
|
|
|
|
# Remove file with deletion from filesystem
|
|
response = await populated_authenticated_client.delete(
|
|
f"/books/1/files?file_ids={file_id}&delete_files=true",
|
|
)
|
|
assert response.status_code == 204
|
|
|
|
async def test_remove_file_idempotent(
|
|
self, populated_authenticated_client: AsyncClient
|
|
) -> None:
|
|
"""Test that removing the same file twice doesn't error on second attempt."""
|
|
book_response = await populated_authenticated_client.get("/books/1")
|
|
book_data = book_response.json()
|
|
|
|
if not book_data["files"]:
|
|
pytest.skip("Book has no files")
|
|
|
|
file_id = book_data["files"][0]["id"]
|
|
|
|
# Remove file first time
|
|
response1 = await populated_authenticated_client.delete(
|
|
f"/books/1/files?file_ids={file_id}&delete_files=false",
|
|
)
|
|
assert response1.status_code == 204
|
|
|
|
# Try to remove same file again
|
|
response2 = await populated_authenticated_client.delete(
|
|
f"/books/1/files?file_ids={file_id}&delete_files=false",
|
|
)
|
|
# Should succeed (idempotent)
|
|
assert response2.status_code == 204
|