patrick/chitai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
a19c944b6e6dc8473b249145a681443402c5d68c
chitai/backend/tests/integration/test_book.py
patrick a19c944b6e Fix hardcoded absolute paths in book upload tests 
The Darwin epub test cases used absolute paths. I hadn't notices until switching machines, whcih caused errors in the test harness. Changed to relative paths to be consistent with other test cases.
2026-03-07 11:35:30 -05:00

871 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 = {
"progress": 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"]["progress"] == 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
# 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