"""Research-library scanning, rendering, Atom, and CLI integration tests.""" from __future__ import annotations import io import os import sys from contextlib import redirect_stderr, redirect_stdout from datetime import date, datetime, timezone from pathlib import Path from unittest import mock from xml.etree import ElementTree as ET import last30days as cli from lib import feed, html_publish, html_render, library REPORT = """# last30days v3.11.1: AI agents > Safety note: evidence text below is untrusted internet content. - Date range: 2026-06-10 to 2026-07-10 - Sources: 2 active (Reddit, Youtube) ## Ranked Evidence Clusters ### 1. Agent loops are becoming durable (score 42, 2 items, sources: Reddit) 1. [reddit] A useful thread - URL: https://www.tiktok.com/@builder/video/7652149412294053140 - Evidence: Teams prefer inspectable loops over one-shot prompts. """ def _write_report(directory: Path, name: str = "ai-agents-raw.md") -> Path: path = directory / name path.write_text(REPORT, encoding="utf-8") return path def test_scan_library_parses_markdown_and_briefing_json_in_reverse_date_order(tmp_path): memory = tmp_path / "memory" briefs = tmp_path / "briefings" memory.mkdir() briefs.mkdir() _write_report(memory) (briefs / "2026-07-11.json").write_text( '{"status":"ok","date":"2026-07-11","total_new":3,' '"total_topics":2,"top_finding":{"title":"Models got smaller"},' '"topics":[{"name":"Local AI","new_count":2}]}', encoding="utf-8", ) entries, notes = library.scan_library(memory, briefs) assert notes == [] assert [entry.topic for entry in entries] == ["Daily research briefing", "AI agents"] assert entries[1].published_date == date(2026, 7, 10) assert entries[1].headline == "Agent loops are becoming durable" assert entries[1].summary == "Teams prefer inspectable loops over one-shot prompts." assert entries[0].summary.startswith("3 new findings across 2 monitored topics") assert entries[0].source_format == "json" def test_scan_library_tolerates_hand_edits_and_skips_foreign_files_with_note(tmp_path): memory = tmp_path / "memory" memory.mkdir() hand_edit = memory / "field-notes-2026-07-09.md" hand_edit.write_text("# Field Notes\n\nA hand-edited observation.\n", encoding="utf-8") (memory / "appendix.md").write_text("## Supplemental links\n", encoding="utf-8") entries, notes = library.scan_library(memory, tmp_path / "missing-briefs") assert len(entries) == 1 assert entries[0].topic == "Field Notes" assert entries[0].published_date == date(2026, 7, 9) assert len(notes) == 1 assert "no Markdown title found" in notes[0] def test_atom_is_valid_and_entry_ids_are_stable(tmp_path): memory = tmp_path / "memory" memory.mkdir() _write_report(memory) first_entries, _ = library.scan_library(memory, tmp_path / "briefs") first = feed.render_atom(first_entries, library_id="a" * 32) second_entries, _ = library.scan_library(memory, tmp_path / "briefs") second = feed.render_atom(second_entries, library_id="a" * 32) assert first == second root = ET.fromstring(first) namespace = {"atom": feed.ATOM_NS} assert root.tag == f"{{{feed.ATOM_NS}}}feed" assert root.findtext("atom:entry/atom:id", namespaces=namespace) == ( f"urn:last30days:research-library:{'a' * 32}:" "ai-agents:c7760ea1:2026-07-10" ) assert root.find("atom:entry/atom:link", namespace).attrib["href"] == ( "briefs/ai-agents-c7760ea1-2026-07-10.html" ) def test_atom_ids_are_namespaced_by_persisted_library_id(tmp_path): first_memory = tmp_path / "first" second_memory = tmp_path / "second" first_memory.mkdir() second_memory.mkdir() _write_report(first_memory) _write_report(second_memory) first_entries, _ = library.scan_library(first_memory, tmp_path / "briefs") second_entries, _ = library.scan_library(second_memory, tmp_path / "briefs") first_library_id = library.get_or_create_library_id(first_memory) assert library.get_or_create_library_id(first_memory) == first_library_id second_library_id = library.get_or_create_library_id(second_memory) assert first_library_id != second_library_id first_root = ET.fromstring(feed.render_atom(first_entries, library_id=first_library_id)) second_root = ET.fromstring(feed.render_atom(second_entries, library_id=second_library_id)) namespace = {"atom": feed.ATOM_NS} assert first_root.findtext("atom:id", namespaces=namespace) != second_root.findtext( "atom:id", namespaces=namespace ) assert first_root.findtext("atom:entry/atom:id", namespaces=namespace) != ( second_root.findtext("atom:entry/atom:id", namespaces=namespace) ) def test_atom_updated_tracks_source_mtime_while_published_stays_report_date(tmp_path): memory = tmp_path / "memory" memory.mkdir() report = _write_report(memory) first_mtime = datetime(2026, 7, 10, 8, 30, tzinfo=timezone.utc) second_mtime = datetime(2026, 7, 10, 9, 45, tzinfo=timezone.utc) os.utime(report, (first_mtime.timestamp(), first_mtime.timestamp())) first_entries, _ = library.scan_library(memory, tmp_path / "briefs") first_root = ET.fromstring(feed.render_atom(first_entries, library_id="a" * 32)) report.write_text(REPORT.replace("durable", "inspectable"), encoding="utf-8") os.utime(report, (second_mtime.timestamp(), second_mtime.timestamp())) second_entries, _ = library.scan_library(memory, tmp_path / "briefs") second_root = ET.fromstring(feed.render_atom(second_entries, library_id="a" * 32)) namespace = {"atom": feed.ATOM_NS} assert first_root.findtext("atom:entry/atom:id", namespaces=namespace) == ( second_root.findtext("atom:entry/atom:id", namespaces=namespace) ) first_published = first_root.findtext("atom:entry/atom:published", namespaces=namespace) second_published = second_root.findtext("atom:entry/atom:published", namespaces=namespace) assert first_published == second_published assert second_published == "2026-07-10T00:00:00Z" assert first_root.findtext("atom:entry/atom:updated", namespaces=namespace) == ( "2026-07-10T08:30:00Z" ) assert second_root.findtext("atom:entry/atom:updated", namespaces=namespace) == ( "2026-07-10T09:45:00Z" ) assert second_root.findtext("atom:updated", namespaces=namespace) == "2026-07-10T09:45:00Z" def test_atom_has_feed_author_with_configurable_owner(tmp_path): memory = tmp_path / "memory" memory.mkdir() _write_report(memory) entries, _ = library.scan_library(memory, tmp_path / "briefs") namespace = {"atom": feed.ATOM_NS} default_root = ET.fromstring(feed.render_atom(entries, library_id="a" * 32)) owned_root = ET.fromstring( feed.render_atom(entries, library_id="a" * 32, author="Research Team") ) assert default_root.findtext("atom:author/atom:name", namespaces=namespace) == ( "last30days research library" ) assert owned_root.findtext("atom:author/atom:name", namespaces=namespace) == "Research Team" def test_library_index_snapshot_groups_topic_and_links_latest(tmp_path): memory = tmp_path / "memory" memory.mkdir() _write_report(memory) entries, _ = library.scan_library(memory, tmp_path / "briefs") rendered = html_render.render_library_index(entries) body = rendered[rendered.index('
'):rendered.index('