tim
8faab2463b
The merge cannot always tell where new chapters go. For example, a site can list an announcement before the chapters of an old book. You can now select chapters and move them, in their order, to after another chapter or to the start. You can also delete them with their revisions. A story keeps at least one chapter. The book is then rebuilt. A deleted chapter that the site still lists comes back with the next download. The Move… and Delete… buttons open a dialog that names the selected chapters before anything changes. A shift-click selects a range of chapters. The Select all checkbox moves into the first heading of the chapter list. On a phone, that heading shows with the text "Select all".
153 lines
7.7 KiB
Python
153 lines
7.7 KiB
Python
"""State-changing API flows, using a real store and queue without site requests."""
|
|
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
from test_api import H
|
|
from test_api import client as client
|
|
from test_build import URL, chapters, metahtml
|
|
|
|
from bindery import config
|
|
from bindery import orchestrator as o
|
|
from bindery.cwa import Book, CwaError
|
|
from bindery.jobs import JobQueue
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def isolated_queue(monkeypatch, tmp_path):
|
|
monkeypatch.setattr(o, "job_queue", JobQueue(tmp_path / "jobs.db"))
|
|
|
|
|
|
@pytest.fixture
|
|
def story(client):
|
|
sid = o.store.create_story("https://www.royalroad.com/fiction/1234", "www.royalroad.com", "Story", "Writer", [])
|
|
c = chapters()[0]
|
|
cid = o.store.add_chapter(sid, c["url"], c["title"], c["html"], "hash", "new", 0)
|
|
return sid, cid
|
|
|
|
|
|
def test_add_list_download_and_cancel_jobs(client, story):
|
|
sid, _ = story
|
|
response = client.post("/api/stories", json={"urls": f" {URL}\ninvalid\n"}, headers=H)
|
|
assert response.status_code == 200
|
|
queued = response.json["jobs"][0]
|
|
assert queued["story_id"] == sid
|
|
assert response.json["errors"] == ["Not a supported story URL: invalid"]
|
|
assert client.get("/api/stories").json[0]["busy"]
|
|
assert client.get(f"/api/jobs?story_id={sid}").json[0]["id"] == queued["id"]
|
|
assert client.get("/api/jobs?story_id=999").json == []
|
|
assert client.post(f"/api/jobs/{queued['id']}/cancel", headers=H).json == {"cancelled": True}
|
|
assert client.post(f"/api/jobs/{queued['id']}/cancel", headers=H).json == {"cancelled": False}
|
|
result = client.post(
|
|
f"/api/stories/{sid}/download", json={"mode": "metadata", "force_fields": ["title"]}, headers=H
|
|
)
|
|
assert result.status_code == 200
|
|
assert result.json["force_fields"] == ["title"]
|
|
assert result.json["mode"] == "metadata"
|
|
app_config = client.get("/api/config").json
|
|
assert app_config["metadata_fields"]["title"]
|
|
assert (app_config["story_jobs_hours"], app_config["story_jobs_count"]) == (36, 10)
|
|
assert (app_config["main_jobs_hours"], app_config["main_jobs_count"]) == (6, 25)
|
|
assert (app_config["job_list_max_hours"], app_config["job_list_max_count"]) == (720, 500)
|
|
|
|
|
|
def test_settings_and_inclusion_persist_and_delete_cascades(client, story):
|
|
sid, cid = story
|
|
response = client.put(f"/api/stories/{sid}/settings", json={"auto_update": False, "recheck_days": 4}, headers=H)
|
|
assert response.status_code == 200
|
|
assert not response.json["auto_update"]
|
|
assert response.json["recheck_days"] == 4
|
|
for data in [{"recheck_days": -1}, {"locked_fields": ["unknown"]}]:
|
|
assert client.put(f"/api/stories/{sid}/settings", json=data, headers=H).status_code == 400
|
|
response = client.post(
|
|
f"/api/stories/{sid}/chapters", json={"chapter_ids": [str(cid)], "included": False}, headers=H
|
|
)
|
|
assert response.json["mode"] == "rebuild"
|
|
assert not o.store.get_chapter(cid)["included"]
|
|
response = client.post(f"/api/stories/{sid}/chapters/move", json={"chapter_ids": [cid], "after": None}, headers=H)
|
|
assert response.json["mode"] == "rebuild"
|
|
move = client.post(f"/api/stories/{sid}/chapters/move", json={"chapter_ids": [cid], "after": 999}, headers=H)
|
|
assert move.status_code == 404
|
|
# The story has only this chapter, so it cannot go.
|
|
response = client.post(f"/api/stories/{sid}/chapters/delete", json={"chapter_ids": [cid]}, headers=H)
|
|
assert response.status_code == 400
|
|
assert "at least one chapter" in response.json["error"]
|
|
o.store.set_rules(sid, [{"pattern": "a", "replacement": "b", "enabled": True}])
|
|
response = client.delete(f"/api/stories/{sid}", headers=H)
|
|
assert response.json == {"deleted": sid}
|
|
assert o.store.get_story(sid) is None
|
|
assert o.store.get_chapter(cid) is None
|
|
assert o.store.revision_html(cid, 1) is None
|
|
assert o.store.rules(sid) == []
|
|
|
|
|
|
def test_library_search_link_import_and_unlink(client, story, monkeypatch):
|
|
sid, _ = story
|
|
book = Book(7, "book-7", "Title", ["Writer"], "www.royalroad.com", {"epub": "/book.epub"})
|
|
cwa = Mock()
|
|
cwa.search.return_value = [book]
|
|
cwa.site_books.return_value = [book]
|
|
cwa.get_book.return_value = book.to_dict()
|
|
monkeypatch.setattr(o, "cwa", lambda: cwa)
|
|
monkeypatch.setattr(config, "CWA_URL", "https://cwa.example")
|
|
assert client.get("/api/library/candidates").json[0]["calibre_id"] == 7
|
|
assert client.get("/api/library/search?q=Title").json[0]["tracked_as"] is None
|
|
cwa.search.assert_called_once_with("Title")
|
|
linked = client.post(f"/api/stories/{sid}/link", json={"uuid": "book-7"}, headers=H)
|
|
assert linked.json["calibre_id"] == 7
|
|
assert client.get("/api/library/candidates").json == []
|
|
assert client.get("/api/library/search?q=Title").json[0]["tracked_as"] == "Story"
|
|
cwa.get_book.side_effect = CwaError("offline")
|
|
assert client.get(f"/api/stories/{sid}").json["library_error"] == "offline"
|
|
cwa.get_book.side_effect = None
|
|
cwa.get_book.return_value = None
|
|
assert client.post(f"/api/stories/{sid}/link", json={"uuid": "missing"}, headers=H).status_code == 404
|
|
unlinked = client.post(f"/api/stories/{sid}/link", json={"uuid": None}, headers=H)
|
|
assert unlinked.json["calibre_id"] is None
|
|
assert unlinked.json["ingested_at"] is None
|
|
o.save_settings(hold_new_links=True)
|
|
assert client.get(f"/api/stories/{sid}").json["story"]["needs_approval"]
|
|
wrong = client.post(f"/api/stories/{sid}/approve", json={"uuid": "book-7"}, headers=H)
|
|
assert wrong.status_code == 400 and "changed" in wrong.json["error"]
|
|
approved = client.post(f"/api/stories/{sid}/approve", json={"uuid": None}, headers=H).json
|
|
assert (approved["story"]["needs_approval"], approved["job"]) == (False, None)
|
|
response = client.post("/api/library/import", json={"books": [{"id": 7}, {}, "invalid"], "check": False}, headers=H)
|
|
assert len(response.json) == 1
|
|
assert response.json[0]["title"] == "Library book 7"
|
|
assert response.json[0]["options"] == {"uuid": "", "check": False}
|
|
|
|
|
|
def test_invalid_rules_and_chapter_requests_do_not_change_stored_data(client, story):
|
|
sid, cid = story
|
|
assert client.put("/api/rules", json={"rules": [{"pattern": "["}]}, headers=H).json["index"] == 0
|
|
assert o.store.rules(None) == []
|
|
for data in [{"recent_chapters": "bad"}, {"recent_chapters": -1}]:
|
|
assert client.put("/api/settings", json=data, headers=H).status_code == 400
|
|
assert client.get("/api/chapters/999/diff").status_code == 404
|
|
assert client.get(f"/api/chapters/{cid}/diff?a=1&b=999").status_code == 404
|
|
assert client.post(f"/api/chapters/{cid}/revision", json={"rev": 999}, headers=H).status_code == 404
|
|
assert o.store.get_chapter(cid)["current_rev"] == 1
|
|
response = client.post(f"/api/stories/{sid + 1}/rules/preview", json={"chapter_id": cid, "rules": []}, headers=H)
|
|
assert response.status_code == 404
|
|
response = client.put("/api/rules", json={"rules": [], "paragraphs": None}, headers=H)
|
|
assert response.json["paragraphs"] is False
|
|
|
|
|
|
def test_download_builds_from_stored_chapters_without_changing_them(client, story):
|
|
sid, cid = story
|
|
o.store.update_story(sid, metahtml=metahtml(), title="A/B")
|
|
original = o.store.revision_html(cid, 1)
|
|
response = client.get(f"/api/stories/{sid}/epub")
|
|
assert response.status_code == 200
|
|
assert response.data.startswith(b"PK")
|
|
assert "filename=A_B.epub" in response.headers["Content-Disposition"]
|
|
assert o.store.revision_html(cid, 1) == original
|
|
response.close()
|
|
|
|
|
|
def test_unexpected_backend_error_returns_json(client, monkeypatch):
|
|
monkeypatch.setattr(o, "list_stories", Mock(side_effect=RuntimeError("broken")))
|
|
response = client.get("/api/stories")
|
|
assert response.status_code == 500
|
|
assert response.json == {"error": "RuntimeError: broken"}
|