293 lines
13 KiB
Python
293 lines
13 KiB
Python
"""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
|
|
|
|
|
|
@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(config, "INGEST_DIR", 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
|