tim
3e29ab6bf0
The new "Book Downloads" group in Settings holds a Calibre template for the file name of a direct download. Without a template, the name stays the title. The name is also the title if the template fails or gives no name for a book.
265 lines
14 KiB
Python
265 lines
14 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["cwa_enabled"] is True
|
|
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]["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.assert_called_with("book-7", timeout=5) # the page does not wait long for CWA
|
|
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 download_name(client, sid):
|
|
response = client.get(f"/api/stories/{sid}/epub")
|
|
response.close()
|
|
return response.headers["Content-Disposition"]
|
|
|
|
|
|
def test_download_template_names_the_file(client, story):
|
|
sid, _ = story
|
|
site = {"title": "The Book", "authors": ["Ann Writer"], "series": "The Saga", "series_index": 2.0}
|
|
o.store.update_story(sid, metahtml=metahtml(), title="The Book", site_values=site)
|
|
assert client.get("/api/config").json["download_fields"]["id"].startswith("The CWA book ID")
|
|
saved = client.put("/api/settings", json={"download_template": " {series:|| - }{title} - {id} "}, headers=H)
|
|
assert saved.json["download_template"] == "{series:|| - }{title} - {id}"
|
|
assert f'filename="Saga, The - Book, The - {sid}.epub"' in download_name(client, sid)
|
|
# The ID is that of the CWA book, when the story has one.
|
|
o.store.update_story(sid, calibre_id=77, pinned={"title": "Caf\u00e9", "series": ["", None]})
|
|
name = download_name(client, sid)
|
|
assert 'filename="Cafe - 77.epub"' in name and "filename*=UTF-8''Caf%C3%A9%20-%2077.epub" in name
|
|
# A template with folders gives the name of the file alone.
|
|
client.put("/api/settings", json={"download_template": "{author_sort}/{title} - {authors}"}, headers=H)
|
|
assert "filename*=UTF-8''Caf%C3%A9%20-%20Ann%20Writer.epub" in download_name(client, sid)
|
|
# Without a template, the name is the title of the story, with its article in front.
|
|
assert client.put("/api/settings", json={"download_template": ""}, headers=H).json["download_template"] == ""
|
|
assert 'filename="The Book.epub"' in download_name(client, sid)
|
|
|
|
|
|
def test_download_template_that_fails_gives_the_title(client, story, caplog):
|
|
sid, _ = story
|
|
o.store.update_story(sid, metahtml=metahtml())
|
|
for template in ("{title:d}", "{series}/{series_index}"):
|
|
client.put("/api/settings", json={"download_template": template}, headers=H)
|
|
assert "filename=Story.epub" in download_name(client, sid)
|
|
assert "The download template fails for https://www.royalroad.com/fiction/1234" in caplog.text
|
|
assert caplog.text.count("The download template fails") == 1
|
|
|
|
|
|
def test_bad_download_template_is_not_saved(client):
|
|
client.put("/api/settings", json={"download_template": "{title}"}, headers=H)
|
|
for template, error in [
|
|
("{title:uppercase()}", "Bindery does not support template functions, such as {title:uppercase()}"),
|
|
("{title", "The braces of the template do not match: expected '}' before end of string"),
|
|
(7, "The download template must be text"),
|
|
]:
|
|
response = client.put("/api/settings", json={"download_template": template, "recent_chapters": 9}, headers=H)
|
|
assert (response.status_code, response.json) == (400, {"error": error})
|
|
assert client.get("/api/settings").json["download_template"] == "{title}"
|
|
assert client.get("/api/settings").json["recent_chapters"] == 0
|
|
|
|
|
|
def test_unexpected_backend_error_returns_json(client, monkeypatch, caplog):
|
|
monkeypatch.setattr(o, "list_stories", Mock(side_effect=RuntimeError("broken")))
|
|
response = client.get("/api/stories")
|
|
assert response.status_code == 500
|
|
assert response.json == {"error": "Bindery hit an unexpected error. The log has the details."}
|
|
assert "RuntimeError: broken" in caplog.text
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("method", "path", "body", "error"),
|
|
[
|
|
("post", "stories", {"urls": 5}, "urls must be a list"),
|
|
("post", "stories/{sid}/chapters", {"chapter_ids": 5}, "chapter_ids must be a list"),
|
|
("post", "stories/{sid}/chapters/move", {"chapter_ids": [[1]]}, "chapter_ids must be a whole number"),
|
|
("post", "stories/{sid}/chapters/move", {"chapter_ids": [1], "after": [2]}, "after must be a whole number"),
|
|
("post", "stories/{sid}/chapters/delete", {"chapter_ids": "all"}, "chapter_ids must be a list"),
|
|
("post", "stories/{sid}/download", {"mode": "recheck", "chapter_ids": ["x"]}, "chapter_ids must be a whole"),
|
|
("post", "stories/{sid}/download", {"force_fields": "title"}, "force_fields must be a list"),
|
|
("post", "stories/{sid}/download", {"force_fields": [["title"]]}, "force_fields must be a list of names"),
|
|
("put", "stories/{sid}/settings", {"locked_fields": [{}]}, "locked_fields must be a list of names"),
|
|
("put", "stories/{sid}/settings", {"recheck_days": [7]}, "recheck_days must be a whole number"),
|
|
("put", "rules", {"rules": ["a"]}, "rules must be a list of rules"),
|
|
("post", "rules/preview", {"rules": [], "chapter_id": None}, "chapter_id must be a whole number"),
|
|
("post", "chapters/{cid}/revision", {"rev": {}}, "rev must be a whole number"),
|
|
("post", "library/import", {"books": 7}, "books must be a list"),
|
|
("post", "library/import", {"books": [{"id": "x"}]}, "id must be a whole number"),
|
|
],
|
|
)
|
|
def test_a_value_of_the_wrong_type_is_a_bad_request(client, story, method, path, body, error):
|
|
sid, cid = story
|
|
response = getattr(client, method)("/api/" + path.format(sid=sid, cid=cid), json=body, headers=H)
|
|
assert response.status_code == 400
|
|
assert error in response.json["error"]
|
|
|
|
|
|
def test_chapter_requests_without_ids_name_what_to_select(client, story):
|
|
sid, _ = story
|
|
for body in ({}, {"chapter_ids": None}):
|
|
response = client.post(f"/api/stories/{sid}/chapters/delete", json=body, headers=H)
|
|
assert (response.status_code, response.json["error"]) == (400, "Select the chapters to delete")
|
|
|
|
|
|
def test_a_story_change_does_not_ask_cwa_for_the_book(client, story, monkeypatch):
|
|
sid, _ = story
|
|
o.store.update_story(sid, calibre_id=7, calibre_uuid="book-7")
|
|
monkeypatch.setattr(config, "CWA_URL", "https://cwa.example")
|
|
cwa = Mock()
|
|
monkeypatch.setattr(o, "cwa", lambda: cwa)
|
|
response = client.put(f"/api/stories/{sid}/settings", json={"recheck_days": 4}, headers=H)
|
|
assert response.json["recheck_days"] == 4
|
|
assert "chapters" not in response.json
|
|
cwa.get_book.assert_not_called()
|
|
|
|
|
|
def test_settings_are_saved_by_name(client):
|
|
with pytest.raises(ValueError, match="Unknown settings: \\['recent'\\]"):
|
|
o.save_settings(recent=3)
|
|
# The route passes only the known settings on.
|
|
response = client.put("/api/settings", json={"recent": 3, "recent_chapters": 4, "main_jobs_count": 9}, headers=H)
|
|
assert (response.status_code, response.json["recent_chapters"], response.json["main_jobs_count"]) == (200, 4, 9)
|
|
for data, error in [
|
|
({"recent_chapters": -1}, "The number of recent chapters cannot be negative"),
|
|
({"main_jobs_count": 501}, "main_jobs_count cannot be more than 500"),
|
|
({"story_jobs_hours": "x"}, "story_jobs_hours must be a whole number"),
|
|
]:
|
|
assert client.put("/api/settings", json=data, headers=H).json == {"error": error}
|