Update Last Changed only when Bindery stores a new chapter or adds a new chapter revision. Manual ePub generation, rebuilds, metadata edits and rule changes leave it unchanged. Save the timestamp in the same transaction as the chapter or revision. This preserves the date if a later build fails and prevents retries from moving it forward without another chapter change. Add regression tests for new chapters, revisions, forced rebuilds, failed builds, retries and transaction rollback. Adjust OPDS test data setup and document what Last Changed records.
279 lines
12 KiB
Python
279 lines
12 KiB
Python
"""OPDS navigation, book lists, downloads and the shared login gate."""
|
|
|
|
from datetime import UTC, datetime
|
|
from xml.etree import ElementTree as ET
|
|
|
|
import pytest
|
|
from test_api import client as client
|
|
from test_api import cwa as cwa
|
|
from test_build import PNG
|
|
|
|
from bindery import diffing, fff, opds, orchestrator
|
|
|
|
NS = {"a": opds.ATOM}
|
|
ACQUIRE = "http://opds-spec.org/acquisition"
|
|
PATHS = ["/opds", "/opds/", "/opds/title", "/opds/author", "/opds/changed"]
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def enabled_catalog(client):
|
|
orchestrator.store.set_setting(None, "opds_enabled", True)
|
|
|
|
|
|
def feed(response):
|
|
assert response.status_code == 200
|
|
assert response.mimetype == "application/atom+xml"
|
|
assert response.headers["Cache-Control"] == "private, no-store"
|
|
root = ET.fromstring(response.data)
|
|
assert root.tag == f"{{{opds.ATOM}}}feed"
|
|
assert root.findtext("a:id", namespaces=NS)
|
|
assert root.findtext("a:title", namespaces=NS)
|
|
assert datetime.fromisoformat(root.findtext("a:updated", namespaces=NS)).tzinfo is not None
|
|
return root
|
|
|
|
|
|
def links(element):
|
|
return {link.attrib["rel"]: link.attrib["href"] for link in element.findall("a:link", NS)}
|
|
|
|
|
|
def entries(root):
|
|
return root.findall("a:entry", NS)
|
|
|
|
|
|
def titles(root):
|
|
return [entry.findtext("a:title", namespaces=NS) for entry in entries(root)]
|
|
|
|
|
|
def add_book(title="A book", author="Writer", changed=None, url=None, included=True, meta=None):
|
|
store = orchestrator.store
|
|
url = url or f"https://example.com/books/{len(store.list_stories()) + 1}"
|
|
sid = store.create_story(url, "example.com", title, author, [])
|
|
if included:
|
|
html = "<p>Stored text.</p>"
|
|
store.add_chapter(sid, url + "/1", "One", html, diffing.text_hash(html), "site", 0)
|
|
store.update_story(sid, last_changed=changed, site_values=meta or {})
|
|
return sid
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def no_site_requests(monkeypatch):
|
|
def fail(*args, **kwargs):
|
|
pytest.fail("OPDS must not contact a story site")
|
|
|
|
for name in ("fetch_story", "fetch_chapter", "collected_images"):
|
|
monkeypatch.setattr(fff, name, fail)
|
|
|
|
|
|
def test_navigation_and_empty_lists(client):
|
|
root = feed(client.get("/opds"))
|
|
assert titles(root) == [title for title, _, _ in opds.LISTS.values()]
|
|
assert len({entry.findtext("a:id", namespaces=NS) for entry in entries(root)}) == 3
|
|
assert client.get("/opds/").data == client.get("/opds").data
|
|
for entry, (_, kind, text) in zip(entries(root), opds.LISTS.values(), strict=True):
|
|
assert entry.findtext("a:content", namespaces=NS) == text
|
|
link = entry.find("a:link", NS)
|
|
assert link.attrib["type"] == kind
|
|
books = feed(client.get(link.attrib["href"]))
|
|
assert entries(books) == []
|
|
assert links(books)["start"] == links(books)["up"] == "/opds"
|
|
assert "next" not in links(books)
|
|
|
|
|
|
def test_sort_orders_dates_and_stable_book_ids(client):
|
|
add_book("The Zebra", "alice", "2024-01-01T12:00:00+00:00")
|
|
add_book("An Apple", "Bob", "2024-03-01T12:00:00+00:00")
|
|
add_book("Banana", "ALICE", "2024-02-01T12:00:00+00:00")
|
|
expected = {
|
|
"title": ["An Apple", "Banana", "The Zebra"],
|
|
"author?name=Alice": ["Banana", "The Zebra"],
|
|
"changed": ["An Apple", "Banana", "The Zebra"],
|
|
}
|
|
ids = []
|
|
for order, names in expected.items():
|
|
root = feed(client.get(f"/opds/{order}"))
|
|
assert titles(root) == names
|
|
if order != "author?name=Alice":
|
|
assert root.findtext("a:updated", namespaces=NS) == "2024-03-01T12:00:00Z"
|
|
ids.append({e.findtext("a:id", namespaces=NS) for e in entries(root)})
|
|
for entry in entries(root):
|
|
assert entry.findtext("a:author/a:name", namespaces=NS)
|
|
assert entry.findtext("a:content", namespaces=NS) == "Chapters: 1"
|
|
assert entry.find("a:link", NS).attrib["type"] == "application/epub+zip"
|
|
assert ids[0] == ids[2] > ids[1]
|
|
|
|
|
|
def test_dates_use_creation_fallback_and_compare_offsets(client):
|
|
sid = add_book("Created")
|
|
created = orchestrator.store.get_story(sid)["created_at"]
|
|
expected = datetime.fromisoformat(created).astimezone(UTC).isoformat().replace("+00:00", "Z")
|
|
root = feed(client.get("/opds/changed"))
|
|
assert entries(root)[0].findtext("a:updated", namespaces=NS) == expected
|
|
orchestrator.store.update_story(sid, last_changed="2024-01-01T12:00:00+02:00")
|
|
add_book("Later", changed="2024-01-01T11:00:00+00:00")
|
|
assert titles(feed(client.get("/opds/changed"))) == ["Later", "Created"]
|
|
|
|
|
|
def test_xml_escaping_unicode_and_missing_metadata(client):
|
|
add_book('Books <&> "日本語"\x00', "A & B <writers>\x0b")
|
|
empty = add_book("", "")
|
|
root = feed(client.get("/opds/title"))
|
|
assert titles(root)[0] == 'Books <&> "日本語"'
|
|
assert entries(root)[0].findtext("a:author/a:name", namespaces=NS) == "A & B <writers>"
|
|
assert titles(root)[1] == orchestrator.store.get_story(empty)["url"]
|
|
assert entries(root)[1].findtext("a:author/a:name", namespaces=NS) == "Unknown"
|
|
|
|
|
|
def test_only_books_with_included_chapters_appear(client):
|
|
add_book("Pending", included=False)
|
|
sid = add_book("Excluded")
|
|
chapter = orchestrator.store.list_chapters(sid)[0]
|
|
orchestrator.store.update_chapter(chapter["id"], included=False)
|
|
add_book("Ready")
|
|
assert titles(feed(client.get("/opds/title"))) == ["Ready"]
|
|
|
|
|
|
@pytest.mark.parametrize("order", ["title", "changed", "author?name=Same", "search?q=same"])
|
|
def test_pagination_with_ties(client, monkeypatch, order):
|
|
monkeypatch.setattr(opds, "PAGE_SIZE", 2)
|
|
for _ in range(5):
|
|
add_book("Same", "Same", "2024-01-01T00:00:00+00:00")
|
|
first = feed(client.get(f"/opds/{order}"))
|
|
second = feed(client.get(links(first)["next"]))
|
|
last = feed(client.get(links(second)["next"]))
|
|
assert [len(entries(root)) for root in (first, second, last)] == [2, 2, 1]
|
|
assert "previous" not in links(first)
|
|
assert "next" not in links(last)
|
|
assert links(second)["previous"] == links(first)["self"]
|
|
assert links(last)["previous"] == links(second)["self"]
|
|
ids = [e.findtext("a:id", namespaces=NS) for root in (first, second, last) for e in entries(root)]
|
|
assert ids == [f"https://example.com/books/{i}" for i in range(1, 6)]
|
|
assert client.get(links(first)["self"].replace("page=1", "page=4")).status_code == 404
|
|
|
|
|
|
@pytest.mark.parametrize("page", ["0", "-1", "1.5", "abc", "", "999999999999999999999"])
|
|
def test_bad_page(client, page):
|
|
assert client.get("/opds/title", query_string={"page": page}).status_code == 400
|
|
|
|
|
|
def test_unknown_list(client):
|
|
assert client.get("/opds/missing").status_code == 404
|
|
|
|
|
|
DC = {"a": opds.ATOM, "dc": opds.DC}
|
|
|
|
|
|
def test_entry_shows_the_book_metadata_and_cover(client):
|
|
sid = add_book(
|
|
"Stored title",
|
|
meta={
|
|
"title": "The Book",
|
|
"authors": ["Ann Writer", "Ben Reader"],
|
|
"comments": "<p>First <b>part</b>.</p><p>Second part.</p>",
|
|
"tags": ["Space", "Fiction"],
|
|
"series": "The Saga",
|
|
"series_index": 2.0,
|
|
"pubdate": "2020-05-06T12:00:00+00:00",
|
|
"languages": "en",
|
|
},
|
|
)
|
|
orchestrator.store.update_story(sid, pinned={"title": "The Kept Title"})
|
|
entry = entries(feed(client.get("/opds/title")))[0]
|
|
assert entry.findtext("a:title", namespaces=NS) == "The Kept Title"
|
|
assert [a.findtext("a:name", namespaces=NS) for a in entry.findall("a:author", NS)] == ["Ann Writer", "Ben Reader"]
|
|
assert entry.find("a:author/a:uri", NS).text == "/opds/author?name=Ann+Writer"
|
|
assert entry.find("a:content", NS).attrib["type"] == "text"
|
|
assert entry.findtext("a:content", namespaces=NS) == (
|
|
"Series: The Saga, book 2\nChapters: 1\n\nFirst part.\n\nSecond part."
|
|
)
|
|
assert [c.attrib["term"] for c in entry.findall("a:category", NS)] == ["Space", "Fiction"]
|
|
assert entry.findtext("dc:language", namespaces=DC) == "en"
|
|
assert entry.findtext("dc:issued", namespaces=DC) == "2020-05-06"
|
|
assert entry.findtext("dc:publisher", namespaces=DC) == "example.com"
|
|
assert "http://opds-spec.org/image" not in links(entry)
|
|
|
|
orchestrator.store.put_image(sid, "https://example.com/cover.png", "png", "image/png", PNG, is_cover=True)
|
|
entry = entries(feed(client.get("/opds/title")))[0]
|
|
image = links(entry)["http://opds-spec.org/image"]
|
|
assert image == links(entry)["http://opds-spec.org/image/thumbnail"]
|
|
assert image.startswith(f"/opds/books/{sid}/cover?v=")
|
|
response = client.get(image)
|
|
assert response.data == PNG
|
|
assert response.mimetype == "image/png"
|
|
assert "private" in response.headers["Cache-Control"]
|
|
orchestrator.store.put_image(sid, "https://example.com/new.png", "png", "image/png", PNG, is_cover=True)
|
|
assert links(entries(feed(client.get("/opds/title")))[0])["http://opds-spec.org/image"] != image
|
|
assert client.get(f"/opds/books/{add_book()}/cover").status_code == 404
|
|
|
|
|
|
def test_entry_shows_the_page_count_of_a_built_book(client):
|
|
sid = add_book()
|
|
orchestrator.store.set_setting(sid, "page_count", 321)
|
|
add_book(title="Not built")
|
|
contents = {
|
|
e.findtext("a:title", namespaces=NS): e.findtext("a:content", namespaces=NS)
|
|
for e in entries(feed(client.get("/opds/title")))
|
|
}
|
|
assert contents == {"A book": "Chapters: 1\nPages: 321", "Not built": "Chapters: 1"}
|
|
|
|
|
|
def test_cover_of_an_unsafe_type_is_left_out(client):
|
|
sid = add_book()
|
|
orchestrator.store.put_image(sid, "https://example.com/c.svg", "svg", "image/svg+xml", b"<svg/>", is_cover=True)
|
|
assert "http://opds-spec.org/image" not in links(entries(feed(client.get("/opds/title")))[0])
|
|
assert client.get(f"/opds/books/{sid}/cover").status_code == 404
|
|
|
|
|
|
def test_authors_in_surname_order_with_their_books(client):
|
|
add_book("One", "Zed Adams")
|
|
add_book("Two", "ann writer")
|
|
add_book("Three", "", meta={"authors": ["Ann Writer", "Bob Brown"]})
|
|
add_book("Four", "")
|
|
root = feed(client.get("/opds/author"))
|
|
assert titles(root) == ["Zed Adams", "Bob Brown", "Unknown", "Ann Writer"]
|
|
assert [e.findtext("a:content", namespaces=NS) for e in entries(root)] == ["1 book", "1 book", "1 book", "2 books"]
|
|
assert links(root)["up"] == "/opds"
|
|
writer = feed(client.get(links(entries(root)[3])["subsection"]))
|
|
assert titles(writer) == ["Three", "Two"]
|
|
assert links(writer)["up"] == "/opds/author"
|
|
assert client.get("/opds/author", query_string={"name": "Nobody"}).status_code == 404
|
|
|
|
|
|
def test_author_list_pages(client, monkeypatch):
|
|
monkeypatch.setattr(opds, "PAGE_SIZE", 2)
|
|
for name in ("A One", "B Two", "C Three"):
|
|
add_book(author=name)
|
|
first = feed(client.get("/opds/author"))
|
|
assert titles(first) == ["A One", "C Three"]
|
|
assert titles(feed(client.get(links(first)["next"]))) == ["B Two"]
|
|
|
|
|
|
def test_search(client):
|
|
add_book("The Long Road", "Ann Writer", meta={"tags": ["Adventure"]})
|
|
add_book("Short Road", "Ben Reader", meta={"series": "Roads"})
|
|
add_book("Other", "Cat Smith")
|
|
|
|
def search(q):
|
|
return titles(feed(client.get("/opds/search", query_string={"q": q})))
|
|
|
|
assert search("road") == ["The Long Road", "Short Road"]
|
|
assert search("ROAD writer") == ["The Long Road"]
|
|
assert search("adventure") == ["The Long Road"]
|
|
assert search("roads") == ["Short Road"]
|
|
assert search("missing") == search("") == search(" ") == []
|
|
assert feed(client.get("/opds/search")).findtext("a:title", namespaces=NS) == "Search: "
|
|
|
|
|
|
def test_every_feed_links_the_search(client):
|
|
add_book()
|
|
for path in [*PATHS, "/opds/author?name=Writer", "/opds/search?q=book"]:
|
|
searches = [link for link in feed(client.get(path)).findall("a:link", NS) if link.attrib["rel"] == "search"]
|
|
found = {link.attrib["type"]: link.attrib["href"] for link in searches}
|
|
assert found == {
|
|
"application/opensearchdescription+xml": "/opds/search.xml",
|
|
opds.ACQUISITION: "/opds/search?q={searchTerms}",
|
|
}
|
|
response = client.get("/opds/search.xml")
|
|
assert response.mimetype == "application/opensearchdescription+xml"
|
|
url = ET.fromstring(response.data).find(f"{{{opds.OPENSEARCH}}}Url")
|
|
assert url.attrib == {"type": opds.ACQUISITION, "template": "/opds/search?q={searchTerms}"}
|