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 solstone.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 data = response.get_json()
81 assert data["error"] == "I couldn't find that import."
82 assert data["reason_code"] == "import_not_found"
83 assert data["detail"] == "Import not found"
84
85
86def test_content_detail_404_for_missing_item(content_client):
87 response = content_client.get(
88 "/app/import/api/20260101_090000/content/missing-item"
89 )
90
91 assert response.status_code == 404
92 data = response.get_json()
93 assert data["error"] == "I couldn't find that import."
94 assert data["reason_code"] == "import_not_found"
95 assert data["detail"] == "Item not found"
96
97
98def test_content_lazy_backfill(tmp_path):
99 journal_root = tmp_path
100 import_dir = journal_root / "imports" / "20260101_120000"
101 seg_dir = journal_root / "chronicle" / "20260101" / "import.chatgpt" / "120000_300"
102 import_dir.mkdir(parents=True)
103 seg_dir.mkdir(parents=True)
104
105 (seg_dir / "conversation_transcript.jsonl").write_text(
106 "\n".join(
107 [
108 json.dumps({"topics": "planning"}),
109 json.dumps({"speaker": "Human", "text": "hello"}),
110 json.dumps({"speaker": "Assistant", "text": "hi"}),
111 ]
112 )
113 + "\n",
114 encoding="utf-8",
115 )
116 (import_dir / "imported.json").write_text(
117 json.dumps(
118 {
119 "source_type": "chatgpt",
120 "all_created_files": [
121 "20260101/import.chatgpt/120000_300/conversation_transcript.jsonl",
122 ],
123 }
124 ),
125 encoding="utf-8",
126 )
127
128 app = create_app(str(journal_root))
129 client = app.test_client()
130
131 response = client.get("/app/import/api/20260101_120000/content")
132
133 assert response.status_code == 200
134 data = response.get_json()
135 assert data["total"] == 1
136 assert (import_dir / "content_manifest.jsonl").exists()