personal memory agent
test: convert convey-bind tests to mocked unit tests
Old tests launched a real convey server subprocess (cold imports +
flask startup), then waited up to 10s probing a LAN socket to verify
unreachability. ~14s combined on a quiet box, sitting right against
the 15s global pytest timeout — one busy-box run from flaking.
The actual logic under test is _resolve_bind_host(), which reads
get_config().convey.allow_network_access and maps to "127.0.0.1" or
"0.0.0.0". Werkzeug's bind behavior is not solstone code; trust it.
Replaces 2 integration tests with 5 unit tests that mock get_config()
and cover the three real branches (allow True / allow False /
default) plus the defensive exception fallback (config-read errors
must NOT silently switch to "0.0.0.0"). 5 tests in 0.09s.
+26
-113
+26
-113
tests/test_convey_bind.py
+26
-113
tests/test_convey_bind.py
···
3
3
4
4
from __future__ import annotations
5
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
6
+
from unittest.mock import patch
14
7
15
-
import pytest
8
+
from convey.cli import _resolve_bind_host
16
9
17
-
ROOT = Path(__file__).resolve().parents[1]
18
10
11
+
def test_resolve_bind_host_returns_localhost_when_network_access_disabled():
12
+
with patch(
13
+
"think.utils.get_config",
14
+
return_value={"convey": {"allow_network_access": False}},
15
+
):
16
+
assert _resolve_bind_host() == "127.0.0.1"
19
17
20
-
def _read_config(journal_copy) -> dict:
21
-
return json.loads((journal_copy / "config" / "journal.json").read_text("utf-8"))
22
18
19
+
def test_resolve_bind_host_returns_localhost_when_key_absent():
20
+
"""Defaults to localhost when the convey block has no allow_network_access."""
21
+
with patch("think.utils.get_config", return_value={"convey": {}}):
22
+
assert _resolve_bind_host() == "127.0.0.1"
23
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
24
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])
25
+
def test_resolve_bind_host_returns_localhost_when_convey_section_absent():
26
+
"""Defaults to localhost when there's no convey block at all."""
27
+
with patch("think.utils.get_config", return_value={}):
28
+
assert _resolve_bind_host() == "127.0.0.1"
35
29
36
30
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
31
+
def test_resolve_bind_host_returns_all_interfaces_when_network_access_enabled():
32
+
with patch(
33
+
"think.utils.get_config",
34
+
return_value={"convey": {"allow_network_access": True}},
35
+
):
36
+
assert _resolve_bind_host() == "0.0.0.0"
45
37
46
38
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"] = 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
39
+
def test_resolve_bind_host_falls_back_to_localhost_when_config_raises():
40
+
"""Defensive default — config read errors must not open the bind to all interfaces."""
41
+
with patch("think.utils.get_config", side_effect=RuntimeError("config unreadable")):
42
+
assert _resolve_bind_host() == "127.0.0.1"