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 importlib import import_module
7
8import pytest
9from flask import Blueprint, Flask, g, request
10
11import solstone.convey.state as convey_state
12import solstone.think.utils as think_utils
13from solstone.convey.secure_listener import ConveyIdentity
14
15journal_sources = import_module("solstone.apps.import.journal_sources")
16ingest = import_module("solstone.apps.import.ingest")
17
18create_state_directory = journal_sources.create_state_directory
19journal_source_state_prefix = journal_sources.journal_source_state_prefix
20save_journal_source = journal_sources.save_journal_source
21register_ingest_routes = ingest.register_ingest_routes
22
23FINGERPRINT = "sha256:" + "d" * 64
24
25
26def _pl_source() -> dict:
27 return {
28 "pair_mode": "pl",
29 "fingerprint": FINGERPRINT,
30 "device_label": "peer laptop",
31 "paired_at": "2026-05-20T00:00:00Z",
32 "created_at": 1000,
33 "enabled": True,
34 "revoked": False,
35 "revoked_at": None,
36 "stats": {
37 "segments_received": 0,
38 "entities_received": 0,
39 "facets_received": 0,
40 "imports_received": 0,
41 "config_received": 0,
42 },
43 }
44
45
46def _pl_identity() -> ConveyIdentity:
47 return ConveyIdentity(
48 mode="pl-direct",
49 fingerprint=FINGERPRINT,
50 device_label="peer laptop",
51 paired_at="2026-05-20T00:00:00Z",
52 session_id="session-1",
53 )
54
55
56@pytest.fixture
57def pl_ingest_env(tmp_path, monkeypatch):
58 monkeypatch.setattr(convey_state, "journal_root", str(tmp_path), raising=False)
59 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
60 think_utils._journal_path_cache = None
61 (tmp_path / "apps" / "import" / "journal_sources").mkdir(
62 parents=True, exist_ok=True
63 )
64
65 source = _pl_source()
66 assert save_journal_source(source) is True
67 prefix = journal_source_state_prefix(source)
68 create_state_directory(tmp_path, prefix)
69
70 app = Flask(__name__)
71 app.config["TESTING"] = True
72
73 @app.before_request
74 def stamp_identity():
75 stamped = request.environ.get("pl.identity")
76 if stamped is not None:
77 g.identity = stamped
78
79 bp = Blueprint("import-test", __name__, url_prefix="/app/import")
80 register_ingest_routes(bp)
81 app.register_blueprint(bp)
82
83 return {"client": app.test_client(), "prefix": prefix}
84
85
86ROUTES = [
87 ("entities", {"json": {"entities": []}}),
88 ("imports", {"json": {"imports": []}}),
89 ("config", {"json": {"config": {}}}),
90]
91
92
93@pytest.mark.parametrize(("route", "request_kwargs"), ROUTES)
94def test_pl_ingest_routes_accept_fingerprint_prefix(
95 pl_ingest_env, route: str, request_kwargs: dict
96) -> None:
97 response = pl_ingest_env["client"].post(
98 f"/app/import/journal/{pl_ingest_env['prefix']}/ingest/{route}",
99 environ_overrides={"pl.identity": _pl_identity()},
100 **request_kwargs,
101 )
102
103 assert response.status_code == 200
104
105
106@pytest.mark.parametrize(("route", "request_kwargs"), ROUTES)
107def test_pl_ingest_routes_reject_wrong_prefix(
108 pl_ingest_env, route: str, request_kwargs: dict
109) -> None:
110 response = pl_ingest_env["client"].post(
111 f"/app/import/journal/deadbeef/ingest/{route}",
112 environ_overrides={"pl.identity": _pl_identity()},
113 **request_kwargs,
114 )
115
116 assert response.status_code == 403