102 lines
3.2 KiB
Python
102 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
from zot.fetch.ledger import RETRY_AFTER, TERMINAL, Attempt, Ledger
|
|
|
|
T0 = datetime(2026, 9, 22, 12, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def test_record_appends_one_json_line(tmp_path: Path):
|
|
led = Ledger(tmp_path / "ledger.jsonl")
|
|
att = led.record(
|
|
"ABCD2345",
|
|
outcome="attached",
|
|
source="libgen",
|
|
md5="a" * 32,
|
|
ids={"doi": "10.1/x"},
|
|
now=T0,
|
|
)
|
|
assert isinstance(att, Attempt)
|
|
lines = (tmp_path / "ledger.jsonl").read_text().splitlines()
|
|
assert len(lines) == 1
|
|
row = json.loads(lines[0])
|
|
assert row == {
|
|
"item_key": "ABCD2345",
|
|
"outcome": "attached",
|
|
"source": "libgen",
|
|
"md5": "a" * 32,
|
|
"ids": {"doi": "10.1/x"},
|
|
"note": "",
|
|
"ts": "2026-09-22T12:00:00+00:00",
|
|
}
|
|
|
|
|
|
def test_unknown_item_should_try(tmp_path: Path):
|
|
led = Ledger(tmp_path / "ledger.jsonl")
|
|
assert led.should_try("NEW00000") is True
|
|
|
|
|
|
def test_attached_item_never_retried(tmp_path: Path):
|
|
led = Ledger(tmp_path / "ledger.jsonl")
|
|
led.record("K1", outcome="attached", source="unpaywall", now=T0)
|
|
assert led.should_try("K1", now=T0 + timedelta(days=400)) is False
|
|
|
|
|
|
def test_terminal_miss_waits_thirty_days(tmp_path: Path):
|
|
led = Ledger(tmp_path / "ledger.jsonl")
|
|
for outcome in TERMINAL:
|
|
led.record("K1", outcome=outcome, now=T0)
|
|
early = T0 + RETRY_AFTER - timedelta(seconds=1)
|
|
assert led.should_try("K1", now=early) is False
|
|
assert led.should_try("K1", now=T0 + RETRY_AFTER) is True
|
|
|
|
|
|
def test_retry_flag_overrides_terminal_miss(tmp_path: Path):
|
|
led = Ledger(tmp_path / "ledger.jsonl")
|
|
led.record("K1", outcome="no_candidate", now=T0)
|
|
assert led.should_try("K1", retry=True, now=T0) is True
|
|
|
|
|
|
def test_quota_miss_retries_immediately(tmp_path: Path):
|
|
led = Ledger(tmp_path / "ledger.jsonl")
|
|
led.record("K1", outcome="quota", now=T0)
|
|
assert led.should_try("K1", now=T0) is True
|
|
|
|
|
|
def test_ledger_reloads_last_attempt_per_key(tmp_path: Path):
|
|
p = tmp_path / "ledger.jsonl"
|
|
Ledger(p).record("K1", outcome="no_candidate", now=T0)
|
|
Ledger(p).record("K1", outcome="attached", now=T0)
|
|
led = Ledger(p)
|
|
assert led.should_try("K1", now=T0 + timedelta(days=60)) is False
|
|
|
|
|
|
def test_ledger_skips_corrupt_lines(tmp_path: Path):
|
|
p = tmp_path / "ledger.jsonl"
|
|
good = (
|
|
'{"item_key": "K1", "outcome": "attached", "ts": "2026-09-22T12:00:00+00:00"}'
|
|
)
|
|
p.write_text(good + "\nnot json\n")
|
|
led = Ledger(p)
|
|
assert led.should_try("K1") is False
|
|
assert led.should_try("K2") is True
|
|
|
|
|
|
def test_summary_counts_this_session_only(tmp_path: Path):
|
|
p = tmp_path / "ledger.jsonl"
|
|
Ledger(p).record("K0", outcome="attached", now=T0)
|
|
led = Ledger(p)
|
|
led.record("K1", outcome="attached", now=T0)
|
|
led.record("K2", outcome="no_candidate", now=T0)
|
|
led.record("K3", outcome="no_candidate", now=T0)
|
|
assert led.summary() == {"attached": 1, "no_candidate": 2}
|
|
|
|
|
|
def test_default_now_is_utc_iso(tmp_path: Path):
|
|
led = Ledger(tmp_path / "ledger.jsonl")
|
|
att = led.record("K1", outcome="attached")
|
|
assert att.ts.endswith("+00:00")
|