98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
"""stack zot fetch — CLI wiring (no network, no docker)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from typer.testing import CliRunner
|
|
|
|
from cli.zot import app
|
|
from tests.zot.fetch.conftest import mk_item
|
|
from zot.db import Db
|
|
from zot.fetch.deliver import AnnasAuthError
|
|
from zot.fetch.run import STATS
|
|
|
|
runner = CliRunner()
|
|
|
|
|
|
@pytest.fixture()
|
|
def env(monkeypatch, tmp_path: Path):
|
|
monkeypatch.setenv("VPN_HTTP_PROXY", "http://127.0.0.1:19035")
|
|
monkeypatch.setenv("ANNAS_ARCHIVE_KEY", "k")
|
|
monkeypatch.setenv("ANNAS_DAILY_RESERVE", "7")
|
|
monkeypatch.setenv("UNPAYWALL_EMAIL", "dev@fhirworx.io")
|
|
monkeypatch.setattr("cli.zot._hold_zotero", lambda fn, hold: fn())
|
|
return tmp_path
|
|
|
|
|
|
def _db_with_item(zdb: Db) -> str:
|
|
mk_item(zdb, title="T", doi="10.1/x")
|
|
zdb.commit()
|
|
return zdb.path
|
|
|
|
|
|
def test_fetch_requires_proxy(monkeypatch, zdb: Db):
|
|
monkeypatch.delenv("VPN_HTTP_PROXY", raising=False)
|
|
monkeypatch.setattr("cli.zot._hold_zotero", lambda fn, hold: fn())
|
|
r = runner.invoke(app, ["fetch", "--all", "--db", _db_with_item(zdb), "--no-hold"])
|
|
assert r.exit_code != 0
|
|
assert "VPN_HTTP_PROXY" in r.output and "gluetun" in r.output
|
|
|
|
|
|
def test_fetch_dry_run_calls_run_with_selection(env, zdb: Db, monkeypatch):
|
|
captured: dict = {}
|
|
|
|
def fake_run(db, refs, **kw):
|
|
captured["refs"] = refs
|
|
captured["kw"] = kw
|
|
return dict.fromkeys(STATS, 0) | {"seen": 1, "attempted": 1}
|
|
|
|
monkeypatch.setattr("cli.zot.run", fake_run)
|
|
r = runner.invoke(
|
|
app,
|
|
[
|
|
"fetch",
|
|
"--all",
|
|
"--dry-run",
|
|
"--limit",
|
|
"3",
|
|
"--db",
|
|
_db_with_item(zdb),
|
|
"--ledger",
|
|
str(env / "ledger.jsonl"),
|
|
"--no-hold",
|
|
],
|
|
)
|
|
assert r.exit_code == 0, r.output
|
|
assert [x.doi for x in captured["refs"]] == ["10.1/x"]
|
|
kw = captured["kw"]
|
|
assert kw["dry_run"] is True and kw["limit"] == 3 and kw["retry"] is False
|
|
assert kw["budget"].annas_key == "k" and kw["budget"].reserve == 7
|
|
assert kw["email"] == "dev@fhirworx.io"
|
|
assert kw["ledger"].path == env / "ledger.jsonl"
|
|
assert kw["storage_dir"].name == "storage"
|
|
assert "1 eligible" in r.output
|
|
assert "seen=1" in r.output and "attempted=1" in r.output
|
|
|
|
|
|
def test_fetch_selection_error_is_a_usage_error(env, zdb: Db):
|
|
r = runner.invoke(app, ["fetch", "--db", _db_with_item(zdb), "--no-hold"])
|
|
assert r.exit_code != 0 and "exactly one" in r.output
|
|
|
|
|
|
def test_fetch_annas_auth_error_exits_nonzero_with_message(env, zdb: Db, monkeypatch):
|
|
def fake_run(db, refs, **kw):
|
|
raise AnnasAuthError("Anna's Archive refused the key (403)")
|
|
|
|
monkeypatch.setattr("cli.zot.run", fake_run)
|
|
r = runner.invoke(app, ["fetch", "--all", "--db", _db_with_item(zdb), "--no-hold"])
|
|
assert r.exit_code != 0 and "refused the key" in r.output
|
|
|
|
|
|
def test_fetch_without_annas_key_prints_note(env, zdb: Db, monkeypatch):
|
|
monkeypatch.delenv("ANNAS_ARCHIVE_KEY")
|
|
monkeypatch.setattr("cli.zot.run", lambda db, refs, **kw: dict.fromkeys(STATS, 0))
|
|
r = runner.invoke(app, ["fetch", "--all", "--db", _db_with_item(zdb), "--no-hold"])
|
|
assert r.exit_code == 0 and "ANNAS_ARCHIVE_KEY not set" in r.output
|