107 lines
3.5 KiB
Python
107 lines
3.5 KiB
Python
"""RSS collector: feed spec parsing, DOI extraction, date filtering, failure isolation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
|
|
import pytest
|
|
|
|
from sebrief.collectors import rss
|
|
from sebrief.collectors.rss import extract_doi, fetch_rss_papers, parse_feed_specs, strip_html
|
|
|
|
START, END = date(2026, 8, 31), date(2026, 9, 6)
|
|
|
|
FEED_XML = """<?xml version="1.0" encoding="UTF-8"?>
|
|
<rss version="2.0"><channel>
|
|
<title>e-Informatica SEJ</title>
|
|
<item>
|
|
<title>Mutation Testing & Coverage</title>
|
|
<link>https://doi.org/10.37190/e-inf/2026.1</link>
|
|
<description><p>We study <b>mutation</b> testing.</p></description>
|
|
<pubDate>Wed, 02 Sep 2026 08:00:00 GMT</pubDate>
|
|
</item>
|
|
<item>
|
|
<title>Too Old Paper</title>
|
|
<link>https://doi.org/10.37190/e-inf/2025.9</link>
|
|
<pubDate>Mon, 03 Aug 2026 08:00:00 GMT</pubDate>
|
|
</item>
|
|
<item>
|
|
<title>No DOI Entry</title>
|
|
<link>https://example.org/announcement</link>
|
|
<pubDate>Thu, 03 Sep 2026 08:00:00 GMT</pubDate>
|
|
</item>
|
|
</channel></rss>
|
|
"""
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, content: bytes):
|
|
self.content = content
|
|
|
|
def raise_for_status(self) -> None:
|
|
pass
|
|
|
|
|
|
def test_parse_feed_specs():
|
|
feeds, invalid = parse_feed_specs(
|
|
["e-Inf|https://example.org/rss", "broken", "|https://x.example", "Name|ftp://bad"]
|
|
)
|
|
assert feeds == [("e-Inf", "https://example.org/rss")]
|
|
assert invalid == ["broken", "|https://x.example", "Name|ftp://bad"]
|
|
|
|
|
|
def test_strip_html():
|
|
assert strip_html("<p>We study <b>mutation</b> testing.</p>") == "We study mutation testing."
|
|
assert strip_html(None) == ""
|
|
assert strip_html("x" * 3000, max_chars=10).endswith("…")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("entry", "expected"),
|
|
[
|
|
({"link": "https://doi.org/10.37190/e-inf/2026.1"}, "10.37190/e-inf/2026.1"),
|
|
({"link": "https://x.org/a", "summary": "see doi 10.1000/xyz."}, "10.1000/xyz"),
|
|
({"id": "urn:doi:10.5555/abc"}, "10.5555/abc"),
|
|
({"link": "https://example.org/none"}, None),
|
|
],
|
|
)
|
|
def test_extract_doi(entry, expected):
|
|
assert extract_doi(entry) == expected
|
|
|
|
|
|
def test_fetch_maps_entries_and_filters_by_date(monkeypatch):
|
|
monkeypatch.setattr(rss.requests, "get", lambda *a, **k: FakeResponse(FEED_XML.encode()))
|
|
|
|
papers = fetch_rss_papers([("e-Inf", "https://example.org/rss")], START, END)
|
|
|
|
assert [p.doi for p in papers] == ["10.37190/e-inf/2026.1"]
|
|
paper = papers[0]
|
|
assert paper.title == "Mutation Testing & Coverage"
|
|
assert paper.abstract == "We study mutation testing."
|
|
assert paper.venue == "e-Inf" and paper.source == "rss:e-Inf"
|
|
assert paper.published_date == date(2026, 9, 2)
|
|
|
|
|
|
def test_one_feed_down_keeps_the_other(monkeypatch):
|
|
def fake_get(url, timeout=None, headers=None):
|
|
if "down" in url:
|
|
raise RuntimeError("boom")
|
|
return FakeResponse(FEED_XML.encode())
|
|
|
|
monkeypatch.setattr(rss.requests, "get", fake_get)
|
|
papers = fetch_rss_papers([("A", "https://down.example/rss"), ("B", "https://ok.example/rss")], START, END)
|
|
assert len(papers) == 1
|
|
|
|
|
|
def test_all_feeds_down_raises(monkeypatch):
|
|
def fake_get(*a, **k):
|
|
raise RuntimeError("boom")
|
|
|
|
monkeypatch.setattr(rss.requests, "get", fake_get)
|
|
with pytest.raises(RuntimeError, match="all rss feeds failed"):
|
|
fetch_rss_papers([("A", "https://a.example/rss")], START, END)
|
|
|
|
|
|
def test_no_feeds_is_a_noop():
|
|
assert fetch_rss_papers([], START, END) == []
|