135 lines
4.3 KiB
Python
135 lines
4.3 KiB
Python
"""Per-project switch for the Sci-Hub fallback. LEAD projects leave it on."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
import prisma.fetch as fetch
|
|
from cli.prisma import app
|
|
from prisma.fetch import PendingItem, fallback_allowed
|
|
|
|
runner = CliRunner()
|
|
|
|
LEAD = (
|
|
"lead-history",
|
|
"lead-limits",
|
|
"lead-scan",
|
|
"lead-precursors",
|
|
"lead-external",
|
|
)
|
|
|
|
|
|
def _mock_db():
|
|
db = MagicMock()
|
|
db.__enter__ = MagicMock(return_value=db)
|
|
db.__exit__ = MagicMock(return_value=False)
|
|
return db
|
|
|
|
|
|
def test_lead_projects_keep_the_fallback():
|
|
for name in LEAD:
|
|
assert fallback_allowed(name) is True
|
|
|
|
|
|
def test_named_project_skips_the_mirror(monkeypatch):
|
|
monkeypatch.setattr(fetch, "SCIHUB_OFF", frozenset({"skin-subs"}))
|
|
assert fallback_allowed("skin-subs") is False
|
|
assert fallback_allowed("lead-history") is True
|
|
|
|
|
|
def test_allow_fallback_false_skips_the_mirror(tmp_path):
|
|
item = PendingItem(zot_id=1, doi="10.1/x", title="T", pmcid="", pmid="")
|
|
path = tmp_path / "paper.pdf"
|
|
client = MagicMock()
|
|
with (
|
|
patch("prisma.fetch.fetch_fallback", return_value="https://mirror/p.pdf") as fb,
|
|
patch("prisma.fetch.download", return_value=path),
|
|
):
|
|
missed = fetch.fetch_one(
|
|
item,
|
|
email="e@x",
|
|
scratch=tmp_path,
|
|
client=client,
|
|
client_proxied=client,
|
|
allow_fallback=False,
|
|
)
|
|
assert missed is None
|
|
fb.assert_not_called()
|
|
hit = fetch.fetch_one(
|
|
item,
|
|
email="e@x",
|
|
scratch=tmp_path,
|
|
client=client,
|
|
client_proxied=client,
|
|
allow_fallback=True,
|
|
)
|
|
assert hit == ("fallback", path)
|
|
|
|
|
|
def test_run_threads_allow_fallback(tmp_path, monkeypatch):
|
|
item = PendingItem(zot_id=1, doi="10.1/x", title="T", pmcid="", pmid="")
|
|
seen: dict[str, object] = {}
|
|
|
|
def spy(one, **kwargs):
|
|
seen.update(kwargs)
|
|
return None
|
|
|
|
monkeypatch.setattr(fetch, "pending_queue", lambda *a, **k: [item])
|
|
monkeypatch.setattr(fetch, "fetch_one", spy)
|
|
stats = fetch.run(
|
|
MagicMock(),
|
|
project="skin-subs",
|
|
storage_dir=tmp_path,
|
|
scratch_dir=tmp_path / "scratch",
|
|
email="e@x",
|
|
fetch_proxy=None,
|
|
allow_fallback=False,
|
|
progress=False,
|
|
)
|
|
assert seen["allow_fallback"] is False
|
|
assert stats["missed"] == 1
|
|
|
|
|
|
def test_fetch_command_passes_the_project_switch(monkeypatch):
|
|
with (
|
|
patch("prisma.fetch.run", return_value={"missed": 0}) as mc_run,
|
|
patch("zot.db.Db", return_value=_mock_db()),
|
|
patch("prisma.vpn.status", return_value={"status": "down"}),
|
|
):
|
|
result = runner.invoke(
|
|
app, ["fetch", "lead-precursors", "--no-hold", "--no-vpn"]
|
|
)
|
|
assert result.exit_code == 0
|
|
assert mc_run.call_args.kwargs["allow_fallback"] is True
|
|
|
|
monkeypatch.setattr(fetch, "SCIHUB_OFF", frozenset({"skin-subs"}))
|
|
with (
|
|
patch("prisma.fetch.run", return_value={"missed": 0}) as mc_run,
|
|
patch("zot.db.Db", return_value=_mock_db()),
|
|
patch("prisma.vpn.status", return_value={"status": "down"}),
|
|
):
|
|
result = runner.invoke(app, ["fetch", "skin-subs", "--no-hold", "--no-vpn"])
|
|
assert result.exit_code == 0
|
|
assert mc_run.call_args.kwargs["allow_fallback"] is False
|
|
|
|
|
|
def test_orchestrated_run_uses_the_same_switch(monkeypatch):
|
|
monkeypatch.setattr(fetch, "SCIHUB_OFF", frozenset({"skin-subs"}))
|
|
with (
|
|
patch("prisma.screen.run", return_value={"screened": 0, "errors": 0}),
|
|
patch("prisma.fetch.run", return_value={"missed": 0}) as mc_fetch,
|
|
patch("prisma.eligibility.run", return_value={"assessed": 0}),
|
|
patch("prisma.extract.run", return_value={"extracted": 0}),
|
|
patch("prisma.flow.count", return_value=MagicMock()),
|
|
patch("prisma.flow.text_summary", return_value="ok"),
|
|
patch("prisma.vpn.status", return_value={"status": "down"}),
|
|
patch("prisma.llm.make_provider", return_value=MagicMock()),
|
|
patch("prisma.project.load", return_value=MagicMock()),
|
|
patch("zot.db.Db", return_value=_mock_db()),
|
|
):
|
|
result = runner.invoke(app, ["run", "skin-subs", "--no-hold"])
|
|
assert result.exit_code == 0
|
|
assert mc_fetch.call_args.kwargs["allow_fallback"] is False
|