tim
9a48356155
The search through CWA gave no results for books that the full list shows. The search now filters the list that "Show all stories in the library" loads, by title, author and site. The buttons stay disabled until the answer comes. A book that the user imports leaves the list.
620 lines
27 KiB
Python
620 lines
27 KiB
Python
"""Rebuild stored stories, import epub files and hand books to CWA, against a fake CWA. No story site access."""
|
|
|
|
import dataclasses
|
|
import json
|
|
import re
|
|
import zipfile
|
|
|
|
import pytest
|
|
from fake_cwa import FakeBook, FakeCwa
|
|
from test_build import COVER, IMG_URL, PNG, URL, chapters, metahtml
|
|
|
|
from bindery import config, fff, orchestrator
|
|
from bindery.diffing import text_hash
|
|
from bindery.epub import EpubError
|
|
from bindery.fff import SiteSnapshot, StoryError
|
|
from bindery.jobs import DownloadMode, Job, JobQueue, QueueStatus
|
|
from bindery.store import Store
|
|
|
|
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, [], {"title": "Library Title"}, path)
|
|
return path.read_bytes()
|
|
|
|
|
|
@pytest.fixture
|
|
def fake(story_epub, monkeypatch, tmp_path):
|
|
server = FakeCwa()
|
|
server.books = [FakeBook(1, "uuid-1", "Test Story", publisher="www.royalroad.com", epub=story_epub)]
|
|
monkeypatch.setattr(config, "CWA_URL", server.url)
|
|
monkeypatch.setattr(config, "CWA_USERNAME", "bindery")
|
|
monkeypatch.setattr(config, "CWA_PASSWORD", "secret")
|
|
monkeypatch.setattr(orchestrator, "store", Store(tmp_path / "bindery.db"))
|
|
monkeypatch.setattr(config, "INGEST_DIR", tmp_path / "ingest")
|
|
config.INGEST_DIR.mkdir()
|
|
monkeypatch.setattr(orchestrator, "finder", orchestrator.BookFinder())
|
|
yield server
|
|
server.close()
|
|
|
|
|
|
def add_story(**changes):
|
|
store = orchestrator.store
|
|
story_id = store.create_story(STORY_URL, "www.royalroad.com", "Test Story", "Writer", ["title"])
|
|
store.update_story(story_id, metahtml=metahtml(), pinned={"title": "My Title"}, **changes)
|
|
for n, c in enumerate(chapters()):
|
|
store.add_chapter(story_id, c["url"], c["title"], c["html"], "h", "site", n)
|
|
return story_id
|
|
|
|
|
|
def ingest_files():
|
|
folder = config.INGEST_DIR
|
|
return sorted(p.name for p in folder.iterdir()) if folder.exists() else []
|
|
|
|
|
|
def opf(name):
|
|
with zipfile.ZipFile(config.INGEST_DIR / name) as z:
|
|
return z.read("content.opf").decode()
|
|
|
|
|
|
def rebuild(story_id):
|
|
"""A rebuild that the user asks for. It builds and sends the book also when the book did not change."""
|
|
return orchestrator._run_rebuild(Job(mode=DownloadMode.REBUILD, story_id=story_id, options={"force": True}))
|
|
|
|
|
|
def test_first_version_goes_in_as_a_new_book(fake):
|
|
fake.books = []
|
|
story_id = add_story()
|
|
assert rebuild(story_id) == "Sent to CWA"
|
|
(name,) = ingest_files()
|
|
assert re.search(r"<dc:title[^>]*>My Title</dc:title>", opf(name))
|
|
assert orchestrator.store.get_story(story_id)["ingested_at"]
|
|
|
|
|
|
def test_ingested_story_is_found_and_updated_with_a_sidecar(fake):
|
|
story_id = add_story(ingested_at="2026-01-01T00:00:00")
|
|
assert "library book 1" in rebuild(story_id)
|
|
story = orchestrator.store.get_story(story_id)
|
|
assert (story["calibre_id"], story["calibre_uuid"]) == (1, "uuid-1")
|
|
epub, manifest = ingest_files()
|
|
assert manifest == epub + ".cwa.json"
|
|
assert json.loads((config.INGEST_DIR / manifest).read_text()) == {
|
|
"action": "add_format",
|
|
"book_id": 1,
|
|
}
|
|
|
|
|
|
def test_update_fails_while_cwa_has_no_such_book(fake):
|
|
fake.books = []
|
|
story_id = add_story(ingested_at="2026-01-01T00:00:00")
|
|
with pytest.raises(orchestrator.StoryError, match="Cannot find this story in CWA"):
|
|
rebuild(story_id)
|
|
assert ingest_files() == [] # a second plain epub would make a duplicate book
|
|
|
|
|
|
def test_update_exports_without_a_library_connection(fake, monkeypatch):
|
|
monkeypatch.setattr(config, "CWA_URL", "")
|
|
story_id = add_story(ingested_at="2026-01-01T00:00:00")
|
|
assert rebuild(story_id) == "Sent to ingest folder"
|
|
assert orchestrator.store.get_story(story_id)["status"] == "ok"
|
|
assert orchestrator.store.get_setting(story_id, "upload_held") is None
|
|
assert len(ingest_files()) == 1
|
|
|
|
|
|
def test_linked_story_still_updates_when_cwa_is_down(fake):
|
|
story_id = add_story(ingested_at="2026-01-01T00:00:00", calibre_id=1, calibre_uuid="uuid-1")
|
|
fake.close()
|
|
assert "library book 1" in rebuild(story_id)
|
|
assert len(ingest_files()) == 2
|
|
|
|
|
|
def test_deleted_book_is_unlinked_and_the_update_fails(fake):
|
|
story_id = add_story(ingested_at="2026-01-01T00:00:00", calibre_id=5, calibre_uuid="uuid-gone")
|
|
fake.books = []
|
|
with pytest.raises(orchestrator.StoryError, match="Cannot find this story in CWA"):
|
|
rebuild(story_id)
|
|
assert orchestrator.store.get_story(story_id)["calibre_id"] is None
|
|
|
|
|
|
def test_unlinked_story_can_go_to_cwa_again_as_a_new_book(fake):
|
|
fake.books = []
|
|
story_id = add_story(ingested_at="2026-01-01T00:00:00", calibre_id=5, calibre_uuid="uuid-gone")
|
|
orchestrator.link_story(story_id, None)
|
|
assert rebuild(story_id) == "Sent to CWA"
|
|
assert len(ingest_files()) == 1
|
|
|
|
|
|
def test_import_from_the_library(fake):
|
|
job = Job(mode=DownloadMode.IMPORT, calibre_id=1, title="Test Story", options={"uuid": "uuid-1", "check": False})
|
|
assert orchestrator._run_import(job) == "Imported 3 chapters"
|
|
story = orchestrator.store.get_story(job.story_id)
|
|
assert story["url"] == STORY_URL
|
|
assert (story["calibre_id"], story["calibre_uuid"]) == (1, "uuid-1")
|
|
assert story["pinned"] == {"title": "Library Title"}
|
|
assert [c["title"] for c in orchestrator.store.list_chapters(job.story_id)] == ["One", "Two", "Three"]
|
|
assert orchestrator._run_import(job) == "Book already tracked"
|
|
assert not list(config.TMP_DIR.glob("import-*"))
|
|
|
|
|
|
def test_library_candidates_and_search(fake):
|
|
candidates = orchestrator.library_candidates()
|
|
assert candidates == [
|
|
{
|
|
"id": 1,
|
|
"uuid": "uuid-1",
|
|
"title": "Test Story",
|
|
"authors": ["Writer"],
|
|
"publisher": "www.royalroad.com",
|
|
"formats": ["epub"],
|
|
}
|
|
]
|
|
story_id = add_story()
|
|
orchestrator.link_story(story_id, "uuid-1")
|
|
assert orchestrator.library_candidates() == []
|
|
assert orchestrator.search_library("test")[0]["tracked_as"] == "Test Story"
|
|
with pytest.raises(KeyError):
|
|
orchestrator.link_story(story_id, "uuid-nope")
|
|
|
|
|
|
# Import chapters from an epub file
|
|
|
|
|
|
@pytest.fixture
|
|
def jobs(monkeypatch):
|
|
queue = JobQueue()
|
|
monkeypatch.setattr(orchestrator, "job_queue", queue)
|
|
return queue
|
|
|
|
|
|
def upload(tmp_path, data):
|
|
path = tmp_path / "upload-1.epub"
|
|
path.write_bytes(data)
|
|
return path
|
|
|
|
|
|
def import_file(jobs, path, story_id=None, check=True):
|
|
queued = orchestrator.queue_epub_file(path, "story.epub", story_id, check)
|
|
job = jobs.get_next()
|
|
assert job.id == queued["id"]
|
|
return orchestrator._run_import_file(job)
|
|
|
|
|
|
def titles(story_id):
|
|
return [c["title"] for c in orchestrator.store.list_chapters(story_id)]
|
|
|
|
|
|
def test_epub_file_starts_a_new_story(fake, jobs, story_epub, tmp_path):
|
|
path = upload(tmp_path, story_epub)
|
|
assert import_file(jobs, path) == "Imported 3 chapters from story.epub"
|
|
assert not path.exists()
|
|
story = orchestrator.store.get_story_by_url(STORY_URL)
|
|
assert story["title"] == "Library Title"
|
|
assert story["pinned"] == {"title": "Library Title"}
|
|
assert titles(story["id"]) == ["One", "Two", "Three"]
|
|
check = jobs.get_next()
|
|
assert (check.mode, check.story_id, check.options) == (DownloadMode.NEW, story["id"], {"build": True})
|
|
|
|
|
|
def test_epub_file_fills_in_a_stubbed_story(fake, jobs, story_epub, tmp_path):
|
|
store = orchestrator.store
|
|
story_id = store.create_story(STORY_URL, "www.royalroad.com", "Test Story", "Writer", ["title"])
|
|
one, _, three = chapters()
|
|
store.add_chapter(story_id, one["url"], "One", "<p>Kept</p>", "h", "site", 0)
|
|
store.add_chapter(story_id, three["url"], "Three", "<p>Kept</p>", "h", "site", 1)
|
|
assert import_file(jobs, upload(tmp_path, story_epub), story_id) == "Imported 1 chapter from story.epub"
|
|
assert titles(story_id) == ["One", "Two", "Three"]
|
|
assert store.chapters_for_build(story_id)[0]["html"] == "<p>Kept</p>"
|
|
|
|
|
|
def test_next_update_adds_newer_chapters_after_the_imported_ones(fake, jobs, story_epub, tmp_path, monkeypatch):
|
|
# The site stubbed the first two chapters and added a fourth after the epub was made.
|
|
fake.books = []
|
|
import_file(jobs, upload(tmp_path, story_epub))
|
|
four = chapters()[2]["url"].replace("15/c15", "16/c16")
|
|
site = [{"url": chapters()[2]["url"], "title": "Three"}, {"url": four, "title": "Four"}]
|
|
monkeypatch.setattr(
|
|
fff,
|
|
"fetch_story",
|
|
lambda adapter: SiteSnapshot(STORY_URL, "www.royalroad.com", "Test Story", "Writer", metahtml(), site, {}),
|
|
)
|
|
monkeypatch.setattr(fff, "fetch_chapter", lambda adapter, url, index: "<div><p>Four</p></div>")
|
|
monkeypatch.setattr(fff, "collected_images", lambda adapter: [])
|
|
check = jobs.get_next()
|
|
assert orchestrator._run_site_job(check).startswith("1 new chapter")
|
|
assert titles(check.story_id) == ["One", "Two", "Three", "Four"]
|
|
assert len(ingest_files()) == 1
|
|
|
|
|
|
def test_epub_file_without_a_check_rebuilds_a_downloaded_story(fake, jobs, story_epub, tmp_path):
|
|
store = orchestrator.store
|
|
story_id = store.create_story(STORY_URL, "www.royalroad.com", "Test Story", "Writer", ["title"])
|
|
store.update_story(story_id, metahtml=metahtml(), ingested_at="2026-01-01T00:00:00")
|
|
store.add_chapter(story_id, chapters()[0]["url"], "One", "<p>Kept</p>", "h", "site", 0)
|
|
assert import_file(jobs, upload(tmp_path, story_epub), story_id, check=False).startswith("Imported 2")
|
|
rebuild_job = jobs.get_next()
|
|
assert (rebuild_job.mode, rebuild_job.story_id) == (DownloadMode.REBUILD, story_id)
|
|
assert "library book 1" in orchestrator._run_rebuild(rebuild_job)
|
|
|
|
|
|
def test_epub_of_another_story_is_refused(fake, jobs, story_epub, tmp_path):
|
|
other = orchestrator.store.create_story(
|
|
"https://www.royalroad.com/fiction/99", "www.royalroad.com", "Other", "Writer", []
|
|
)
|
|
path = upload(tmp_path, story_epub)
|
|
with pytest.raises(StoryError, match="story.epub is an epub of another story"):
|
|
orchestrator.queue_epub_file(path, "story.epub", other)
|
|
assert not path.exists()
|
|
assert jobs.get_next() is None
|
|
|
|
|
|
def test_file_that_is_not_an_epub_is_refused(fake, jobs, tmp_path):
|
|
path = upload(tmp_path, b"not a zip file")
|
|
with pytest.raises(EpubError, match="story.epub is not a readable epub"):
|
|
orchestrator.queue_epub_file(path, "story.epub")
|
|
assert not path.exists()
|
|
|
|
|
|
# Updates from the site
|
|
|
|
CHAPTER = "https://www.royalroad.com/fiction/1234/test-story/chapter/{0}/c{0}"
|
|
|
|
|
|
def linked_story(chapter_list):
|
|
"""A story in CWA with the chapters (number, title, text)."""
|
|
store = orchestrator.store
|
|
story_id = store.create_story(STORY_URL, "www.royalroad.com", "Test Story", "Writer", ["title"], 1, "uuid-1")
|
|
store.update_story(story_id, metahtml=metahtml(), ingested_at="2026-01-01T00:00:00")
|
|
for n, (number, title, text, *source) in enumerate(chapter_list):
|
|
html = f"<div><p>{text}</p></div>"
|
|
store.add_chapter(
|
|
story_id, CHAPTER.format(number), title, html, text_hash(html), source[0] if source else "new", n
|
|
)
|
|
return story_id
|
|
|
|
|
|
def site(monkeypatch, chapter_list, broken=()):
|
|
"""The site lists these chapters (number, title, text). Chapters in `broken` fail to download."""
|
|
texts = {CHAPTER.format(n): text for n, _, text in chapter_list}
|
|
listed = [{"url": CHAPTER.format(n), "title": title} for n, title, _ in chapter_list]
|
|
monkeypatch.setattr(
|
|
fff,
|
|
"fetch_story",
|
|
lambda adapter: SiteSnapshot(STORY_URL, "www.royalroad.com", "Test Story", "Writer", metahtml(), listed, {}),
|
|
)
|
|
monkeypatch.setattr(fff, "collected_images", lambda adapter: [])
|
|
|
|
def fetch(adapter, url, index):
|
|
if url in {CHAPTER.format(n) for n in broken}:
|
|
raise OSError("timeout")
|
|
return f"<div><p>{texts[url]}</p></div>"
|
|
|
|
monkeypatch.setattr(fff, "fetch_chapter", fetch)
|
|
|
|
|
|
def update(story_id, mode=DownloadMode.NEW, **fields):
|
|
return orchestrator._run_site_job(Job(mode=mode, story_id=story_id, **fields))
|
|
|
|
|
|
def book(story_id):
|
|
"""The title and text of each chapter in the book."""
|
|
return [(c["title"], re.sub(r"<[^>]+>", "", c["html"])) for c in orchestrator.store.chapters_for_build(story_id)]
|
|
|
|
|
|
def test_the_site_cover_goes_in_the_book(fake, jobs, monkeypatch, tmp_path):
|
|
# A personal.ini without include_images, as the calibre plugin often has.
|
|
empty = tmp_path / "personal.ini"
|
|
empty.write_text("")
|
|
monkeypatch.setattr(config, "PERSONAL_INI", empty)
|
|
story_id = linked_story([(1, "One", "a")])
|
|
site(monkeypatch, [(1, "One", "a"), (2, "Two", "b")])
|
|
snapshot = fff.fetch_story
|
|
monkeypatch.setattr(fff, "fetch_story", lambda adapter: dataclasses.replace(snapshot(adapter), cover=COVER))
|
|
assert update(story_id).startswith("1 new chapter")
|
|
assert orchestrator.store.cover(story_id)["data"] == PNG
|
|
(epub,) = [n for n in ingest_files() if n.endswith(".epub")]
|
|
with zipfile.ZipFile(config.INGEST_DIR / epub) as z:
|
|
assert "OEBPS/cover.xhtml" in z.namelist()
|
|
|
|
|
|
def test_new_chapter_after_stubbed_chapters_goes_last(fake, jobs, monkeypatch):
|
|
story_id = linked_story([(1, "One", "a"), (2, "Two", "b"), (3, "Three", "c")])
|
|
site(monkeypatch, [(1, "One", "a"), (4, "Four", "d")])
|
|
assert update(story_id).startswith("1 new chapter")
|
|
assert book(story_id) == [("One", "a"), ("Two", "b"), ("Three", "c"), ("Four", "d")]
|
|
|
|
|
|
def test_renamed_chapter_is_rechecked_and_rebuilt(fake, jobs, monkeypatch):
|
|
story_id = linked_story([(1, "One", "a"), (2, "Two", "b")])
|
|
site(monkeypatch, [(1, "One", "a"), (2, "Two (rewritten)", "b2")])
|
|
message = update(story_id)
|
|
assert "1 renamed, 1 of 1 rechecked chapters changed" in message and "library book 1" in message
|
|
assert book(story_id) == [("One", "a"), ("Two (rewritten)", "b2")]
|
|
|
|
|
|
def test_renamed_chapter_with_the_same_text_is_rebuilt(fake, jobs, monkeypatch):
|
|
story_id = linked_story([(1, "One", "a")])
|
|
site(monkeypatch, [(1, "Chapter One", "a")])
|
|
message = update(story_id)
|
|
assert "1 renamed, 0 of 1 rechecked chapters changed" in message and "library book 1" in message
|
|
assert len(orchestrator.store.list_revisions(orchestrator.store.list_chapters(story_id)[0]["id"])) == 1
|
|
|
|
|
|
def test_job_results_show_chapter_titles_as_text(fake, jobs, monkeypatch):
|
|
story_id = linked_story([(1, "One", "a")])
|
|
site(monkeypatch, [(1, "One", "a"), (2, "Folly & Magic", "b")], broken=[2])
|
|
job = Job(mode=DownloadMode.NEW, story_id=story_id)
|
|
orchestrator._run_site_job(job)
|
|
assert job.result["failed"] == ["Folly & Magic: timeout"]
|
|
site(monkeypatch, [(1, "One", "a"), (2, "Folly & Magic", "b")])
|
|
orchestrator._run_site_job(Job(mode=DownloadMode.NEW, story_id=story_id))
|
|
# The store keeps FanFicFare's form, for the epub writer.
|
|
assert [c["title"] for c in orchestrator.store.list_chapters(story_id)] == ["One", "Folly & Magic"]
|
|
|
|
|
|
def test_renamed_chapter_keeps_its_title_until_its_recheck_works(fake, jobs, monkeypatch):
|
|
story_id = linked_story([(1, "One", "a")])
|
|
site(monkeypatch, [(1, "Chapter One", "a2")], broken=[1])
|
|
update(story_id)
|
|
assert book(story_id) == [("One", "a")]
|
|
site(monkeypatch, [(1, "Chapter One", "a2")])
|
|
update(story_id)
|
|
assert book(story_id) == [("Chapter One", "a2")]
|
|
|
|
|
|
def test_insert_on_a_site_that_numbers_its_chapter_urls(fake, jobs, monkeypatch):
|
|
# FanFiction.net and similar sites. A chapter inserted at 2 moves B and C up one URL.
|
|
story_id = linked_story([(1, "1. A", "a"), (2, "2. B", "b"), (3, "3. C", "c")])
|
|
site(monkeypatch, [(1, "1. A", "a"), (2, "2. New", "new"), (3, "3. B", "b"), (4, "4. C", "c")])
|
|
update(story_id)
|
|
assert book(story_id) == [("1. A", "a"), ("2. New", "new"), ("3. B", "b"), ("4. C", "c")]
|
|
|
|
|
|
def test_imported_chapter_that_differs_only_by_the_rules_is_unchanged(fake, jobs, monkeypatch):
|
|
orchestrator.save_rules(None, [{"pattern": "Zorian", "replacement": "Z"}])
|
|
story_id = linked_story([(1, "One", "Z wakes.", "import")])
|
|
site(monkeypatch, [(1, "One", "Zorian wakes.")])
|
|
assert "0 of 1 rechecked chapters changed" in update(story_id, DownloadMode.RECHECK_ALL)
|
|
# The stored text is now the site text, so the rules run on it once.
|
|
(chapter,) = orchestrator.store.list_chapters(story_id)
|
|
assert orchestrator.store.revision_html(chapter["id"], 1) == "<div><p>Zorian wakes.</p></div>"
|
|
# A real change is still a change.
|
|
site(monkeypatch, [(1, "One", "Zorian sleeps.")])
|
|
assert "1 of 1 rechecked chapters changed" in update(story_id, DownloadMode.RECHECK_ALL)
|
|
|
|
|
|
def test_a_chosen_revision_stays_until_the_site_changes_again(fake, jobs, monkeypatch):
|
|
story_id = linked_story([(1, "One", "a")])
|
|
(chapter,) = orchestrator.store.list_chapters(story_id)
|
|
site(monkeypatch, [(1, "One", "b")])
|
|
assert "1 of 1 rechecked chapters changed" in update(story_id, DownloadMode.RECHECK_ALL)
|
|
orchestrator.store.update_chapter(chapter["id"], current_rev=1)
|
|
assert "0 of 1 rechecked chapters changed" in update(story_id, DownloadMode.RECHECK_ALL)
|
|
assert book(story_id) == [("One", "a")]
|
|
assert len(orchestrator.store.list_revisions(chapter["id"])) == 2
|
|
site(monkeypatch, [(1, "One", "c")])
|
|
assert "1 of 1 rechecked chapters changed" in update(story_id, DownloadMode.RECHECK_ALL)
|
|
assert book(story_id) == [("One", "c")]
|
|
|
|
|
|
def test_a_site_that_goes_back_to_an_older_text_changes_the_book(fake, jobs, monkeypatch):
|
|
story_id = linked_story([(1, "One", "a")])
|
|
site(monkeypatch, [(1, "One", "b")])
|
|
update(story_id, DownloadMode.RECHECK_ALL)
|
|
site(monkeypatch, [(1, "One", "a")])
|
|
assert "1 of 1 rechecked chapters changed" in update(story_id, DownloadMode.RECHECK_ALL)
|
|
assert book(story_id) == [("One", "a")]
|
|
|
|
|
|
def test_new_chapter_download_rechecks_the_latest_chapters(fake, jobs, monkeypatch):
|
|
orchestrator.store.set_setting(None, "recent_chapters", 2)
|
|
story_id = linked_story([(1, "One", "a"), (2, "Two", "b"), (3, "Three", "c")])
|
|
site(monkeypatch, [(1, "One", "a (edited)"), (2, "Two", "b"), (3, "Three", "c (edited)")])
|
|
assert "1 of 2 rechecked chapters changed" in update(story_id)
|
|
assert book(story_id) == [("One", "a"), ("Two", "b"), ("Three", "c (edited)")]
|
|
# A story with its own full recheck interval does not check its latest chapters.
|
|
orchestrator.update_story_settings(story_id, recheck_days=7)
|
|
site(monkeypatch, [(1, "One", "a"), (2, "Two", "b"), (3, "Three", "c (again)")])
|
|
assert update(story_id).startswith("No changes")
|
|
|
|
|
|
def test_scheduler_runs_a_full_recheck_when_one_is_due(fake, jobs, monkeypatch):
|
|
monkeypatch.setattr(config, "UPDATE_INTERVAL_HOURS", 12)
|
|
story_id = linked_story([(1, "One", "a")])
|
|
orchestrator.update_story_settings(story_id, recheck_days=7)
|
|
orchestrator._queue_due_updates()
|
|
job = jobs.get_next()
|
|
assert job.mode == DownloadMode.RECHECK_ALL
|
|
site(monkeypatch, [(1, "One", "a")])
|
|
orchestrator._run_site_job(job)
|
|
jobs.update(job, status=QueueStatus.DONE)
|
|
assert orchestrator.story_details(story_id)["story"]["last_full_recheck"]
|
|
orchestrator.store.update_story(story_id, last_checked="2020-01-01T00:00:00")
|
|
orchestrator._queue_due_updates()
|
|
assert jobs.get_next().mode == DownloadMode.NEW
|
|
with pytest.raises(ValueError):
|
|
orchestrator.update_story_settings(story_id, recheck_days=-1)
|
|
orchestrator.update_story_settings(story_id, recheck_days=0)
|
|
assert orchestrator.story_details(story_id)["story"]["recheck_days"] is None
|
|
|
|
|
|
# Metadata edits
|
|
|
|
|
|
def test_metadata_edit_rebuilds_the_book(fake, jobs):
|
|
story_id = add_story(calibre_id=1, calibre_uuid="uuid-1", ingested_at="2026-01-01T00:00:00")
|
|
result = orchestrator.edit_metadata(story_id, {"title": "Edited", "tags": "Mine, Yours"})
|
|
assert result["story"]["title"] == "Edited"
|
|
assert result["story"]["pinned"] == {"title": "Edited", "tags": ["Mine", "Yours"]}
|
|
job = jobs.get_next()
|
|
assert result["job"]["id"] == job.id and job.mode == DownloadMode.REBUILD
|
|
orchestrator._run_rebuild(job)
|
|
epub, _ = ingest_files()
|
|
assert re.search(r"<dc:title[^>]*>Edited</dc:title>", opf(epub))
|
|
assert re.findall(r"<dc:subject>([^<]*)</dc:subject>", opf(epub)) == ["Mine", "Yours"]
|
|
# Without the edit, the unlocked tags come from the site again.
|
|
orchestrator.edit_metadata(story_id, {"tags": None})
|
|
assert orchestrator.store.get_story(story_id)["pinned"] == {"title": "Edited"}
|
|
|
|
|
|
def test_unlocking_a_field_drops_its_kept_value(fake, jobs):
|
|
story_id = add_story()
|
|
orchestrator.update_story_settings(story_id, locked_fields=[])
|
|
assert orchestrator.store.get_story(story_id)["pinned"] == {}
|
|
|
|
|
|
def test_metadata_edit_of_a_story_that_cwa_does_not_have_waits_for_the_next_build(fake, jobs):
|
|
story_id = add_story()
|
|
assert orchestrator.edit_metadata(story_id, {"series": "Saga [2]"})["job"] is None
|
|
assert jobs.get_next() is None
|
|
with pytest.raises(ValueError, match="Unknown metadata fields"):
|
|
orchestrator.edit_metadata(story_id, {"publisher": "x"})
|
|
|
|
|
|
# Restarts
|
|
|
|
|
|
class Crash(BaseException):
|
|
"""Bindery stops in the middle of a job."""
|
|
|
|
|
|
def test_a_download_that_stops_keeps_its_chapters_and_resumes(fake, story_epub, tmp_path, monkeypatch):
|
|
store = orchestrator.store
|
|
story_id = store.create_story(STORY_URL, "www.royalroad.com", "Test Story", "Writer", ["title"])
|
|
store.update_story(story_id, metahtml=metahtml(), ingested_at="2026-01-01T00:00:00")
|
|
one, two, three = chapters()
|
|
store.add_chapter(story_id, one["url"], "One", "<p>1</p>", "h", "site", 0)
|
|
four = three["url"].replace("15/c15", "16/c16")
|
|
site = [one, two, three, {"url": four, "title": "Four"}]
|
|
image = {"url": IMG_URL, "ext": "png", "mime": "image/png", "data": PNG}
|
|
monkeypatch.setattr(
|
|
fff,
|
|
"fetch_story",
|
|
lambda adapter: SiteSnapshot(STORY_URL, "www.royalroad.com", "Test Story", "Writer", metahtml(), site, {}),
|
|
)
|
|
monkeypatch.setattr(fff, "collected_images", lambda adapter: [image])
|
|
crash_at = [three["url"]]
|
|
|
|
def fetch(adapter, url, index):
|
|
if url in crash_at:
|
|
raise Crash
|
|
return f"<div><p>{url}</p></div>"
|
|
|
|
monkeypatch.setattr(fff, "fetch_chapter", fetch)
|
|
|
|
queue = JobQueue(tmp_path / "jobs.db")
|
|
monkeypatch.setattr(orchestrator, "job_queue", queue)
|
|
job = queue.add(Job(mode=DownloadMode.NEW, story_id=story_id))
|
|
queue.update(queue.get_next(), status=QueueStatus.RUNNING)
|
|
with pytest.raises(Crash):
|
|
orchestrator._run_site_job(job)
|
|
# Chapter Two and its image are stored, in their place.
|
|
assert titles(story_id) == ["One", "Two"]
|
|
assert store.image_urls(story_id) == {IMG_URL}
|
|
|
|
queue = JobQueue(tmp_path / "jobs.db")
|
|
monkeypatch.setattr(orchestrator, "job_queue", queue)
|
|
crash_at.clear()
|
|
resumed = queue.get_next()
|
|
assert (resumed.id, resumed.restarts, resumed.result["new"]) == (job.id, 1, 1)
|
|
message = orchestrator._run_site_job(resumed)
|
|
assert message.startswith("3 new chapters") and "library book 1" in message
|
|
assert titles(story_id) == ["One", "Two", "Three", "Four"]
|
|
|
|
|
|
def test_an_upload_that_is_gone_fails_clearly(fake, jobs, story_epub, tmp_path):
|
|
path = upload(tmp_path, story_epub)
|
|
orchestrator.queue_epub_file(path, "story.epub")
|
|
path.unlink()
|
|
with pytest.raises(StoryError, match="Uploaded file story.epub is gone. Upload it again."):
|
|
orchestrator._run_import_file(jobs.get_next())
|
|
|
|
|
|
# Upload holds
|
|
|
|
|
|
def test_a_new_story_waits_for_approval_of_its_first_upload(fake, jobs, monkeypatch):
|
|
orchestrator.save_settings(hold_new_links=True)
|
|
fake.books = []
|
|
site(monkeypatch, [(1, "One", "a")])
|
|
job = Job(mode=DownloadMode.NEW, url=STORY_URL)
|
|
assert "Approve as new book" in orchestrator._run_site_job(job)
|
|
assert ingest_files() == []
|
|
story = orchestrator.story_details(job.story_id)["story"]
|
|
assert (story["upload_held"], story["needs_approval"], story["ingested_at"]) == (True, True, None)
|
|
assert orchestrator.list_stories()[0]["upload_held"]
|
|
queued = orchestrator.approve_upload(job.story_id, None)
|
|
assert jobs.get_next().id == queued["id"]
|
|
assert rebuild(job.story_id) == "Sent to CWA"
|
|
story = orchestrator.story_details(job.story_id)["story"]
|
|
assert (story["upload_held"], story["needs_approval"]) == (False, False)
|
|
|
|
|
|
def test_a_new_story_linked_to_a_library_book_waits_for_approval(fake, jobs, monkeypatch):
|
|
orchestrator.save_settings(hold_new_links=True)
|
|
site(monkeypatch, [(1, "One", "a")])
|
|
job = Job(mode=DownloadMode.NEW, url=STORY_URL)
|
|
assert "Approve link to library book 1" in orchestrator._run_site_job(job)
|
|
assert ingest_files() == []
|
|
with pytest.raises(ValueError, match="link of this story changed"):
|
|
orchestrator.approve_upload(job.story_id, None)
|
|
orchestrator.approve_upload(job.story_id, "uuid-1")
|
|
assert "library book 1" in rebuild(job.story_id)
|
|
assert len(ingest_files()) == 2
|
|
|
|
|
|
def test_a_changed_link_waits_for_approval(fake, jobs):
|
|
orchestrator.save_settings(hold_new_links=True)
|
|
fake.books.append(FakeBook(2, "uuid-2", "Other Story", epub=b"x"))
|
|
story_id = add_story(ingested_at="2026-01-01T00:00:00", calibre_id=1, calibre_uuid="uuid-1")
|
|
assert "library book 1" in rebuild(story_id)
|
|
orchestrator.link_story(story_id, "uuid-2")
|
|
assert "Approve link to library book 2" in rebuild(story_id)
|
|
assert len(ingest_files()) == 2
|
|
orchestrator.approve_upload(story_id, "uuid-2")
|
|
assert "library book 2" in rebuild(story_id)
|
|
|
|
|
|
def test_a_book_found_after_the_linked_book_is_gone_waits_for_approval(fake, jobs):
|
|
orchestrator.save_settings(hold_new_links=True)
|
|
story_id = add_story(ingested_at="2026-01-01T00:00:00", calibre_id=5, calibre_uuid="uuid-gone")
|
|
assert "Approve link to library book 1" in rebuild(story_id)
|
|
assert ingest_files() == []
|
|
|
|
|
|
def test_the_book_that_cwa_imports_from_the_first_upload_needs_no_approval(fake, jobs):
|
|
orchestrator.save_settings(hold_new_links=True)
|
|
books, fake.books = fake.books, []
|
|
story_id = add_story()
|
|
assert rebuild(story_id) == "Sent to CWA"
|
|
fake.books = books
|
|
assert "library book 1" in rebuild(story_id)
|
|
|
|
|
|
def test_a_link_that_went_to_cwa_before_the_setting_needs_no_approval(fake, jobs):
|
|
story_id = add_story(ingested_at="2026-01-01T00:00:00", calibre_id=1, calibre_uuid="uuid-1")
|
|
orchestrator.link_story(story_id, None)
|
|
orchestrator.link_story(story_id, "uuid-1")
|
|
assert "library book 1" in rebuild(story_id)
|
|
orchestrator.save_settings(hold_new_links=True)
|
|
assert "library book 1" in rebuild(story_id)
|
|
|
|
|
|
def test_paused_uploads_go_to_cwa_when_the_pause_ends(fake, jobs):
|
|
story_id = add_story(ingested_at="2026-01-01T00:00:00", calibre_id=1, calibre_uuid="uuid-1")
|
|
orchestrator.save_settings(pause_uploads=True)
|
|
assert "pause" in rebuild(story_id)
|
|
assert ingest_files() == []
|
|
assert orchestrator.approve_upload(story_id, "uuid-1") is None
|
|
assert jobs.get_next() is None
|
|
orchestrator.save_settings(pause_uploads=True)
|
|
assert jobs.get_next() is None
|
|
orchestrator.save_settings(pause_uploads=False)
|
|
assert "library book 1" in orchestrator._run_rebuild(jobs.get_next())
|
|
assert len(ingest_files()) == 2
|
|
assert not orchestrator.story_details(story_id)["story"]["upload_held"]
|
|
orchestrator.save_settings(pause_uploads=False)
|
|
assert jobs.get_next() is None
|