Add the JSON API and logins
Logins follow shelfmark: none, CWA accounts, or a reverse proxy header. CWA accounts are checked against the OPDS feed, with a lockout after repeated wrong passwords. An API key works in every mode, and changes need an X-Requested-With header or the key. URL_BASE serves the app under a path prefix.
This commit is contained in:
@@ -0,0 +1,311 @@
|
||||
"""The web API: login modes, the CSRF header, URL_BASE and the web interface."""
|
||||
|
||||
import io
|
||||
import time
|
||||
|
||||
import pytest
|
||||
from fake_cwa import FakeCwa
|
||||
from itsdangerous import TimestampSigner
|
||||
from test_build import PNG
|
||||
|
||||
from bindery import auth, config, orchestrator, web
|
||||
from bindery.store import Store
|
||||
|
||||
H = {"X-Requested-With": "bindery"}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(orchestrator, "store", Store(tmp_path / "bindery.db"))
|
||||
monkeypatch.setattr(auth, "limiter", auth.LoginLimiter())
|
||||
web.app.config["TESTING"] = True
|
||||
return web.app.test_client()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cwa(monkeypatch):
|
||||
server = FakeCwa(username="reader", password="pw")
|
||||
monkeypatch.setattr(config, "CWA_URL", server.url)
|
||||
monkeypatch.setattr(config, "AUTH_METHOD", "cwa")
|
||||
yield server
|
||||
server.close()
|
||||
|
||||
|
||||
def login(client, username="reader", password="pw", address="127.0.0.1", forwarded=None):
|
||||
headers = {**H, "X-Forwarded-For": forwarded} if forwarded else H
|
||||
return client.post(
|
||||
"/api/auth/login",
|
||||
json={"username": username, "password": password},
|
||||
headers=headers,
|
||||
environ_base={"REMOTE_ADDR": address},
|
||||
)
|
||||
|
||||
|
||||
def test_health_and_no_login_mode(client):
|
||||
assert client.get("/api/health").json == {"status": "ok"}
|
||||
check = client.get("/api/auth/check").json
|
||||
assert check["authenticated"] and check["mode"] == "none"
|
||||
assert client.get("/api/stories").json == []
|
||||
|
||||
|
||||
def test_changes_need_the_csrf_header(client):
|
||||
assert client.post("/api/stories", json={"urls": []}).status_code == 403
|
||||
response = client.post("/api/stories", json={"urls": ["https://example.com/x"]}, headers=H)
|
||||
assert response.status_code == 200
|
||||
assert response.json["errors"] == ["Not a supported story URL: https://example.com/x"]
|
||||
|
||||
|
||||
def test_errors_are_json(client):
|
||||
assert client.get("/api/stories/99").json == {"error": "No story with id 99"}
|
||||
assert client.get("/api/stories/99").status_code == 404
|
||||
assert client.get("/api/nothing").status_code == 404
|
||||
|
||||
|
||||
def test_cwa_login(client, cwa):
|
||||
assert client.get("/api/stories").status_code == 401
|
||||
assert client.get("/api/auth/check").json == {
|
||||
"authenticated": False,
|
||||
"mode": "cwa",
|
||||
"user": None,
|
||||
"logout_url": None,
|
||||
}
|
||||
assert login(client, password="wrong").status_code == 401
|
||||
assert login(client, username="guest", password="").status_code == 401
|
||||
response = login(client)
|
||||
assert response.status_code == 200 and response.json["user"] == "reader"
|
||||
assert client.get("/api/stories").status_code == 200
|
||||
client.post("/api/auth/logout", headers=H)
|
||||
assert client.get("/api/stories").status_code == 401
|
||||
|
||||
|
||||
def test_a_session_ends_after_one_day(client, cwa, monkeypatch):
|
||||
response = client.post(
|
||||
"/api/auth/login", json={"username": "reader", "password": "pw", "remember": True}, headers=H
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert client.get("/api/stories").status_code == 200
|
||||
later = int(time.time()) + 24 * 60 * 60 + 1
|
||||
monkeypatch.setattr(TimestampSigner, "get_timestamp", lambda self: later)
|
||||
assert client.get("/api/stories").status_code == 401
|
||||
|
||||
|
||||
def test_cwa_login_when_cwa_is_down(client, cwa):
|
||||
cwa.close()
|
||||
assert login(client).json["error"] == "Cannot check the login with CWA. The log has the details."
|
||||
|
||||
|
||||
def test_cwa_login_lockout_and_allowed_users(client, cwa, monkeypatch):
|
||||
for _ in range(auth.MAX_FAILURES):
|
||||
login(client, password="wrong")
|
||||
assert login(client).json["error"].startswith("Too many wrong passwords")
|
||||
monkeypatch.setattr(auth, "limiter", auth.LoginLimiter())
|
||||
monkeypatch.setattr(config, "CWA_ALLOWED_USERS", ["someone-else"])
|
||||
assert login(client).status_code == 401
|
||||
monkeypatch.setattr(config, "CWA_ALLOWED_USERS", ["someone-else", "READER"])
|
||||
assert login(client).status_code == 200
|
||||
|
||||
|
||||
def test_wrong_passwords_lock_a_name_only_for_their_address(client, cwa):
|
||||
for _ in range(auth.MAX_FAILURES):
|
||||
login(client, password="wrong", address="10.0.0.9")
|
||||
assert login(client, address="10.0.0.9").json["error"].startswith("Too many wrong passwords")
|
||||
assert login(client, address="10.0.0.1").status_code == 200
|
||||
|
||||
|
||||
def test_an_address_has_a_limit_for_all_names(client, cwa):
|
||||
for n in range(auth.MAX_ADDRESS_FAILURES):
|
||||
login(client, username=f"user{n}", password="wrong", address="10.0.0.9")
|
||||
assert login(client, address="10.0.0.9").json["error"].startswith("Too many wrong passwords")
|
||||
assert login(client, address="10.0.0.1").status_code == 200
|
||||
|
||||
|
||||
def test_client_address(monkeypatch):
|
||||
headers = {"X-Forwarded-For": "1.1.1.1, 10.0.0.9"}
|
||||
assert auth.client_address(headers, "172.16.0.2") == "172.16.0.2"
|
||||
monkeypatch.setattr(config, "TRUST_PROXY", True)
|
||||
assert auth.client_address(headers, "172.16.0.2") == "10.0.0.9"
|
||||
assert auth.client_address({}, "172.16.0.2") == "172.16.0.2"
|
||||
|
||||
|
||||
def test_only_a_trusted_proxy_sets_the_client_address(client, cwa, monkeypatch):
|
||||
# Without TRUST_PROXY, a client cannot escape the lock with a false header.
|
||||
for n in range(auth.MAX_FAILURES):
|
||||
login(client, password="wrong", address="10.0.0.9", forwarded=f"1.1.1.{n}")
|
||||
assert login(client, address="10.0.0.9", forwarded="1.1.1.9").status_code == 401
|
||||
|
||||
# With TRUST_PROXY, clients behind the proxy have their own locks. The proxy adds the last entry.
|
||||
monkeypatch.setattr(config, "TRUST_PROXY", True)
|
||||
monkeypatch.setattr(auth, "limiter", auth.LoginLimiter())
|
||||
for n in range(auth.MAX_FAILURES):
|
||||
login(client, password="wrong", address="172.16.0.2", forwarded=f"1.1.1.{n}, 10.0.0.9")
|
||||
assert login(client, address="172.16.0.2", forwarded="10.0.0.9").status_code == 401
|
||||
assert login(client, address="172.16.0.2", forwarded="10.0.0.1").status_code == 200
|
||||
|
||||
|
||||
def test_proxy_header(client, monkeypatch):
|
||||
monkeypatch.setattr(config, "AUTH_METHOD", "proxy")
|
||||
monkeypatch.setattr(config, "PROXY_AUTH_LOGOUT_URL", "https://auth.example.com/logout")
|
||||
assert client.get("/api/stories").status_code == 401
|
||||
assert client.get("/api/stories", headers={"X-Remote-User": "alice"}).status_code == 200
|
||||
check = client.get("/api/auth/check", headers={"X-Remote-User": "alice"}).json
|
||||
assert check == {
|
||||
"authenticated": True,
|
||||
"mode": "proxy",
|
||||
"user": "alice",
|
||||
"logout_url": "https://auth.example.com/logout",
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("setting", "value", "error"),
|
||||
[
|
||||
("AUTH_METHOD", "cwa", "AUTH_METHOD=cwa needs CWA_URL or CWA_DB_PATH"),
|
||||
("AUTH_METHOD", "cwa-login", "AUTH_METHOD must be none, cwa or proxy, not 'cwa-login'"),
|
||||
("PROXY_AUTH_USER_HEADER", "", "AUTH_METHOD=proxy needs PROXY_AUTH_USER_HEADER"),
|
||||
],
|
||||
)
|
||||
def test_incomplete_login_settings_refuse_every_request(client, monkeypatch, setting, value, error):
|
||||
monkeypatch.setattr(config, "AUTH_METHOD", "proxy")
|
||||
monkeypatch.setattr(config, "API_KEY", "k" * 32)
|
||||
monkeypatch.setattr(config, setting, value)
|
||||
message = f"{error}. Fix the setting and restart Bindery."
|
||||
for response in (
|
||||
client.get("/api/auth/check"),
|
||||
client.get("/api/stories", headers={"X-Remote-User": "alice"}),
|
||||
client.get("/api/stories", headers={"X-Api-Key": "k" * 32}),
|
||||
client.post("/api/auth/login", json={"username": "a", "password": "b"}, headers=H),
|
||||
):
|
||||
assert (response.status_code, response.json) == (503, {"error": message})
|
||||
assert client.get("/api/health").status_code == 200
|
||||
|
||||
|
||||
def test_api_key(client, cwa, monkeypatch):
|
||||
monkeypatch.setattr(config, "API_KEY", "k" * 32)
|
||||
assert client.get("/api/stories", headers={"X-Api-Key": "wrong"}).status_code == 401
|
||||
assert client.get("/api/stories", headers={"Authorization": "Bearer " + "k" * 32}).status_code == 200
|
||||
# A key needs no CSRF header.
|
||||
response = client.post("/api/stories", json={"urls": []}, headers={"X-Api-Key": "k" * 32})
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_epub_upload(client, monkeypatch, tmp_path):
|
||||
calls = []
|
||||
|
||||
def queue(path, name, story_id, check):
|
||||
calls.append((path.read_bytes(), name, story_id, check))
|
||||
return {"id": "job"}
|
||||
|
||||
monkeypatch.setattr(orchestrator, "queue_epub_file", queue)
|
||||
upload = {"file": (io.BytesIO(b"epub"), "dir/story.epub"), "check": "false"}
|
||||
assert client.post("/api/stories/3/import-epub", data=upload, headers=H).json == {"id": "job"}
|
||||
assert calls == [(b"epub", "story.epub", 3, False)]
|
||||
upload = {"file": (io.BytesIO(b"epub"), "story.epub")}
|
||||
assert client.post("/api/stories/import-epub", data=upload, headers=H).status_code == 200
|
||||
assert calls[1][2:] == (None, True)
|
||||
missing = client.post("/api/stories/import-epub", data={}, headers=H)
|
||||
assert (missing.status_code, missing.json) == (400, {"error": "Choose an epub file"})
|
||||
assert client.post("/api/stories/import-epub", data={}).status_code == 403
|
||||
|
||||
|
||||
def test_the_cover_of_a_story(client):
|
||||
store = orchestrator.store
|
||||
story_id = store.create_story("https://www.royalroad.com/fiction/1", "www.royalroad.com", "T", "A", [])
|
||||
assert client.get(f"/api/stories/{story_id}").json["story"]["cover"] is None
|
||||
assert client.get(f"/api/stories/{story_id}/cover").status_code == 404
|
||||
|
||||
store.put_image(story_id, "https://example.com/cover.png", "png", "image/png", PNG, is_cover=True)
|
||||
version = client.get(f"/api/stories/{story_id}").json["story"]["cover"]
|
||||
response = client.get(f"/api/stories/{story_id}/cover?v={version}")
|
||||
assert (response.status_code, response.mimetype, response.data) == (200, "image/png", PNG)
|
||||
assert response.cache_control.max_age == 365 * 24 * 60 * 60
|
||||
|
||||
# A new cover gets a new version, so the browser loads it.
|
||||
store.put_image(story_id, "https://example.com/new.jpg", "jpg", "image/jpeg", b"new", is_cover=True)
|
||||
assert client.get(f"/api/stories/{story_id}").json["story"]["cover"] not in (None, version)
|
||||
|
||||
# An SVG can hold a script, so Bindery does not serve it.
|
||||
store.put_image(story_id, "https://example.com/cover.svg", "svg", "image/svg+xml", b"<svg/>", is_cover=True)
|
||||
assert client.get(f"/api/stories/{story_id}").json["story"]["cover"] is None
|
||||
assert client.get(f"/api/stories/{story_id}/cover").status_code == 404
|
||||
assert client.get("/api/stories/999/cover").status_code == 404
|
||||
|
||||
|
||||
def test_rules_and_the_paragraph_setting(client):
|
||||
story_id = orchestrator.store.create_story("https://www.royalroad.com/fiction/1", "www.royalroad.com", "T", "A", [])
|
||||
chapter_id = orchestrator.store.add_chapter(story_id, "u1", "One", "<div>a<br>b</div>", "h", "new", 0)
|
||||
saved = client.put("/api/rules", json={"rules": [], "paragraphs": True}, headers=H).json
|
||||
assert saved == {"rules": [], "paragraphs": True, "global_paragraphs": True}
|
||||
story = client.get(f"/api/stories/{story_id}/rules").json
|
||||
assert (story["paragraphs"], story["global_paragraphs"]) == (None, True)
|
||||
assert orchestrator._paragraphs(story_id)
|
||||
client.put(f"/api/stories/{story_id}/rules", json={"rules": [], "paragraphs": False}, headers=H)
|
||||
assert not orchestrator._paragraphs(story_id)
|
||||
# Rules without the setting keep it.
|
||||
client.put(f"/api/stories/{story_id}/rules", json={"rules": []}, headers=H)
|
||||
assert client.get(f"/api/stories/{story_id}/rules").json["paragraphs"] is False
|
||||
preview = client.post(
|
||||
f"/api/stories/{story_id}/rules/preview",
|
||||
json={"rules": [], "chapter_id": chapter_id, "source": True},
|
||||
headers=H,
|
||||
).json
|
||||
assert preview["identical"] is False # follows the global setting
|
||||
preview = client.post(
|
||||
f"/api/stories/{story_id}/rules/preview",
|
||||
json={"rules": [], "chapter_id": chapter_id, "paragraphs": False},
|
||||
headers=H,
|
||||
).json
|
||||
assert preview["identical"] is True
|
||||
|
||||
|
||||
def test_delete_revisions(client):
|
||||
store = orchestrator.store
|
||||
story_id = store.create_story("https://www.royalroad.com/fiction/1", "www.royalroad.com", "T", "A", [])
|
||||
chapter_id = store.add_chapter(story_id, "u1", "One", "<p>one</p>", "h1", "new", 0)
|
||||
store.add_revision(chapter_id, "<p>two</p>", "h2", "recheck")
|
||||
store.add_revision(chapter_id, "<p>three</p>", "h3", "recheck")
|
||||
response = client.delete(f"/api/chapters/{chapter_id}/revisions/3", headers=H)
|
||||
assert response.status_code == 400 and "in the book" in response.json["error"]
|
||||
assert [r["rev"] for r in client.delete(f"/api/chapters/{chapter_id}/revisions/2", headers=H).json] == [1, 3]
|
||||
diff = client.get(f"/api/chapters/{chapter_id}/diff").json
|
||||
assert (diff["a"], diff["b"], diff["identical"], diff["story_title"]) == (1, 3, False, "T")
|
||||
assert client.delete(f"/api/chapters/{chapter_id}/revisions/2", headers=H).status_code == 404
|
||||
# The newest revision stays, also when an older revision is in the book.
|
||||
client.post(f"/api/chapters/{chapter_id}/revision", json={"rev": 1}, headers=H)
|
||||
response = client.delete(f"/api/chapters/{chapter_id}/revisions/3", headers=H)
|
||||
assert response.status_code == 400 and "newest text from the site" in response.json["error"]
|
||||
assert [r["rev"] for r in client.get(f"/api/chapters/{chapter_id}/diff").json["revisions"]] == [1, 3]
|
||||
|
||||
|
||||
def test_settings(client, monkeypatch, tmp_path):
|
||||
ini = tmp_path / "personal.ini"
|
||||
ini.write_text("[defaults]\nis_adult:true\n")
|
||||
monkeypatch.setattr(config, "PERSONAL_INI", ini)
|
||||
assert client.get("/api/settings").json == {
|
||||
"recent_chapters": 0,
|
||||
"hold_new_links": False,
|
||||
"pause_uploads": False,
|
||||
"personal_ini": "[defaults]\nis_adult:true\n",
|
||||
}
|
||||
saved = client.put("/api/settings", json={"recent_chapters": 3}, headers=H).json
|
||||
assert (saved["recent_chapters"], saved["warnings"]) == (3, [])
|
||||
saved = client.put("/api/settings", json={"hold_new_links": True, "pause_uploads": True}, headers=H).json
|
||||
assert (saved["hold_new_links"], saved["pause_uploads"], saved["recent_chapters"]) == (True, True, 3)
|
||||
saved = client.put("/api/settings", json={"personal_ini": "[www.notasite.com]\nx:1\n"}, headers=H).json
|
||||
assert saved["warnings"] == ["Line 1: Bad Section Name: [www.notasite.com]"]
|
||||
assert ini.read_text() == "[www.notasite.com]\nx:1\n"
|
||||
bad = client.put("/api/settings", json={"personal_ini": "[defaults\nfoo"}, headers=H)
|
||||
assert bad.status_code == 400 and "line 1" in bad.json["error"]
|
||||
assert ini.read_text() == "[www.notasite.com]\nx:1\n"
|
||||
assert client.put("/api/settings", json={"recent_chapters": -1}, headers=H).status_code == 400
|
||||
|
||||
|
||||
def test_metadata_edit(client):
|
||||
story_id = orchestrator.store.create_story(
|
||||
"https://www.royalroad.com/fiction/1", "www.royalroad.com", "T", "A", ["title"]
|
||||
)
|
||||
response = client.put(f"/api/stories/{story_id}/metadata", json={"values": {"authors": "A & B"}}, headers=H)
|
||||
assert response.json["story"]["pinned"] == {"authors": ["A", "B"]}
|
||||
response = client.put(f"/api/stories/{story_id}/metadata", json={"values": {"title": ""}}, headers=H)
|
||||
assert (response.status_code, response.json) == (400, {"error": "A story needs a title"})
|
||||
assert client.put(f"/api/stories/{story_id}/metadata", json={}, headers=H).status_code == 400
|
||||
assert client.put("/api/stories/999/metadata", json={"values": {}}, headers=H).status_code == 404
|
||||
@@ -0,0 +1,140 @@
|
||||
"""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"
|
||||
assert client.get("/api/config").json["metadata_fields"]["title"]
|
||||
|
||||
|
||||
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"]
|
||||
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"}
|
||||
@@ -0,0 +1,293 @@
|
||||
"""Read-only Calibre-Web accounts and calibre book IDs, with independent fallbacks."""
|
||||
|
||||
import json
|
||||
import sqlite3
|
||||
from unittest.mock import Mock
|
||||
|
||||
import pytest
|
||||
from test_api import H, login
|
||||
from test_api import client as client
|
||||
from test_deliver import STORY_URL, add_story, rebuild
|
||||
from test_deliver import fake as fake
|
||||
from test_deliver import story_epub as story_epub
|
||||
from werkzeug.security import generate_password_hash
|
||||
|
||||
from bindery import auth, calibre_db, config, orchestrator
|
||||
from bindery.cwa import CwaClient
|
||||
from bindery.ingest import IngestFolder
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def user_db(tmp_path):
|
||||
path = tmp_path / "users #1?.db"
|
||||
with sqlite3.connect(path) as db:
|
||||
db.execute("CREATE TABLE user (id INTEGER PRIMARY KEY, name TEXT, password TEXT, role INTEGER)")
|
||||
db.execute(
|
||||
"INSERT INTO user (name, password, role) VALUES (?, ?, 0)",
|
||||
("reader", generate_password_hash("pw", method="pbkdf2:sha256:1000")),
|
||||
)
|
||||
return path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def library_db(tmp_path):
|
||||
path = tmp_path / "metadata.db"
|
||||
with sqlite3.connect(path) as db:
|
||||
db.executescript(
|
||||
"CREATE TABLE books (id INTEGER PRIMARY KEY, uuid TEXT, title TEXT);"
|
||||
"CREATE TABLE authors (id INTEGER PRIMARY KEY, name TEXT);"
|
||||
"CREATE TABLE books_authors_link (id INTEGER PRIMARY KEY, book INTEGER, author INTEGER);"
|
||||
"CREATE TABLE data (book INTEGER, format TEXT);"
|
||||
"CREATE TABLE identifiers (book INTEGER, type TEXT COLLATE NOCASE, val TEXT COLLATE NOCASE);"
|
||||
"INSERT INTO books VALUES (42, 'uuid-42', 'Library Title');"
|
||||
"INSERT INTO authors VALUES (1, 'Writer');"
|
||||
"INSERT INTO books_authors_link VALUES (1, 42, 1);"
|
||||
"INSERT INTO data VALUES (42, 'EPUB');"
|
||||
)
|
||||
db.execute("INSERT INTO identifiers VALUES (42, 'url', ?)", (STORY_URL,))
|
||||
return path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db_login(client, user_db, monkeypatch):
|
||||
monkeypatch.setattr(config, "AUTH_METHOD", "cwa")
|
||||
monkeypatch.setattr(config, "CWA_DB_PATH", user_db)
|
||||
remote = Mock(spec=CwaClient)
|
||||
remote.login_ok.return_value = True
|
||||
monkeypatch.setattr(orchestrator, "cwa", lambda: remote)
|
||||
return remote
|
||||
|
||||
|
||||
@pytest.mark.parametrize("method", ["pbkdf2:sha256:1000", "scrypt:1024:8:1"])
|
||||
def test_calibre_web_hashes_and_read_only_access(user_db, method):
|
||||
with sqlite3.connect(user_db) as db:
|
||||
db.execute("UPDATE user SET password = ?", (generate_password_hash("pw", method=method),))
|
||||
before = user_db.read_bytes()
|
||||
assert calibre_db.login_ok(user_db, "READER", "pw")
|
||||
assert not calibre_db.login_ok(user_db, "reader", "wrong")
|
||||
assert not calibre_db.login_ok(user_db, "reader", "")
|
||||
assert not calibre_db.login_ok(user_db, "reader' OR 1=1 --", "pw")
|
||||
assert not calibre_db.login_ok(user_db, "missing", "pw")
|
||||
with pytest.raises(calibre_db.DatabaseUnavailable, match="readonly"):
|
||||
with calibre_db._connect(user_db) as db:
|
||||
db.execute("DELETE FROM user")
|
||||
assert user_db.read_bytes() == before
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name,role,password", [("Guest", 0, "pw"), ("reader", 32, "pw"), ("reader", 0, "")])
|
||||
def test_anonymous_or_passwordless_accounts_are_rejected(user_db, name, role, password):
|
||||
with sqlite3.connect(user_db) as db:
|
||||
db.execute("UPDATE user SET name = ?, role = ?, password = ?", (name, role, password))
|
||||
assert not calibre_db.login_ok(user_db, name, "pw")
|
||||
|
||||
|
||||
def test_ambiguous_user_names_are_rejected(user_db):
|
||||
with sqlite3.connect(user_db) as db:
|
||||
db.execute("INSERT INTO user (name, password, role) SELECT 'READER', password, role FROM user")
|
||||
assert not calibre_db.login_ok(user_db, "reader", "pw")
|
||||
|
||||
|
||||
def test_database_login_and_logout_work_without_cwa_url(client, db_login):
|
||||
assert not config.CWA_URL
|
||||
assert auth.config_error() is None
|
||||
assert client.get("/api/stories").status_code == 401
|
||||
assert login(client).status_code == 200
|
||||
assert client.get("/api/auth/check").json["user"] == "reader"
|
||||
assert client.get("/api/stories").status_code == 200
|
||||
assert client.post("/api/auth/logout", headers=H).status_code == 200
|
||||
assert client.get("/api/stories").status_code == 401
|
||||
db_login.login_ok.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("username,password", [("reader", "wrong"), ("missing", "pw"), ("Guest", "pw")])
|
||||
def test_rejected_database_login_never_falls_back(client, db_login, username, password):
|
||||
assert login(client, username=username, password=password).status_code == 401
|
||||
db_login.login_ok.assert_not_called()
|
||||
|
||||
|
||||
def test_database_login_keeps_allowlist_and_rate_limits(client, db_login, monkeypatch):
|
||||
monkeypatch.setattr(config, "CWA_ALLOWED_USERS", ["someone-else"])
|
||||
assert login(client).status_code == 401
|
||||
monkeypatch.setattr(config, "CWA_ALLOWED_USERS", ["READER"])
|
||||
for _ in range(auth.MAX_FAILURES - 1):
|
||||
assert login(client, password="wrong").status_code == 401
|
||||
assert "Too many wrong passwords" in login(client).json["error"]
|
||||
assert login(client, address="10.0.0.2").status_code == 200
|
||||
db_login.login_ok.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("failure", ["missing", "corrupt", "schema", "hash", "role", "binary_hash", "locked"])
|
||||
def test_unavailable_user_database_falls_back_to_opds(client, db_login, user_db, failure, monkeypatch):
|
||||
lock = None
|
||||
if failure == "missing":
|
||||
user_db.unlink()
|
||||
elif failure == "corrupt":
|
||||
user_db.write_bytes(b"not a database")
|
||||
elif failure == "schema":
|
||||
with sqlite3.connect(user_db) as db:
|
||||
db.execute("DROP TABLE user")
|
||||
elif failure == "hash":
|
||||
with sqlite3.connect(user_db) as db:
|
||||
db.execute("UPDATE user SET password = 'unsupported$salt$hash'")
|
||||
elif failure == "role":
|
||||
with sqlite3.connect(user_db) as db:
|
||||
db.execute("UPDATE user SET role = NULL")
|
||||
elif failure == "binary_hash":
|
||||
with sqlite3.connect(user_db) as db:
|
||||
db.execute("UPDATE user SET password = ?", (b"not text",))
|
||||
else:
|
||||
lock = sqlite3.connect(user_db)
|
||||
lock.execute("BEGIN EXCLUSIVE")
|
||||
try:
|
||||
# No metadata.db is needed for this fallback.
|
||||
assert config.CALIBRE_DB_PATH is None
|
||||
assert login(client).status_code == 200
|
||||
db_login.login_ok.assert_called_once_with("reader", "pw")
|
||||
if failure == "missing":
|
||||
assert not user_db.exists()
|
||||
finally:
|
||||
if lock:
|
||||
lock.close()
|
||||
|
||||
|
||||
def test_failed_fallback_does_not_open_a_session(client, db_login, user_db):
|
||||
user_db.unlink()
|
||||
db_login.login_ok.return_value = False
|
||||
assert login(client).status_code == 401
|
||||
assert client.get("/api/stories").status_code == 401
|
||||
|
||||
|
||||
def test_user_database_reads_live_wal_changes(user_db):
|
||||
db = sqlite3.connect(user_db)
|
||||
try:
|
||||
db.execute("PRAGMA journal_mode=WAL")
|
||||
db.execute("UPDATE user SET password = ?", (generate_password_hash("new", method="pbkdf2:sha256:1000"),))
|
||||
db.commit()
|
||||
assert not calibre_db.login_ok(user_db, "reader", "pw")
|
||||
assert calibre_db.login_ok(user_db, "reader", "new")
|
||||
db.execute("DELETE FROM user")
|
||||
db.commit()
|
||||
assert not calibre_db.login_ok(user_db, "reader", "new")
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def test_book_lookup_uses_uuid_or_exact_story_url(library_db):
|
||||
before = library_db.read_bytes()
|
||||
expected = {"id": 42, "uuid": "uuid-42", "title": "Library Title", "authors": ["Writer"], "formats": ["EPUB"]}
|
||||
assert calibre_db.book_by_uuid(library_db, "uuid-42") == expected
|
||||
assert calibre_db.book_by_url(library_db, STORY_URL) == expected
|
||||
assert calibre_db.book_by_uuid(library_db, "missing") is None
|
||||
assert calibre_db.book_by_uuid(library_db, "' OR 1=1 --") is None
|
||||
assert calibre_db.book_by_url(library_db, STORY_URL.upper()) is None
|
||||
assert library_db.read_bytes() == before
|
||||
|
||||
|
||||
@pytest.mark.parametrize("duplicate", ["uuid", "url", "no_uuid"])
|
||||
def test_uncertain_book_identity_does_not_choose_an_update_target(library_db, duplicate):
|
||||
with sqlite3.connect(library_db) as db:
|
||||
if duplicate == "no_uuid":
|
||||
db.execute("UPDATE books SET uuid = NULL")
|
||||
else:
|
||||
db.execute(
|
||||
"INSERT INTO books VALUES (43, ?, 'Duplicate')", ("uuid-42" if duplicate == "uuid" else "uuid-43",)
|
||||
)
|
||||
db.execute("INSERT INTO identifiers VALUES (43, 'url', ?)", (STORY_URL,))
|
||||
assert calibre_db.book_by_url(library_db, STORY_URL) is None
|
||||
if duplicate == "uuid":
|
||||
assert calibre_db.book_by_uuid(library_db, "uuid-42") is None
|
||||
|
||||
|
||||
def test_library_database_alone_does_not_enable_cwa_logins(library_db, monkeypatch):
|
||||
monkeypatch.setattr(config, "CALIBRE_DB_PATH", library_db)
|
||||
monkeypatch.setattr(config, "AUTH_METHOD", "cwa")
|
||||
assert auth.config_error() == "AUTH_METHOD=cwa needs CWA_URL or CWA_DB_PATH"
|
||||
|
||||
|
||||
def test_url_lookup_is_disabled_without_a_library_database():
|
||||
assert CwaClient("").book_by_url(STORY_URL) is None
|
||||
|
||||
|
||||
def test_book_id_reads_do_not_use_app_db(library_db, monkeypatch):
|
||||
monkeypatch.setattr(config, "CALIBRE_DB_PATH", library_db)
|
||||
monkeypatch.setattr(config, "CWA_DB_PATH", library_db.parent / "missing-app.db")
|
||||
client = CwaClient("")
|
||||
client.session.get = Mock(side_effect=AssertionError("No HTTP lookup needed"))
|
||||
assert client.get_book("uuid-42")["id"] == 42
|
||||
assert client.book_by_url(STORY_URL)["id"] == 42
|
||||
assert client.get_book("../bad") is None
|
||||
client.session.get.assert_not_called()
|
||||
|
||||
|
||||
def test_app_db_login_does_not_use_metadata_db(client, db_login, monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(config, "CALIBRE_DB_PATH", tmp_path / "missing-metadata.db")
|
||||
assert login(client).status_code == 200
|
||||
assert not config.CALIBRE_DB_PATH.exists()
|
||||
db_login.login_ok.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("failure", ["missing", "corrupt", "no_match"])
|
||||
def test_library_database_failure_keeps_the_http_lookup(library_db, monkeypatch, failure):
|
||||
monkeypatch.setattr(config, "CALIBRE_DB_PATH", library_db)
|
||||
if failure == "missing":
|
||||
library_db.unlink()
|
||||
elif failure == "corrupt":
|
||||
library_db.write_bytes(b"not a database")
|
||||
else:
|
||||
with sqlite3.connect(library_db) as db:
|
||||
db.execute("DELETE FROM books")
|
||||
client = CwaClient("https://cwa.example")
|
||||
response = Mock(status_code=200, text="record")
|
||||
response.json.return_value = {"application_id": 7, "uuid": "uuid-42", "title": "From CWA"}
|
||||
client.session.get = Mock(return_value=response)
|
||||
assert client.get_book("uuid-42")["id"] == 7
|
||||
assert client.book_by_url(STORY_URL) is None
|
||||
assert client.session.get.call_args.args[0] == "https://cwa.example/ajax/book/uuid-42"
|
||||
if failure == "missing":
|
||||
assert not library_db.exists()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("saved_id,saved_uuid", [(None, None), (1, "uuid-42"), (42, None)])
|
||||
def test_database_book_id_reaches_the_epub_sidecar(client, library_db, monkeypatch, tmp_path, saved_id, saved_uuid):
|
||||
monkeypatch.setattr(config, "CALIBRE_DB_PATH", library_db)
|
||||
monkeypatch.setattr(orchestrator, "ingest_folder", IngestFolder(tmp_path / "ingest"))
|
||||
remote = CwaClient("")
|
||||
remote.session.get = Mock(side_effect=AssertionError("No HTTP lookup needed"))
|
||||
monkeypatch.setattr(orchestrator, "cwa", lambda: remote)
|
||||
story_id = add_story(ingested_at="2026-01-01", calibre_id=saved_id, calibre_uuid=saved_uuid)
|
||||
assert "library book 42" in rebuild(story_id)
|
||||
story = orchestrator.store.get_story(story_id)
|
||||
assert (story["calibre_id"], story["calibre_uuid"]) == (42, "uuid-42")
|
||||
assert client.get(f"/api/stories/{story_id}").json["book"]["id"] == 42
|
||||
manifest = next((tmp_path / "ingest").glob("*.cwa.json"))
|
||||
assert json.loads(manifest.read_text()) == {"action": "add_format", "book_id": 42}
|
||||
remote.session.get.assert_not_called()
|
||||
|
||||
|
||||
def test_metadata_database_miss_falls_back_to_opds(fake, library_db, monkeypatch):
|
||||
monkeypatch.setattr(config, "CALIBRE_DB_PATH", library_db)
|
||||
with sqlite3.connect(library_db) as db:
|
||||
db.execute("DELETE FROM books")
|
||||
story_id = add_story(ingested_at="2026-01-01")
|
||||
assert "library book 1" in rebuild(story_id)
|
||||
assert orchestrator.store.get_story(story_id)["calibre_uuid"] == "uuid-1"
|
||||
|
||||
|
||||
def test_user_database_fallback_does_not_disable_library_database(client, db_login, user_db, library_db, monkeypatch):
|
||||
monkeypatch.setattr(config, "CALIBRE_DB_PATH", library_db)
|
||||
user_db.unlink()
|
||||
assert login(client).status_code == 200
|
||||
db_login.login_ok.assert_called_once_with("reader", "pw")
|
||||
assert CwaClient("").get_book("uuid-42")["id"] == 42
|
||||
|
||||
|
||||
def test_database_match_does_not_take_another_storys_link(client, library_db, monkeypatch):
|
||||
monkeypatch.setattr(config, "CALIBRE_DB_PATH", library_db)
|
||||
other = add_story(calibre_id=42, calibre_uuid="uuid-42")
|
||||
story_id = orchestrator.store.create_story(STORY_URL + "/other", "site", "Story", "Writer", [])
|
||||
story = orchestrator.store.get_story(story_id)
|
||||
with sqlite3.connect(library_db) as db:
|
||||
db.execute("INSERT INTO identifiers VALUES (42, 'url', ?)", (story["url"],))
|
||||
assert orchestrator._linked_book(story) is None
|
||||
assert orchestrator.store.get_story(other)["calibre_id"] == 42
|
||||
assert orchestrator.store.get_story(story_id)["calibre_id"] is None
|
||||
Reference in New Issue
Block a user