solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Tests for think.importers.gemini — Gemini/Bard activity importer."""
5
6import json
7import os
8import tempfile
9import zipfile
10from pathlib import Path
11
12from solstone.think.importers.gemini import GeminiImporter, _parse_activity, _strip_html
13
14importer = GeminiImporter()
15
16
17def _sample_activity(
18 prompt: str = "What is Python?",
19 response: str = "Python is a programming language.",
20 time: str = "2026-01-15T10:30:00Z",
21 header: str = "Gemini Apps",
22 title: str | None = None,
23) -> dict:
24 """Build a sample Gemini activity record."""
25 act: dict = {
26 "header": header,
27 "title": title or f"Asked Gemini: {prompt[:40]}",
28 "time": time,
29 "products": ["Gemini"],
30 "subtitles": [{"value": prompt}],
31 "safeHtmlItem": [{"html": f"<p>{response}</p>"}],
32 }
33 return act
34
35
36def _bard_activity() -> dict:
37 return _sample_activity(
38 prompt="Tell me a joke",
39 response="Why did the chicken cross the road?",
40 header="Bard",
41 title="Talked to Bard",
42 )
43
44
45# --- Unit tests for helpers ---
46
47
48def test_strip_html():
49 assert _strip_html("<p>Hello <b>world</b></p>") == "Hello world"
50 assert _strip_html("No tags") == "No tags"
51 assert _strip_html("& entities <") == "& entities <"
52
53
54def test_parse_activity_basic():
55 act = _sample_activity()
56 messages = _parse_activity(act)
57 assert len(messages) == 2
58 assert set(messages[0]) == {"create_time", "speaker", "text", "model_slug"}
59 assert messages[0]["speaker"] == "Human"
60 assert messages[0]["text"] == "What is Python?"
61 assert messages[0]["model_slug"] is None
62 assert messages[1]["speaker"] == "Assistant"
63 assert "programming language" in messages[1]["text"]
64 assert messages[1]["create_time"] == messages[0]["create_time"]
65
66
67def test_parse_activity_no_content():
68 act = {"header": "Gemini Apps", "time": "2026-01-15T10:00:00Z"}
69 assert _parse_activity(act) == []
70
71
72def test_parse_activity_no_time():
73 act = _sample_activity()
74 del act["time"]
75 assert _parse_activity(act) == []
76
77
78def test_parse_activity_prompt_only():
79 act = {
80 "header": "Gemini Apps",
81 "title": "Asked Gemini",
82 "time": "2026-01-15T10:00:00Z",
83 "subtitles": [{"value": "What is the meaning of life?"}],
84 "products": ["Gemini"],
85 }
86 messages = _parse_activity(act)
87 assert len(messages) == 1
88 assert messages[0]["speaker"] == "Human"
89 assert "meaning of life" in messages[0]["text"]
90
91
92# --- Detection tests ---
93
94
95def test_detect_json_file():
96 with tempfile.NamedTemporaryFile(suffix=".json", mode="w", delete=False) as f:
97 json.dump([_sample_activity()], f)
98 f.flush()
99 try:
100 assert importer.detect(Path(f.name)) is True
101 finally:
102 os.unlink(f.name)
103
104
105def test_detect_json_wrong_format():
106 with tempfile.NamedTemporaryFile(suffix=".json", mode="w", delete=False) as f:
107 json.dump([{"not": "gemini"}], f)
108 f.flush()
109 try:
110 assert importer.detect(Path(f.name)) is False
111 finally:
112 os.unlink(f.name)
113
114
115def test_detect_zip():
116 with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as tmp:
117 with zipfile.ZipFile(tmp, "w") as zf:
118 data = json.dumps([_sample_activity()])
119 zf.writestr("Takeout/My Activity/Gemini Apps/MyActivity.json", data)
120 try:
121 assert importer.detect(Path(tmp.name)) is True
122 finally:
123 os.unlink(tmp.name)
124
125
126def test_detect_directory():
127 with tempfile.TemporaryDirectory() as tmpdir:
128 activity_dir = Path(tmpdir) / "My Activity" / "Gemini Apps"
129 activity_dir.mkdir(parents=True)
130 (activity_dir / "MyActivity.json").write_text(json.dumps([_sample_activity()]))
131 assert importer.detect(Path(tmpdir)) is True
132
133
134def test_detect_directory_no_activity():
135 with tempfile.TemporaryDirectory() as tmpdir:
136 assert importer.detect(Path(tmpdir)) is False
137
138
139# --- Preview tests ---
140
141
142def test_preview_json():
143 with tempfile.NamedTemporaryFile(suffix=".json", mode="w", delete=False) as f:
144 activities = [
145 _sample_activity(time="2026-01-15T10:00:00Z"),
146 _sample_activity(time="2026-02-20T14:00:00Z"),
147 _bard_activity(),
148 ]
149 json.dump(activities, f)
150 f.flush()
151 try:
152 preview = importer.preview(Path(f.name))
153 assert preview.item_count == 3
154 assert preview.date_range[0] == "20260115"
155 assert "Bard" in preview.summary or "bard" in preview.summary.lower()
156 finally:
157 os.unlink(f.name)
158
159
160# --- Process tests ---
161
162
163def test_process_json(monkeypatch):
164 with tempfile.NamedTemporaryFile(suffix=".json", mode="w", delete=False) as f:
165 activities = [
166 _sample_activity(time="2026-01-15T10:00:00Z"),
167 _sample_activity(
168 prompt="How to sort a list?",
169 response="Use sorted().",
170 time="2026-01-15T14:00:00Z",
171 ),
172 ]
173 json.dump(activities, f)
174 f.flush()
175
176 try:
177 with tempfile.TemporaryDirectory() as journal:
178 monkeypatch.setenv("SOLSTONE_JOURNAL", journal)
179 result = importer.process(Path(f.name), Path(journal))
180 assert result.entries_written == 4
181 assert result.errors == []
182 assert result.segments is not None
183 assert len(result.segments) >= 1
184 assert any(
185 Path(p).name == "conversation_transcript.jsonl"
186 for p in result.files_created
187 )
188
189 first_path = Path(result.files_created[0])
190 assert first_path.exists()
191 lines = first_path.read_text().strip().split("\n")
192 metadata = json.loads(lines[0])
193 entries = [json.loads(line) for line in lines[1:]]
194 assert "imported" in metadata
195 assert entries[0]["start"] == "00:00:00"
196 assert entries[0]["speaker"] == "Human"
197 assert entries[0]["source"] == "import"
198 assert entries[1]["speaker"] == "Assistant"
199 finally:
200 os.unlink(f.name)
201
202
203def test_process_zip(monkeypatch):
204 with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as tmp:
205 with zipfile.ZipFile(tmp, "w") as zf:
206 activities = [_sample_activity(time="2026-03-01T09:00:00Z")]
207 zf.writestr(
208 "Takeout/My Activity/Gemini Apps/MyActivity.json",
209 json.dumps(activities),
210 )
211 try:
212 with tempfile.TemporaryDirectory() as journal:
213 monkeypatch.setenv("SOLSTONE_JOURNAL", journal)
214 result = importer.process(Path(tmp.name), Path(journal))
215 assert result.entries_written == 2
216 assert result.segments is not None
217 assert len(result.segments) == 1
218 assert any(Path(p).suffix == ".jsonl" for p in result.files_created)
219 finally:
220 os.unlink(tmp.name)
221
222
223def test_process_multiple_windows(monkeypatch):
224 """Activities more than 5 minutes apart land in different segments."""
225 with tempfile.NamedTemporaryFile(suffix=".json", mode="w", delete=False) as f:
226 activities = [
227 _sample_activity(time="2026-01-15T10:00:00Z"),
228 _sample_activity(
229 prompt="Second question",
230 response="Second answer",
231 time="2026-01-15T10:10:00Z",
232 ),
233 ]
234 json.dump(activities, f)
235 f.flush()
236 try:
237 with tempfile.TemporaryDirectory() as journal:
238 monkeypatch.setenv("SOLSTONE_JOURNAL", journal)
239 result = importer.process(Path(f.name), Path(journal))
240 assert result.entries_written == 4
241 assert result.segments is not None
242 assert len(result.segments) == 2
243 assert len(result.files_created) == 2
244 finally:
245 os.unlink(f.name)
246
247
248# --- Registry test ---
249
250
251def test_registered_in_registry():
252 from solstone.think.importers.file_importer import (
253 FILE_IMPORTER_REGISTRY,
254 get_file_importer,
255 )
256
257 assert "gemini" in FILE_IMPORTER_REGISTRY
258 imp = get_file_importer("gemini")
259 assert imp is not None
260 assert imp.name == "gemini"