patrick/chitai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
bd8d68b9babde59fe21ee5fdb09ede6cf34d19bc
chitai/backend/tests/integration/test_book.py
T
patrick bd8d68b9ba fix: resolve file paths and stream the multi-book zip download 
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.
2026-08-12 15:17:48 -04:00

878 lines
28 KiB
Python
Raw Blame History

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.get("items") or data.get("data")) >= 1
# 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
Reference in New Issue View Git Blame Copy Permalink