tim
9196dd75a8
A CWA with anonymous browsing can answer an OPDS request whatever the password, so each password logged a user in to Bindery. Bindery now tests CWA with a login that cannot exist, and refuses OPDS logins if CWA accepts it. A login check no longer shares the session of the account of Bindery. The CWA account stays optional for a CWA that asks for no login, and a missing account now gets its own message.
252 lines
9.9 KiB
Python
252 lines
9.9 KiB
Python
"""The OPDS client, the book finder, the login check, the ingest folder and the epub reader."""
|
|
|
|
import json
|
|
import time
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
from fake_cwa import FakeBook, FakeCwa
|
|
from test_build import URL, chapters, metahtml
|
|
|
|
from bindery import fff
|
|
from bindery.cwa import LOGIN_TEST_SECONDS, BookFinder, CwaClient, CwaError, plain_title, search_text, title_rank
|
|
from bindery.epub import epub_metadata, epub_urls
|
|
from bindery.ingest import MAX_NAME_LENGTH, IngestError, IngestFolder
|
|
|
|
STORY_URL = "https://www.royalroad.com/fiction/1234"
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def story_epub(tmp_path_factory):
|
|
path = tmp_path_factory.mktemp("epub") / "story.epub"
|
|
fff.build_epub(URL, metahtml(), chapters(), [], None, [], {"series": ["Saga", 2.0]}, path)
|
|
return path.read_bytes()
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def other_epub(tmp_path_factory):
|
|
path = tmp_path_factory.mktemp("epub") / "other.epub"
|
|
other_url = "https://www.royalroad.com/fiction/9999/other-story"
|
|
other_chapters = [dict(c, url=c["url"].replace("1234/test-story", "9999/other-story")) for c in chapters()]
|
|
fff.build_epub(other_url, metahtml(other_url), other_chapters, [], None, [], {}, path)
|
|
return path.read_bytes()
|
|
|
|
|
|
@pytest.fixture
|
|
def fake(story_epub, other_epub):
|
|
server = FakeCwa(page_size=1)
|
|
other = other_epub
|
|
server.books = [
|
|
FakeBook(1, "uuid-1", "Test Story", publisher="www.royalroad.com", epub=other),
|
|
FakeBook(2, "uuid-2", "Test Story", publisher="www.royalroad.com", epub=story_epub),
|
|
FakeBook(3, "uuid-3", "Another Book", ["Someone"], publisher="Tor", epub=b"not a zip"),
|
|
FakeBook(4, "uuid-4", "Test Story Two", publisher="www.royalroad.com"),
|
|
]
|
|
yield server
|
|
server.close()
|
|
|
|
|
|
def client(fake, password="secret"):
|
|
return CwaClient(fake.url, "bindery", password)
|
|
|
|
|
|
def test_search_reads_ids_and_uuids(fake):
|
|
books = client(fake).search("test story")
|
|
# Book 4 has no files, so OPDS shows no download link and no book id for it.
|
|
assert [(b.id, b.uuid) for b in books] == [(1, "uuid-1"), (2, "uuid-2")]
|
|
assert books[0].formats == {"epub": "/opds/download/1/epub/"}
|
|
assert books[0].publisher == "www.royalroad.com"
|
|
|
|
|
|
def test_wrong_password_and_missing_address(fake):
|
|
with pytest.raises(CwaError, match="refused"):
|
|
client(fake, "wrong").search("x")
|
|
with pytest.raises(CwaError, match="not set"):
|
|
CwaClient("").search("x")
|
|
|
|
|
|
def test_cwa_that_needs_no_login_works_without_an_account(fake):
|
|
fake.anonymous = True
|
|
open_client = CwaClient(fake.url)
|
|
assert [b.id for b in open_client.search("Another")] == [3]
|
|
assert open_client.get_book("uuid-2")["id"] == 2
|
|
fake.anonymous = False
|
|
with pytest.raises(CwaError, match="CWA asks for a login. Set CWA_USERNAME and CWA_PASSWORD."):
|
|
open_client.search("Test Story")
|
|
|
|
|
|
def test_a_login_check_does_not_use_the_session_of_the_account(fake, monkeypatch):
|
|
c = client(fake)
|
|
monkeypatch.setattr(c.session, "get", Mock(side_effect=AssertionError("The login check used the session")))
|
|
assert c.login_ok("bindery", "secret")
|
|
assert not c.login_ok("bindery", "wrong")
|
|
|
|
|
|
def test_the_test_of_the_logins_of_cwa_is_kept_for_a_time(fake, monkeypatch):
|
|
c = client(fake)
|
|
assert not c.accepts_any_login()
|
|
fake.anonymous = True
|
|
assert not c.accepts_any_login() # the last test still counts
|
|
assert len([r for r in fake.requests if r == "/opds"]) == 1
|
|
clock = time.monotonic()
|
|
monkeypatch.setattr(time, "monotonic", lambda: clock + LOGIN_TEST_SECONDS + 1)
|
|
assert c.accepts_any_login()
|
|
assert CwaClient(fake.url).accepts_any_login()
|
|
|
|
|
|
def test_get_book(fake):
|
|
c = client(fake)
|
|
assert c.get_book("uuid-2") == {
|
|
"id": 2,
|
|
"uuid": "uuid-2",
|
|
"title": "Test Story",
|
|
"authors": ["Writer"],
|
|
"formats": ["EPUB"],
|
|
}
|
|
assert c.get_book("nope") is None
|
|
|
|
|
|
def test_get_book_can_wait_for_a_shorter_time(fake, monkeypatch):
|
|
c = client(fake)
|
|
get = c.session.get
|
|
timeouts = []
|
|
monkeypatch.setattr(c.session, "get", lambda url, **kw: timeouts.append(kw["timeout"]) or get(url, **kw))
|
|
c.get_book("uuid-2", timeout=5)
|
|
c.get_book("uuid-2")
|
|
c.search("Story")
|
|
assert timeouts == [5, 60, 60]
|
|
|
|
|
|
@pytest.mark.parametrize("book_uuid", ["../opds", "uuid-2?x=1", "uuid-2#x", "uuid%2F2", ""])
|
|
def test_get_book_refuses_a_uuid_that_changes_the_path(book_uuid):
|
|
# Nothing listens on this port, so a request would raise CwaError.
|
|
assert CwaClient("http://127.0.0.1:9").get_book(book_uuid) is None
|
|
|
|
|
|
def test_site_books(fake):
|
|
books = client(fake).site_books(fff.is_supported_site)
|
|
assert [b.id for b in books] == [1, 2] # book 4 has no epub, book 3 is not from a story site
|
|
assert any("offset=2" in r for r in fake.requests) # the publisher feed has three pages
|
|
|
|
|
|
def test_site_books_names_the_cwa_setting_that_shows_publishers(fake, monkeypatch):
|
|
monkeypatch.setattr(fake, "publisher_view", False)
|
|
with pytest.raises(CwaError, match='Turn on "Show Publisher Section" for Bindery in CWA'):
|
|
client(fake).site_books(fff.is_supported_site)
|
|
|
|
|
|
def test_finder_checks_the_url_in_each_epub(fake, tmp_path):
|
|
finder = BookFinder()
|
|
found = finder.find(client(fake), STORY_URL, ["test story"], fff.normalize_url, tmp_path)
|
|
assert found is not None
|
|
book, path = found
|
|
assert book.id == 2 and path.exists()
|
|
# With book 2 excluded, book 1 is checked once and remembered as another story.
|
|
assert finder.find(client(fake), STORY_URL, ["Test Story"], fff.normalize_url, tmp_path, {2}) is None
|
|
fake.requests.clear()
|
|
assert finder.find(client(fake), STORY_URL, ["Test Story"], fff.normalize_url, tmp_path, {2}) is None
|
|
assert not [r for r in fake.requests if r.startswith("/opds/download/")]
|
|
assert sorted(p.name for p in tmp_path.iterdir()) == [path.name] # rejected downloads are deleted
|
|
|
|
|
|
TAGGED = "Story A [Tag A, Tag B] (Book 1 Complete)"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"title, plain, search",
|
|
[
|
|
(TAGGED, "story a", "Story A"),
|
|
("Story A - (Book 1)", "story a", "Story A"),
|
|
("[AU] Story A [Tag] Part I", "story a part i", "story a part i"),
|
|
("Story (A [B]) C", "story c", "Story"),
|
|
("Plain Title", "plain title", "Plain Title"),
|
|
],
|
|
)
|
|
def test_titles_without_their_bracketed_parts(title, plain, search):
|
|
assert (plain_title(title), search_text(title)) == (plain, search)
|
|
|
|
|
|
def test_title_rank_prefers_the_closest_title():
|
|
titles = [
|
|
"Story A [Tag A, Tag B] Part I",
|
|
"Story A",
|
|
"Story A [Tag A, Tag B]",
|
|
TAGGED,
|
|
"Story Ab",
|
|
"Another Story A",
|
|
]
|
|
ranked = sorted((t for t in titles if title_rank(t, [TAGGED])), key=lambda t: title_rank(t, [TAGGED]))
|
|
assert ranked == [TAGGED, "Story A [Tag A, Tag B]", "Story A", "Story A [Tag A, Tag B] Part I"]
|
|
assert title_rank("Story A [Tag A, Tag B]", ["Story A"])[0] == 1
|
|
assert title_rank("Story A Part I", ["Story A Part II"]) is None
|
|
|
|
|
|
def test_finder_downloads_the_closest_titles_first(fake, other_epub, tmp_path):
|
|
fake.books = [
|
|
FakeBook(9, "uuid-9", TAGGED, epub=other_epub),
|
|
FakeBook(10, "uuid-10", "Story A", epub=other_epub),
|
|
FakeBook(11, "uuid-11", "Story A [Tag A, Tag B] Part I", epub=other_epub),
|
|
FakeBook(12, "uuid-12", "Story A [Tag A, Tag B]", epub=other_epub),
|
|
FakeBook(13, "uuid-13", "Story Ab", epub=other_epub),
|
|
FakeBook(14, "uuid-14", "Another Story A", epub=other_epub),
|
|
]
|
|
assert BookFinder().find(client(fake), STORY_URL, [TAGGED], fff.normalize_url, tmp_path) is None
|
|
assert [r for r in fake.requests if r.startswith("/opds/search")] == ["/opds/search?query=Story+A"]
|
|
downloads = [int(r.split("/")[3]) for r in fake.requests if r.startswith("/opds/download/")]
|
|
assert downloads == [9, 12, 10, 11]
|
|
|
|
|
|
def test_login_ok(fake):
|
|
c = CwaClient(fake.url)
|
|
assert c.login_ok("bindery", "secret")
|
|
assert not c.login_ok("bindery", "wrong")
|
|
fake.close()
|
|
with pytest.raises(CwaError, match="Cannot reach"):
|
|
c.login_ok("bindery", "secret")
|
|
|
|
|
|
def test_epub_reader(story_epub, tmp_path):
|
|
path = tmp_path / "s.epub"
|
|
path.write_bytes(story_epub)
|
|
assert STORY_URL in [fff.normalize_url(u)[0] for u in epub_urls(path)]
|
|
meta = epub_metadata(path)
|
|
assert meta["title"] == "Test Story"
|
|
assert meta["authors"] == ["Writer"]
|
|
assert meta["series"] == "Saga" and meta["series_index"] == 2.0
|
|
assert meta["pubdate"].startswith("2024-01-02")
|
|
|
|
|
|
def test_ingest_uses_part_names_and_unique_names(tmp_path):
|
|
folder = IngestFolder(tmp_path / "ingest")
|
|
folder.path.mkdir()
|
|
epub = tmp_path / "x.epub"
|
|
epub.write_bytes(b"PK")
|
|
first = folder.ingest(epub, 'A: "Story" / Name')
|
|
second = folder.ingest(epub, 'A: "Story" / Name')
|
|
assert first != second
|
|
assert first.name.startswith("A_ _Story_ _ Name ") and first.suffix == ".epub"
|
|
assert sorted(p.name for p in folder.path.iterdir()) == sorted([first.name, second.name])
|
|
|
|
|
|
def test_send_update_writes_sidecar(tmp_path):
|
|
folder = IngestFolder(tmp_path / "ingest")
|
|
folder.path.mkdir()
|
|
epub = tmp_path / "x.epub"
|
|
epub.write_bytes(b"PK")
|
|
path = folder.send_update(epub, 7, "T" * 300)
|
|
assert sorted(p.name for p in folder.path.iterdir()) == [path.name, path.name + ".cwa.json"]
|
|
assert len(path.name) <= MAX_NAME_LENGTH + len(".epub")
|
|
manifest = json.loads((folder.path / (path.name + ".cwa.json")).read_text())
|
|
assert manifest == {"action": "add_format", "book_id": 7}
|
|
assert path.read_bytes() == b"PK"
|
|
|
|
|
|
def test_a_missing_ingest_folder_is_an_error_and_is_not_made(tmp_path, story_epub):
|
|
source = tmp_path / "book.epub"
|
|
source.write_bytes(story_epub)
|
|
folder = IngestFolder(tmp_path / "missing" / "ingest")
|
|
for send in (lambda: folder.ingest(source, "Title"), lambda: folder.send_update(source, 5, "Title")):
|
|
with pytest.raises(IngestError, match="does not exist. Check INGEST_DIR and the volume mount."):
|
|
send()
|
|
assert not (tmp_path / "missing").exists()
|