mvanhorn--last30days-skill/tests/test_arxiv_source.py

"""Tests for the arXiv source adapter (lib/arxiv.py).

Covers query construction (quoted phrase + relevance sort), the recency
cutoff that doubles as off-topic gating, field mapping, and graceful
degradation when the binary is absent.
"""

from __future__ import annotations

from datetime import datetime, timedelta, timezone

from lib import arxiv


NOW = datetime(2026, 6, 27, tzinfo=timezone.utc)


def _entry(title, published, summary="An abstract.", authors=("Ada Lovelace",), abs_url="https://arxiv.org/abs/2606.00001v1"):
    return {
        "id": "http://arxiv.org/abs/2606.00001v1",
        "title": title,
        "summary": summary,
        "published": published,
        "authors": [{"name": a} for a in authors],
        "categories": [{"term": "cs.AI"}],
        "links": [
            {"rel": "alternate", "href": abs_url, "type": "text/html"},
            {"rel": "related", "href": "https://arxiv.org/pdf/2606.00001v1", "title": "pdf"},
        ],
    }


# ---- query construction ----

def test_search_query_is_quoted_phrase_relevance_sorted():
    args = arxiv._build_search_args("AI coding agents", 10)
    joined = " ".join(args)
    assert '--search-query' in args
    assert 'all:"AI coding agents"' in args  # quoted phrase
    assert "--sort-by" in args and "relevance" in args  # NOT submittedDate
    assert "submittedDate" not in joined


def test_search_query_strips_inner_quotes():
    q = arxiv._build_search_query('say "hello" world')
    assert q == 'all:"say hello world"'


# ---- envelope extraction ----

def test_extract_entries_handles_nested_results_envelope():
    data = {"meta": {"source": "live"}, "results": {"entries": [{"title": "X"}]}}
    entries = arxiv._extract_entries(data)
    assert len(entries) == 1 and entries[0]["title"] == "X"


# ---- happy path + field mapping ----

def test_happy_path_maps_fields():
    recent = (NOW - timedelta(days=10)).strftime("%Y-%m-%dT12:00:00Z")
    resp = {"results": [_entry("Agent Memory as a Database", recent, authors=("A", "B"))]}
    items = arxiv.parse_arxiv_response(resp, query="agent memory", today=NOW)
    assert len(items) == 1
    it = items[0]
    assert it["title"] == "Agent Memory as a Database"
    assert it["url"] == "https://arxiv.org/abs/2606.00001v1"  # alternate, not pdf
    assert it["date"] == (NOW - timedelta(days=10)).date().isoformat()
    assert it["authors"] == ["A", "B"]
    assert it["author"] == "A et al."  # multi-author label


def test_url_prefers_alternate_over_pdf():
    recent = (NOW - timedelta(days=5)).strftime("%Y-%m-%dT12:00:00Z")
    resp = {"results": [_entry("T", recent)]}
    items = arxiv.parse_arxiv_response(resp, query="t", today=NOW)
    assert "/pdf/" not in items[0]["url"]
    assert "/abs/" in items[0]["url"]


# ---- recency cutoff (also off-topic gating) ----

def test_recency_cutoff_drops_stale_paper():
    stale = (NOW - timedelta(days=arxiv.RECENCY_DAYS + 30)).strftime("%Y-%m-%dT12:00:00Z")
    resp = {"results": [_entry("Old Paper", stale)]}
    items = arxiv.parse_arxiv_response(resp, query="old paper", today=NOW)
    assert items == []


def test_off_topic_control_drops_via_recency():
    # The "Golden State Warriors" keyword match is a 2017 stats paper -- the
    # recency cutoff is what keeps arXiv quiet on non-research topics.
    old = "2017-06-12T12:00:00Z"
    resp = {"results": [_entry("Do Steph Curry and Klay Thompson Have Hot Hands?", old)]}
    items = arxiv.parse_arxiv_response(resp, query="Golden State Warriors", today=NOW)
    assert items == []


def test_unparseable_or_missing_date_is_dropped():
    resp = {"results": [_entry("No Date", None), _entry("Bad Date", "not-a-date")]}
    items = arxiv.parse_arxiv_response(resp, query="x", today=NOW)
    assert items == []


def test_future_dated_entry_is_dropped():
    future = (NOW + timedelta(days=5)).strftime("%Y-%m-%dT12:00:00Z")
    resp = {"results": [_entry("Future", future)]}
    items = arxiv.parse_arxiv_response(resp, query="future", today=NOW)
    assert items == []


# ---- error / degradation paths ----

def test_binary_absent_returns_empty(monkeypatch):
    monkeypatch.setattr(arxiv.shutil, "which", lambda _bin: None)
    resp = arxiv.search_arxiv("anything", "2026-06-01", "2026-06-27")
    assert resp["results"] == []
    assert "error" in resp


def test_empty_topic_returns_empty():
    assert arxiv.search_arxiv("   ", "2026-06-01", "2026-06-27") == {"results": []}


def test_parse_handles_non_list_results():
    assert arxiv.parse_arxiv_response({"results": "oops"}, query="x", today=NOW) == []
    assert arxiv.parse_arxiv_response({}, query="x", today=NOW) == []


def test_quote_only_topic_returns_empty():
    # A topic of only quote characters cleans to an empty phrase; don't search.
    assert arxiv.search_arxiv('"', "2026-06-01", "2026-06-27") == {"results": []}


def test_same_day_future_timestamp_is_kept():
    # A paper announced later the same UTC day yields age_days == -1; the
    # one-day grace keeps it rather than dropping it as "future".
    later_today = NOW.replace(hour=23, minute=59).strftime("%Y-%m-%dT%H:%M:00Z")
    resp = {"results": [_entry("Fresh paper", later_today)]}
    items = arxiv.parse_arxiv_response(resp, query="fresh paper", today=NOW.replace(hour=1))
    assert len(items) == 1


class _Proc:
    def __init__(self, rc, out, err=""):
        self.returncode, self.stdout, self.stderr = rc, out, err


def test_run_cli_flattens_nested_envelope(monkeypatch):
    monkeypatch.setattr(arxiv, "_is_available", lambda: True)
    payload = '{"meta":{},"results":{"entries":[{"title":"X"}]}}'
    monkeypatch.setattr(arxiv.subproc, "run_with_timeout", lambda cmd, timeout: _Proc(0, payload))
    resp = arxiv.search_arxiv("topic", "2026-06-01", "2026-06-27")
    assert resp["results"] == [{"title": "X"}]


def test_run_cli_nonzero_exit_returns_error(monkeypatch):
    monkeypatch.setattr(arxiv, "_is_available", lambda: True)
    monkeypatch.setattr(arxiv.subproc, "run_with_timeout", lambda cmd, timeout: _Proc(1, "", "boom\nmore"))
    resp = arxiv.search_arxiv("topic", "2026-06-01", "2026-06-27")
    assert resp["results"] == [] and "boom" in resp["error"]


def test_run_cli_bad_json_returns_error(monkeypatch):
    monkeypatch.setattr(arxiv, "_is_available", lambda: True)
    monkeypatch.setattr(arxiv.subproc, "run_with_timeout", lambda cmd, timeout: _Proc(0, "not json"))
    resp = arxiv.search_arxiv("topic", "2026-06-01", "2026-06-27")
    assert resp["results"] == [] and "error" in resp