solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4import json
5from pathlib import Path
6
7import pytest
8
9from convey import create_app
10
11
12@pytest.fixture
13def content_client():
14 journal = Path(__file__).resolve().parent / "fixtures" / "journal"
15 app = create_app(str(journal))
16 return app.test_client()
17
18
19def test_content_list_endpoint(content_client):
20 response = content_client.get("/app/import/api/20260101_090000/content")
21
22 assert response.status_code == 200
23 data = response.get_json()
24 assert data["total"] == 5
25 assert data["source_type"] == "ics"
26 assert data["source_display"] == "Calendar"
27 assert data["months"] == {"202601": 5}
28
29
30def test_content_list_pagination(content_client):
31 response = content_client.get(
32 "/app/import/api/20260101_090000/content?page=2&per_page=2"
33 )
34
35 assert response.status_code == 200
36 data = response.get_json()
37 assert data["page"] == 2
38 assert data["per_page"] == 2
39 assert data["pages"] == 3
40 assert len(data["items"]) == 2
41
42
43def test_content_list_search_filter(content_client):
44 response = content_client.get("/app/import/api/20260101_090000/content?q=betaworks")
45
46 assert response.status_code == 200
47 data = response.get_json()
48 assert data["total"] == 2
49 assert all(
50 "betaworks" in (item["title"] + item["preview"]).lower()
51 for item in data["items"]
52 )
53
54
55def test_content_list_month_filter(content_client):
56 response = content_client.get(
57 "/app/import/api/20260101_100000/content?month=202601"
58 )
59
60 assert response.status_code == 200
61 data = response.get_json()
62 assert data["total"] == 3
63 assert all(item["date"].startswith("202601") for item in data["items"])
64
65
66def test_content_detail_endpoint(content_client):
67 response = content_client.get("/app/import/api/20260101_090000/content/event-0")
68
69 assert response.status_code == 200
70 data = response.get_json()
71 assert data["item"]["title"] == "Weekly Engineering Standup"
72 assert data["content"][0]["type"] == "markdown"
73 assert "Weekly Engineering Standup" in data["content"][0]["content"]
74
75
76def test_content_endpoint_404_for_missing_import(content_client):
77 response = content_client.get("/app/import/api/20990101_000000/content")
78
79 assert response.status_code == 404
80 assert response.get_json()["error"] == "Import not found"
81
82
83def test_content_detail_404_for_missing_item(content_client):
84 response = content_client.get(
85 "/app/import/api/20260101_090000/content/missing-item"
86 )
87
88 assert response.status_code == 404
89 assert response.get_json()["error"] == "Item not found"
90
91
92def test_content_lazy_backfill(tmp_path):
93 journal_root = tmp_path
94 import_dir = journal_root / "imports" / "20260101_120000"
95 seg_dir = journal_root / "20260101" / "import.chatgpt" / "120000_300"
96 import_dir.mkdir(parents=True)
97 seg_dir.mkdir(parents=True)
98
99 (seg_dir / "conversation_transcript.jsonl").write_text(
100 "\n".join(
101 [
102 json.dumps({"topics": "planning"}),
103 json.dumps({"speaker": "Human", "text": "hello"}),
104 json.dumps({"speaker": "Assistant", "text": "hi"}),
105 ]
106 )
107 + "\n",
108 encoding="utf-8",
109 )
110 (import_dir / "imported.json").write_text(
111 json.dumps(
112 {
113 "source_type": "chatgpt",
114 "all_created_files": [
115 "20260101/import.chatgpt/120000_300/conversation_transcript.jsonl",
116 ],
117 }
118 ),
119 encoding="utf-8",
120 )
121
122 app = create_app(str(journal_root))
123 client = app.test_client()
124
125 response = client.get("/app/import/api/20260101_120000/content")
126
127 assert response.status_code == 200
128 data = response.get_json()
129 assert data["total"] == 1
130 assert (import_dir / "content_manifest.jsonl").exists()