solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4from __future__ import annotations
5
6from flask import Flask, g, request
7
8from solstone.convey import create_app
9from solstone.convey.push import push_bp
10from solstone.convey.secure_listener.identity import ConveyIdentity
11from solstone.convey.sol_initiated.copy import APNS_CATEGORY_SOL_CHAT_REQUEST
12from solstone.think.push.devices import load_devices
13from solstone.think.push.runtime import stop_all_push_runtime
14
15
16def _identity(fingerprint: str = "fp-1") -> ConveyIdentity:
17 return ConveyIdentity(
18 mode="pl-direct",
19 fingerprint=fingerprint,
20 device_label="iPhone",
21 paired_at="2026-05-20T00:00:00Z",
22 session_id="s1",
23 )
24
25
26def _register_app() -> Flask:
27 app = Flask(__name__)
28 app.config["TESTING"] = True
29
30 @app.before_request
31 def stamp_identity() -> None:
32 stamped = request.environ.get("pl.identity")
33 if stamped is not None:
34 g.identity = stamped
35
36 app.register_blueprint(push_bp)
37 return app
38
39
40def _register_body(token: str = "A" * 64) -> dict[str, str]:
41 return {
42 "device_token": token,
43 "bundle_id": "org.solpbc.solstone-swift",
44 "environment": "development",
45 "platform": "ios",
46 }
47
48
49def _post_register(
50 client, *, identity: ConveyIdentity | None = None, token: str = "A" * 64
51):
52 environ_overrides = {"pl.identity": identity or _identity()}
53 return client.post(
54 "/api/push/register",
55 json=_register_body(token),
56 environ_overrides=environ_overrides,
57 )
58
59
60def _device_row() -> dict[str, object]:
61 return {
62 "fingerprint": "fp-1",
63 "token": "a" * 64,
64 "bundle_id": "org.solpbc.solstone-swift",
65 "environment": "development",
66 "platform": "ios",
67 "registered_at": 1,
68 }
69
70
71def test_register_push_device_happy_path(monkeypatch, tmp_path):
72 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
73 ensure_calls: list[bool] = []
74 monkeypatch.setattr(
75 "solstone.convey.push.ensure_reach_token",
76 lambda: ensure_calls.append(True) or None,
77 )
78 client = _register_app().test_client()
79
80 response = _post_register(client, identity=_identity("fp-1"), token="A" * 64)
81
82 assert response.status_code == 200
83 assert response.get_json() == {"registered": True, "device_count": 1}
84 assert ensure_calls == [True]
85 assert load_devices() == [
86 {
87 "fingerprint": "fp-1",
88 "token": "a" * 64,
89 "bundle_id": "org.solpbc.solstone-swift",
90 "environment": "development",
91 "platform": "ios",
92 "registered_at": load_devices()[0]["registered_at"],
93 }
94 ]
95
96
97def test_register_push_device_replaces_same_fingerprint(monkeypatch, tmp_path):
98 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
99 client = _register_app().test_client()
100
101 first = _post_register(client, identity=_identity("fp-1"), token="A" * 64)
102 second = _post_register(client, identity=_identity("fp-1"), token="B" * 64)
103
104 assert first.get_json()["device_count"] == 1
105 assert second.get_json() == {"registered": True, "device_count": 1}
106 stored = load_devices()
107 assert len(stored) == 1
108 assert stored[0]["fingerprint"] == "fp-1"
109 assert stored[0]["token"] == "b" * 64
110
111
112def test_delete_push_device_by_fingerprint(monkeypatch, tmp_path):
113 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
114 client = _register_app().test_client()
115 identity = _identity("fp-1")
116 _post_register(client, identity=identity)
117
118 response = client.delete(
119 "/api/push/register",
120 environ_overrides={"pl.identity": identity},
121 )
122
123 assert response.status_code == 200
124 assert response.get_json() == {"removed": True, "device_count": 0}
125 assert load_devices() == []
126
127
128def test_register_refuses_none_fingerprint_on_loopback(journal_copy):
129 app = create_app(str(journal_copy))
130 app.config["TESTING"] = True
131 client = app.test_client()
132 try:
133 response = client.post("/api/push/register", json=_register_body())
134 finally:
135 stop_all_push_runtime()
136
137 assert response.status_code == 400
138 assert response.get_json()["reason_code"] == "push_request_invalid"
139
140
141def test_register_refuses_missing_paired_identity(journal_copy):
142 app = create_app(str(journal_copy))
143 app.config["TESTING"] = True
144 client = app.test_client()
145 try:
146 response = client.post(
147 "/api/push/register",
148 json=_register_body(),
149 environ_overrides={"REMOTE_ADDR": "203.0.113.7"},
150 )
151 finally:
152 stop_all_push_runtime()
153
154 assert response.status_code == 400
155 data = response.get_json()
156 assert data["reason_code"] == "push_request_invalid"
157 assert data["detail"] == "push registration requires a paired device"
158
159
160def test_status_shape(monkeypatch):
161 client = _register_app().test_client()
162 monkeypatch.setattr(
163 "solstone.convey.push.load_devices",
164 lambda: [
165 {
166 "fingerprint": "fp-1",
167 "token": "a" * 64,
168 "bundle_id": "org.solpbc.solstone-swift",
169 "environment": "development",
170 "platform": "ios",
171 "registered_at": 1713528000,
172 }
173 ],
174 )
175 monkeypatch.setattr("solstone.convey.push.read_reach_token", lambda: "tok")
176
177 response = client.get("/api/push/status")
178
179 assert response.status_code == 200
180 data = response.get_json()
181 assert set(data) == {"device_count", "relay_available", "devices"}
182 assert data["device_count"] == 1
183 assert data["relay_available"] is True
184 assert data["devices"] == [
185 {
186 "token_suffix": "...aaaa",
187 "bundle_id": "org.solpbc.solstone-swift",
188 "environment": "development",
189 "platform": "ios",
190 "registered_at": "2024-04-19T12:00:00Z",
191 }
192 ]
193
194
195def test_status_relay_unavailable(monkeypatch):
196 client = _register_app().test_client()
197 monkeypatch.setattr("solstone.convey.push.load_devices", lambda: [])
198 monkeypatch.setattr("solstone.convey.push.read_reach_token", lambda: "")
199
200 response = client.get("/api/push/status")
201
202 assert response.status_code == 200
203 assert response.get_json() == {
204 "device_count": 0,
205 "relay_available": False,
206 "devices": [],
207 }
208
209
210def test_push_test_relays(monkeypatch):
211 client = _register_app().test_client()
212 calls: list[dict[str, str]] = []
213 monkeypatch.setattr("solstone.convey.push.load_devices", lambda: [_device_row()])
214 monkeypatch.setattr(
215 "solstone.convey.push.dispatch_via_portal",
216 lambda **kwargs: calls.append(kwargs) or {"ok": True},
217 )
218
219 response = client.post("/api/push/test", json={"body": "hi"})
220
221 assert response.status_code == 200
222 data = response.get_json()
223 assert data["dispatched"] is True
224 assert data["request_id"].startswith("push-test-")
225 assert calls == [
226 {
227 "request_id": data["request_id"],
228 "summary": "hi",
229 "category": APNS_CATEGORY_SOL_CHAT_REQUEST,
230 }
231 ]
232
233
234def test_push_test_returns_503_when_no_devices(monkeypatch):
235 client = _register_app().test_client()
236 monkeypatch.setattr("solstone.convey.push.load_devices", lambda: [])
237
238 response = client.post("/api/push/test")
239
240 assert response.status_code == 503
241 data = response.get_json()
242 assert data["reason_code"] == "feature_unavailable"
243 assert data["detail"] == "no devices to reach"
244
245
246def test_push_test_returns_503_when_relay_dispatch_fails(monkeypatch):
247 client = _register_app().test_client()
248 monkeypatch.setattr("solstone.convey.push.load_devices", lambda: [_device_row()])
249 monkeypatch.setattr("solstone.convey.push.dispatch_via_portal", lambda **_: None)
250
251 response = client.post("/api/push/test")
252
253 assert response.status_code == 503
254 data = response.get_json()
255 assert data["reason_code"] == "feature_unavailable"
256 assert data["detail"] == "push relay dispatch failed"