tim
22478c3a56
The pause of the user and a missing ingest folder are separate checks. One function gives the upload state to the settings and the story page. A route that returns only a story no longer loads its chapters or asks CWA for its book. The story page data loses the values that no page reads.
202 lines
11 KiB
Python
202 lines
11 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]["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.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 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()
|