personal memory agent
tests: cover convey auth matrix and bind probes
Add focused coverage for the four-cell localhost auth contract, real bind-address behavior, and the config/export fixture changes that support the new convey defaults. These tests pin the network-access flip without widening the runtime surface again later.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
+259
-6
+2
-1
tests/test_config_ingest.py
+2
-1
tests/test_config_ingest.py
···
91
91
return {
92
92
"identity": {"name": "Remote User", "preferred": "Remote", "timezone": "UTC"},
93
93
"convey": {
94
+
"allow_network_access": False,
94
95
"password_hash": "secret_hash",
95
96
"secret": "secret_value",
96
97
"trust_localhost": True,
···
181
182
source = load_journal_source(env["key"])
182
183
183
184
assert response.status_code == 200
184
-
assert body == {"staged": True, "skipped": False, "diff_fields": 11}
185
+
assert body == {"staged": True, "skipped": False, "diff_fields": 12}
185
186
assert (state_dir / "source_config.json").exists()
186
187
assert (state_dir / "diff.json").exists()
187
188
assert "last_hash" in _read_json(state_dir / "state.json")
+114
tests/test_convey_auth.py
+114
tests/test_convey_auth.py
···
1
+
# SPDX-License-Identifier: AGPL-3.0-only
2
+
# Copyright (c) 2026 sol pbc
3
+
4
+
from __future__ import annotations
5
+
6
+
import json
7
+
8
+
import pytest
9
+
from flask import Flask, g, jsonify
10
+
from werkzeug.security import generate_password_hash
11
+
12
+
from convey import create_app
13
+
from convey.auth import require_paired_device
14
+
from think.pairing.devices import register_device
15
+
from think.pairing.keys import hash_session_key
16
+
17
+
18
+
def _read_config(journal_copy) -> dict:
19
+
return json.loads((journal_copy / "config" / "journal.json").read_text("utf-8"))
20
+
21
+
22
+
def _write_config(journal_copy, payload: dict) -> None:
23
+
(journal_copy / "config" / "journal.json").write_text(
24
+
json.dumps(payload, indent=2),
25
+
encoding="utf-8",
26
+
)
27
+
28
+
29
+
def _set_auth_matrix_cell(
30
+
journal_copy,
31
+
*,
32
+
allow_network_access: bool,
33
+
trust_localhost: bool,
34
+
) -> None:
35
+
payload = _read_config(journal_copy)
36
+
payload["convey"]["allow_network_access"] = allow_network_access
37
+
payload["convey"]["trust_localhost"] = trust_localhost
38
+
payload["convey"]["password_hash"] = generate_password_hash("test123")
39
+
payload["setup"] = {"completed_at": 1700000000000}
40
+
_write_config(journal_copy, payload)
41
+
42
+
43
+
def _paired_device_app() -> Flask:
44
+
app = Flask(__name__)
45
+
app.secret_key = "test-secret"
46
+
app.config["TESTING"] = True
47
+
48
+
@app.get("/paired")
49
+
@require_paired_device
50
+
def paired_view():
51
+
return jsonify({"device_id": g.paired_device["id"]})
52
+
53
+
return app
54
+
55
+
56
+
@pytest.mark.parametrize("allow_network_access", [False, True])
57
+
@pytest.mark.parametrize("trust_localhost", [True, False])
58
+
@pytest.mark.parametrize("remote_addr", ["127.0.0.1", "10.0.0.9"])
59
+
def test_require_login_matrix(
60
+
journal_copy,
61
+
allow_network_access: bool,
62
+
trust_localhost: bool,
63
+
remote_addr: str,
64
+
):
65
+
_set_auth_matrix_cell(
66
+
journal_copy,
67
+
allow_network_access=allow_network_access,
68
+
trust_localhost=trust_localhost,
69
+
)
70
+
client = create_app(str(journal_copy)).test_client()
71
+
72
+
response = client.get("/", environ_overrides={"REMOTE_ADDR": remote_addr})
73
+
74
+
assert response.status_code == 302
75
+
if remote_addr == "127.0.0.1" and trust_localhost:
76
+
assert "/login" not in response.headers["Location"]
77
+
assert "/init" not in response.headers["Location"]
78
+
else:
79
+
assert "/login" in response.headers["Location"]
80
+
81
+
82
+
@pytest.mark.parametrize("allow_network_access", [False, True])
83
+
@pytest.mark.parametrize("trust_localhost", [True, False])
84
+
@pytest.mark.parametrize("remote_addr", ["127.0.0.1", "10.0.0.9"])
85
+
def test_bearer_auth_succeeds_in_every_matrix_cell(
86
+
journal_copy,
87
+
allow_network_access: bool,
88
+
trust_localhost: bool,
89
+
remote_addr: str,
90
+
):
91
+
_set_auth_matrix_cell(
92
+
journal_copy,
93
+
allow_network_access=allow_network_access,
94
+
trust_localhost=trust_localhost,
95
+
)
96
+
device = register_device(
97
+
name="Phone",
98
+
platform="ios",
99
+
public_key="ssh-ed25519 AAAAbearer",
100
+
session_key_hash=hash_session_key("dsk_real"),
101
+
bundle_id="org.solpbc.solstone-swift",
102
+
app_version="0.1.0",
103
+
paired_at="2026-04-20T15:31:02Z",
104
+
)
105
+
client = _paired_device_app().test_client()
106
+
107
+
response = client.get(
108
+
"/paired",
109
+
headers={"Authorization": "Bearer dsk_real"},
110
+
environ_overrides={"REMOTE_ADDR": remote_addr},
111
+
)
112
+
113
+
assert response.status_code == 200
114
+
assert response.get_json() == {"device_id": device["id"]}
+129
tests/test_convey_bind.py
+129
tests/test_convey_bind.py
···
1
+
# SPDX-License-Identifier: AGPL-3.0-only
2
+
# Copyright (c) 2026 sol pbc
3
+
4
+
from __future__ import annotations
5
+
6
+
import contextlib
7
+
import json
8
+
import os
9
+
import socket
10
+
import subprocess
11
+
import sys
12
+
import time
13
+
from pathlib import Path
14
+
15
+
import pytest
16
+
17
+
ROOT = Path(__file__).resolve().parents[1]
18
+
19
+
20
+
def _read_config(journal_copy) -> dict:
21
+
return json.loads((journal_copy / "config" / "journal.json").read_text("utf-8"))
22
+
23
+
24
+
def _write_config(journal_copy, payload: dict) -> None:
25
+
(journal_copy / "config" / "journal.json").write_text(
26
+
json.dumps(payload, indent=2) + "\n",
27
+
encoding="utf-8",
28
+
)
29
+
30
+
31
+
def _free_port() -> int:
32
+
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
33
+
sock.bind(("127.0.0.1", 0))
34
+
return int(sock.getsockname()[1])
35
+
36
+
37
+
def _detect_lan_ipv4() -> str | None:
38
+
try:
39
+
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
40
+
sock.connect(("8.8.8.8", 80))
41
+
host = sock.getsockname()[0]
42
+
except OSError:
43
+
return None
44
+
return host if host and not host.startswith("127.") else None
45
+
46
+
47
+
def _launch_convey(journal_copy, port: int) -> subprocess.Popen[str]:
48
+
code = (
49
+
f"from convey import create_app; "
50
+
f"from convey.cli import _resolve_bind_host, run_service; "
51
+
f"app = create_app({str(journal_copy)!r}); "
52
+
f"run_service(app, host=_resolve_bind_host(), port={port}, start_watcher=False)"
53
+
)
54
+
env = os.environ.copy()
55
+
env["_SOLSTONE_JOURNAL_OVERRIDE"] = str(journal_copy)
56
+
env["SOL_SKIP_SUPERVISOR_CHECK"] = "1"
57
+
return subprocess.Popen(
58
+
[sys.executable, "-c", code],
59
+
cwd=str(ROOT),
60
+
env=env,
61
+
stdout=subprocess.PIPE,
62
+
stderr=subprocess.PIPE,
63
+
text=True,
64
+
)
65
+
66
+
67
+
def _wait_for_connect(
68
+
host: str, port: int, *, should_succeed: bool, timeout: float = 10.0
69
+
):
70
+
deadline = time.time() + timeout
71
+
last_error = None
72
+
while time.time() < deadline:
73
+
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
74
+
sock.settimeout(0.2)
75
+
try:
76
+
sock.connect((host, port))
77
+
return True
78
+
except OSError as exc:
79
+
last_error = exc
80
+
time.sleep(0.1)
81
+
if should_succeed:
82
+
raise AssertionError(f"failed to connect to {host}:{port}: {last_error}")
83
+
return False
84
+
85
+
86
+
@contextlib.contextmanager
87
+
def _running_convey(journal_copy, *, allow_network_access: bool):
88
+
payload = _read_config(journal_copy)
89
+
payload["convey"]["allow_network_access"] = allow_network_access
90
+
_write_config(journal_copy, payload)
91
+
port = _free_port()
92
+
process = _launch_convey(journal_copy, port)
93
+
try:
94
+
_wait_for_connect("127.0.0.1", port, should_succeed=True)
95
+
yield port
96
+
finally:
97
+
process.terminate()
98
+
try:
99
+
process.wait(timeout=5)
100
+
except subprocess.TimeoutExpired:
101
+
process.kill()
102
+
process.wait(timeout=5)
103
+
if process.returncode not in (0, -15):
104
+
stdout, stderr = process.communicate(timeout=1)
105
+
raise AssertionError(
106
+
f"convey exited unexpectedly ({process.returncode})\nstdout:\n{stdout}\nstderr:\n{stderr}"
107
+
)
108
+
109
+
110
+
@pytest.mark.skipif(sys.platform == "win32", reason="POSIX-only socket probe")
111
+
def test_convey_binds_localhost_only_when_network_access_disabled(journal_copy):
112
+
lan_ip = _detect_lan_ipv4()
113
+
if lan_ip is None:
114
+
pytest.skip("no non-loopback IPv4 available for bind probe")
115
+
116
+
with _running_convey(journal_copy, allow_network_access=False) as port:
117
+
assert _wait_for_connect("127.0.0.1", port, should_succeed=True) is True
118
+
assert _wait_for_connect(lan_ip, port, should_succeed=False) is False
119
+
120
+
121
+
@pytest.mark.skipif(sys.platform == "win32", reason="POSIX-only socket probe")
122
+
def test_convey_binds_all_interfaces_when_network_access_enabled(journal_copy):
123
+
lan_ip = _detect_lan_ipv4()
124
+
if lan_ip is None:
125
+
pytest.skip("no non-loopback IPv4 available for bind probe")
126
+
127
+
with _running_convey(journal_copy, allow_network_access=True) as port:
128
+
assert _wait_for_connect("127.0.0.1", port, should_succeed=True) is True
129
+
assert _wait_for_connect(lan_ip, port, should_succeed=True) is True
+5
-1
tests/test_export.py
+5
-1
tests/test_export.py
···
216
216
config = {
217
217
"identity": {"name": "Test", "preferred": "Tester", "timezone": "UTC"},
218
218
"convey": {
219
+
"allow_network_access": False,
219
220
"password_hash": "secret_hash",
220
221
"secret": "secret_val",
221
222
"trust_localhost": True,
···
1121
1122
"json"
1122
1123
) or mock_session.post.call_args[1].get("json")
1123
1124
posted_config = posted_data["config"]
1124
-
assert posted_config["convey"] == {"trust_localhost": True}
1125
+
assert posted_config["convey"] == {
1126
+
"allow_network_access": False,
1127
+
"trust_localhost": True,
1128
+
}
1125
1129
assert posted_config["setup"] == {"completed_at": 12345}
1126
1130
assert posted_config["env"] == {"KEY": "val"}
+5
-1
tests/test_export_integration.py
+5
-1
tests/test_export_integration.py
···
281
281
{
282
282
"identity": {"name": "Remote User"},
283
283
"retention": {"days": 30},
284
-
"convey": {"trust_localhost": True, "secret": "shhh"},
284
+
"convey": {
285
+
"allow_network_access": False,
286
+
"trust_localhost": True,
287
+
"secret": "shhh",
288
+
},
285
289
"env": {"API_KEY": "xyz"},
286
290
},
287
291
)
+4
-3
tests/test_init.py
+4
-3
tests/test_init.py
···
280
280
from werkzeug.security import check_password_hash
281
281
282
282
assert check_password_hash(config["convey"]["password_hash"], "securepass123")
283
+
assert config["convey"]["allow_network_access"] is False
283
284
assert config["convey"]["trust_localhost"] is True
284
285
# Identity
285
286
assert config["identity"]["name"] == "Jane Doe"
···
448
449
assert resp.status_code == 302
449
450
assert "/init" in resp.headers["Location"]
450
451
451
-
def test_localhost_no_trust_redirects_to_login(self, journal_copy):
452
-
"""Localhost + setup.completed_at but no trust_localhost → redirect to /login."""
452
+
def test_localhost_trust_disabled_redirects_to_login(self, journal_copy):
453
+
"""Localhost + setup.completed_at + trust_localhost false → redirect to /login."""
453
454
config = _read_config(journal_copy)
454
-
config["convey"].pop("trust_localhost", None)
455
+
config["convey"]["trust_localhost"] = False
455
456
config["setup"] = {"completed_at": 1700000000000}
456
457
(journal_copy / "config" / "journal.json").write_text(
457
458
json.dumps(config, indent=2)