feat(mnemosyne): add CLI commands and tests for temporal queries
Closes #1244 - CLI: 'timeline <start> <end>' for date range queries - CLI: 'neighbors <entry_id> [--days N]' for temporal proximity - Tests: 16 test cases covering parsing, ranges, boundaries, sorting, limits, neighbor ordering, and edge cases
This commit is contained in:
BIN
nexus/mnemosyne/__pycache__/cli.cpython-311.pyc
Normal file
BIN
nexus/mnemosyne/__pycache__/cli.cpython-311.pyc
Normal file
Binary file not shown.
@@ -3,7 +3,8 @@
|
||||
Provides: mnemosyne ingest, mnemosyne search, mnemosyne link, mnemosyne stats,
|
||||
mnemosyne topics, mnemosyne remove, mnemosyne export,
|
||||
mnemosyne clusters, mnemosyne hubs, mnemosyne bridges, mnemosyne rebuild,
|
||||
mnemosyne tag, mnemosyne untag, mnemosyne retag
|
||||
mnemosyne tag, mnemosyne untag, mnemosyne retag,
|
||||
mnemosyne timeline, mnemosyne neighbors
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -180,8 +181,45 @@ def cmd_retag(args):
|
||||
print(f" Topics: {', '.join(entry.topics) if entry.topics else '(none)'}")
|
||||
|
||||
|
||||
def cmd_timeline(args):
|
||||
archive = MnemosyneArchive()
|
||||
results = archive.by_date_range(args.start, args.end, limit=args.limit)
|
||||
if not results:
|
||||
print(f"No entries between {args.start} and {args.end}.")
|
||||
return
|
||||
print(f"Timeline: {args.start} → {args.end} ({len(results)} entries)")
|
||||
print()
|
||||
for entry in results:
|
||||
print(f" [{entry.created_at[:10]}] {entry.title}")
|
||||
print(f" ID: {entry.id[:8]} | Source: {entry.source} | Topics: {', '.join(entry.topics)}")
|
||||
print()
|
||||
|
||||
|
||||
def cmd_neighbors(args):
|
||||
archive = MnemosyneArchive()
|
||||
# Resolve prefix to full ID
|
||||
matches = [e for e in archive._entries.values() if e.id.startswith(args.entry_id)]
|
||||
if not matches:
|
||||
print(f"No entry matching '{args.entry_id}'.")
|
||||
return
|
||||
if len(matches) > 1:
|
||||
print(f"Ambiguous — {len(matches)} entries match '{args.entry_id}'. Use a longer prefix.")
|
||||
return
|
||||
entry = matches[0]
|
||||
results = archive.temporal_neighbors(entry.id, window_days=args.days)
|
||||
if not results:
|
||||
print(f"No entries within {args.days} days of [{entry.id[:8]}] {entry.title}.")
|
||||
return
|
||||
print(f"Neighbors of [{entry.id[:8]}] {entry.title} (±{args.days} days):")
|
||||
print()
|
||||
for neighbor in results:
|
||||
print(f" [{neighbor.created_at[:10]}] {neighbor.title}")
|
||||
print(f" ID: {neighbor.id[:8]} | Source: {neighbor.source} | Topics: {', '.join(neighbor.topics)}")
|
||||
print()
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(prog="mnemosyne", description="The Living Holographic Archive")
|
||||
parser = argparse.ArgumentParser(prog="mnemosyne", description="Mnemosyne — Living Holographic Archive")
|
||||
sub = parser.add_subparsers(dest="command")
|
||||
|
||||
sub.add_parser("stats", help="Show archive statistics")
|
||||
@@ -233,6 +271,15 @@ def main():
|
||||
rt.add_argument("entry_id", help="Entry ID")
|
||||
rt.add_argument("tags", help="Comma-separated new tag list")
|
||||
|
||||
tl = sub.add_parser("timeline", help="Show entries within a date range")
|
||||
tl.add_argument("start", help="Start date (YYYY-MM-DD or ISO datetime)")
|
||||
tl.add_argument("end", help="End date (YYYY-MM-DD or ISO datetime)")
|
||||
tl.add_argument("-n", "--limit", type=int, default=50, help="Max entries to show")
|
||||
|
||||
nb = sub.add_parser("neighbors", help="Show entries temporally near a reference entry")
|
||||
nb.add_argument("entry_id", help="Reference entry ID (or prefix)")
|
||||
nb.add_argument("-d", "--days", type=int, default=7, help="Window in days (default 7)")
|
||||
|
||||
args = parser.parse_args()
|
||||
if not args.command:
|
||||
parser.print_help()
|
||||
@@ -253,6 +300,8 @@ def main():
|
||||
"tag": cmd_tag,
|
||||
"untag": cmd_untag,
|
||||
"retag": cmd_retag,
|
||||
"timeline": cmd_timeline,
|
||||
"neighbors": cmd_neighbors,
|
||||
}
|
||||
dispatch[args.command](args)
|
||||
|
||||
|
||||
160
nexus/mnemosyne/tests/test_temporal.py
Normal file
160
nexus/mnemosyne/tests/test_temporal.py
Normal file
@@ -0,0 +1,160 @@
|
||||
"""Tests for Mnemosyne temporal query methods.
|
||||
|
||||
Tests: by_date_range, temporal_neighbors, _parse_datetime.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
|
||||
from nexus.mnemosyne.archive import MnemosyneArchive
|
||||
from nexus.mnemosyne.entry import ArchiveEntry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def archive():
|
||||
"""Create a fresh archive in a temp directory."""
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
path = Path(tmp) / "test_archive.json"
|
||||
a = MnemosyneArchive(archive_path=path)
|
||||
yield a
|
||||
|
||||
|
||||
class TestParseDatetime:
|
||||
"""Test _parse_datetime static method."""
|
||||
|
||||
def test_iso_datetime(self):
|
||||
dt = MnemosyneArchive._parse_datetime("2026-03-15T10:30:00+00:00")
|
||||
assert dt is not None
|
||||
assert dt.year == 2026
|
||||
assert dt.month == 3
|
||||
assert dt.hour == 10
|
||||
|
||||
def test_date_only_start(self):
|
||||
dt = MnemosyneArchive._parse_datetime("2026-03-15", end_of_day=False)
|
||||
assert dt is not None
|
||||
assert dt.hour == 0
|
||||
assert dt.minute == 0
|
||||
|
||||
def test_date_only_end(self):
|
||||
dt = MnemosyneArchive._parse_datetime("2026-03-15", end_of_day=True)
|
||||
assert dt is not None
|
||||
assert dt.hour == 23
|
||||
assert dt.minute == 59
|
||||
assert dt.second == 59
|
||||
|
||||
def test_naive_datetime_gets_utc(self):
|
||||
dt = MnemosyneArchive._parse_datetime("2026-03-15T10:30:00")
|
||||
assert dt is not None
|
||||
assert dt.tzinfo is not None
|
||||
|
||||
def test_empty_string(self):
|
||||
assert MnemosyneArchive._parse_datetime("") is None
|
||||
|
||||
def test_invalid_string(self):
|
||||
assert MnemosyneArchive._parse_datetime("not-a-date") is None
|
||||
|
||||
|
||||
class TestByDateRange:
|
||||
"""Test by_date_range() method."""
|
||||
|
||||
def _add_dated_entry(self, archive, title, created_at):
|
||||
"""Add an entry with a specific created_at timestamp."""
|
||||
entry = ArchiveEntry(title=title, content=f"Content for {title}")
|
||||
entry.created_at = created_at
|
||||
archive._entries[entry.id] = entry
|
||||
archive._save()
|
||||
return entry
|
||||
|
||||
def test_empty_archive(self, archive):
|
||||
results = archive.by_date_range("2026-01-01", "2026-12-31")
|
||||
assert results == []
|
||||
|
||||
def test_single_entry_in_range(self, archive):
|
||||
self._add_dated_entry(archive, "March entry", "2026-03-15T10:00:00+00:00")
|
||||
results = archive.by_date_range("2026-03-01", "2026-03-31")
|
||||
assert len(results) == 1
|
||||
assert results[0].title == "March entry"
|
||||
|
||||
def test_single_entry_out_of_range(self, archive):
|
||||
self._add_dated_entry(archive, "January entry", "2026-01-10T10:00:00+00:00")
|
||||
results = archive.by_date_range("2026-03-01", "2026-03-31")
|
||||
assert results == []
|
||||
|
||||
def test_boundary_inclusive(self, archive):
|
||||
self._add_dated_entry(archive, "Start", "2026-03-01T00:00:00+00:00")
|
||||
self._add_dated_entry(archive, "End", "2026-03-31T23:59:59+00:00")
|
||||
results = archive.by_date_range("2026-03-01", "2026-03-31")
|
||||
assert len(results) == 2
|
||||
|
||||
def test_sorted_by_date(self, archive):
|
||||
self._add_dated_entry(archive, "Late", "2026-03-20T10:00:00+00:00")
|
||||
self._add_dated_entry(archive, "Early", "2026-03-05T10:00:00+00:00")
|
||||
self._add_dated_entry(archive, "Mid", "2026-03-12T10:00:00+00:00")
|
||||
results = archive.by_date_range("2026-03-01", "2026-03-31")
|
||||
assert [r.title for r in results] == ["Early", "Mid", "Late"]
|
||||
|
||||
def test_limit(self, archive):
|
||||
for i in range(10):
|
||||
self._add_dated_entry(archive, f"Entry {i}", f"2026-03-{i+1:02d}T10:00:00+00:00")
|
||||
results = archive.by_date_range("2026-03-01", "2026-03-31", limit=3)
|
||||
assert len(results) == 3
|
||||
|
||||
def test_date_only_strings(self, archive):
|
||||
self._add_dated_entry(archive, "Mid March", "2026-03-15T10:00:00+00:00")
|
||||
results = archive.by_date_range("2026-03-01", "2026-03-31")
|
||||
assert len(results) == 1
|
||||
|
||||
|
||||
class TestTemporalNeighbors:
|
||||
"""Test temporal_neighbors() method."""
|
||||
|
||||
def _add_dated_entry(self, archive, title, created_at):
|
||||
entry = ArchiveEntry(title=title, content=f"Content for {title}")
|
||||
entry.created_at = created_at
|
||||
archive._entries[entry.id] = entry
|
||||
archive._save()
|
||||
return entry
|
||||
|
||||
def test_no_neighbors(self, archive):
|
||||
ref = self._add_dated_entry(archive, "Reference", "2026-03-15T10:00:00+00:00")
|
||||
self._add_dated_entry(archive, "Far away", "2026-06-01T10:00:00+00:00")
|
||||
results = archive.temporal_neighbors(ref.id, window_days=7)
|
||||
assert results == []
|
||||
|
||||
def test_closest_first(self, archive):
|
||||
ref = self._add_dated_entry(archive, "Reference", "2026-03-15T10:00:00+00:00")
|
||||
far = self._add_dated_entry(archive, "6 days away", "2026-03-21T10:00:00+00:00")
|
||||
close = self._add_dated_entry(archive, "1 day away", "2026-03-16T10:00:00+00:00")
|
||||
results = archive.temporal_neighbors(ref.id, window_days=7)
|
||||
assert len(results) == 2
|
||||
assert results[0].title == "1 day away"
|
||||
assert results[1].title == "6 days away"
|
||||
|
||||
def test_excludes_reference(self, archive):
|
||||
ref = self._add_dated_entry(archive, "Reference", "2026-03-15T10:00:00+00:00")
|
||||
self._add_dated_entry(archive, "Nearby", "2026-03-16T10:00:00+00:00")
|
||||
results = archive.temporal_neighbors(ref.id, window_days=7)
|
||||
ids = [r.id for r in results]
|
||||
assert ref.id not in ids
|
||||
|
||||
def test_default_window(self, archive):
|
||||
ref = self._add_dated_entry(archive, "Reference", "2026-03-15T10:00:00+00:00")
|
||||
self._add_dated_entry(archive, "7 days before", "2026-03-08T10:00:00+00:00")
|
||||
self._add_dated_entry(archive, "8 days before", "2026-03-07T10:00:00+00:00")
|
||||
results = archive.temporal_neighbors(ref.id) # default 7 days
|
||||
titles = [r.title for r in results]
|
||||
assert "7 days before" in titles
|
||||
assert "8 days before" not in titles
|
||||
|
||||
def test_custom_window(self, archive):
|
||||
ref = self._add_dated_entry(archive, "Reference", "2026-03-15T10:00:00+00:00")
|
||||
self._add_dated_entry(archive, "14 days away", "2026-03-29T10:00:00+00:00")
|
||||
results = archive.temporal_neighbors(ref.id, window_days=7)
|
||||
assert len(results) == 0
|
||||
results = archive.temporal_neighbors(ref.id, window_days=15)
|
||||
assert len(results) == 1
|
||||
|
||||
def test_bad_entry_id(self, archive):
|
||||
with pytest.raises(KeyError):
|
||||
archive.temporal_neighbors("nonexistent-id")
|
||||
Reference in New Issue
Block a user