marfrit/bullpen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
bullpen/tests/test_mneme_server.py
T

659 lines
26 KiB
Python
Raw Permalink Blame History

"""Executable contract for mneme v1.1 — src/mneme/server.py (+ store.py v1.1 notes).
This suite IS the spec. It must FAIL until server.py exists and store.py is
upgraded to v1.1. Do NOT weaken a test to make it pass.
Pinned server contract (stdlib only, mirror bullseye/server.py house style):
python3 -m mneme.server --port P --db PATH --drain-interval SECS
- binds 127.0.0.1 only (like bullseye), ThreadingHTTPServer.
- --db default /var/lib/mneme/mneme.db (tests always pass a tmp path).
- --drain-interval float seconds, default 60. A background thread
drains the write_queue every interval. It must wait on a
threading.Event (interruptible), NOT a busy-poll / bare sleep spin:
SIGTERM must (a) run ONE final drain and (b) exit promptly even when
the interval is huge (test_sigterm_final_drain enforces both).
- drains never overlap (serialize with a lock or single thread).
Auth (bullseye _post_token/_authorized pattern):
- token from env MNEME_TOKEN, else first non-empty non-comment line of
~/.config/mneme/token. None configured => writes DISABLED fail-closed.
- POST /save: no token configured -> 503; wrong/absent Bearer -> 401
(+ WWW-Authenticate: Bearer); good Bearer -> 200.
- comparison via hmac.compare_digest (constant time).
- GET endpoints require no auth (reads stay open).
Endpoints (JSON in/out, charset utf-8):
POST /save {"ns": str, "text": str} -> 200 {"id": int}.
Queue-append ONLY (Store.save); must NOT drain/index
synchronously. Content-Type must be application/json
(else 415); malformed JSON -> 400; declared
Content-Length > 1 MiB -> 413 before reading body.
GET /query?q=&k= -> 200 JSON array of {"ns","id","text","score"},
BM25 best-first. Hostile FTS5 metacharacters in q,
or garbage k, must never yield a 5xx.
GET /namespaces -> 200 JSON object {ns: {"count","last"}}.
GET /recent?n= -> 200 JSON array (newest first).
GET /keywords?ns= -> 200 JSON array of strings.
GET /similar?to=<id> -> 200 JSON array (related drained rows).
GET /status -> 200 {"degraded": bool, "queue_depth": int,
"queue_age": float|null}.
GET /healthz -> 200, cheap liveness: opens NO DB connection,
performs NO DB write (db/-wal bytes must not change).
Pinned store.py v1.1 notes (source-level, deliberate):
- drain() inserts with plain INSERT (never INSERT OR REPLACE): if two
drains ever overlap, the second must ERROR, not silently re-write rows.
- memories_fts declares id and ns as UNINDEXED: namespace/id tokens must
not be matchable (behavioural twin: test_query_does_not_match_ns_tokens).
- every per-batch connection is explicitly closed ("with sqlite3.connect"
commits but does NOT close).
Run: python3 -m pytest tests/test_mneme_server.py -q
Grinder-sandbox notes (bwrap --unshare-all): the fresh netns starts with 'lo'
DOWN — a session fixture raises it via SIOCSIFFLAGS ioctl (userns grants
CAP_NET_ADMIN, no `ip` binary needed); if that fails, every HTTP test fails
immediately instead of eating its startup timeout. The suite watchdog aborts
at 110 s so it always beats the grinder's own <=150 s kill.
"""
import faulthandler
import hashlib
import http.client
import json
import os
import re
import signal
import socket
import sqlite3
import subprocess
import sys
import threading
import time
import urllib.parse
from pathlib import Path
import pytest
# abspath, NOT resolve(): a sandboxed working copy may reach here via symlink,
# and resolve() would escape it back to the original checkout.
ROOT = Path(os.path.abspath(__file__)).parents[1]
SRC = ROOT / "src"
STORE_PY = SRC / "mneme" / "store.py"
SERVER_PY = SRC / "mneme" / "server.py"
TOKEN = "sesame-t0ken-42"
LONG = "3600" # drain interval that never fires during a test
SHORT = "0.2" # drain interval for background-drain tests
HTTP_TIMEOUT = 5.0 # socket timeout on every HTTP call
STARTUP_DEADLINE = 6.0 # server must accept a connection within this
STOP_GRACE = 5.0 # SIGTERM grace before the group gets SIGKILL
SUITE_BUDGET = 110.0 # hard wall-clock cap: must beat the grinder's
# own <=150s kill, or the grinder SIGKILLs pytest
# mid-flight and reports a crash instead of failures
# Every live server's process-group id, so the watchdog can nuke them all.
_live_pgids = set()
_live_lock = threading.Lock()
def _killpg(pgid, sig):
try:
os.killpg(pgid, sig)
except (ProcessLookupError, PermissionError):
pass
# ----------------------------------------------------------------------
# loopback bring-up: the grinder sandbox (bwrap --unshare-all) starts a
# fresh netns whose 'lo' is DOWN, so every 127.0.0.1 connect fails until
# it is raised. Inside the user namespace we hold CAP_NET_ADMIN, so a
# plain SIOCSIFFLAGS ioctl works — no `ip` binary needed.
# ----------------------------------------------------------------------
IFF_UP = 0x1
SIOCGIFFLAGS = 0x8913
SIOCSIFFLAGS = 0x8914
# Non-None => loopback is unusable; Server() fails each HTTP test fast
# with this message instead of burning STARTUP_DEADLINE per test.
_LO_ERROR = None
def _ensure_loopback_up():
"""Bring 'lo' UP if it is down (no-op when already up). Returns None on
success, else a string describing why loopback is unusable."""
import fcntl
import struct
pad = b"\0" * 22 # struct ifreq is 40 bytes; flags live at offset 16
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
except OSError as e:
return "cannot open AF_INET socket: %s" % e
try:
req = struct.pack("16sH", b"lo", 0) + pad
try:
flags = struct.unpack_from("16sH", fcntl.ioctl(s, SIOCGIFFLAGS, req))[1]
except OSError as e:
return "SIOCGIFFLAGS(lo) failed: %s" % e
if not flags & IFF_UP:
try:
fcntl.ioctl(s, SIOCSIFFLAGS,
struct.pack("16sH", b"lo", flags | IFF_UP) + pad)
except OSError as e:
return "SIOCSIFFLAGS could not bring 'lo' UP: %s" % e
flags = struct.unpack_from("16sH", fcntl.ioctl(s, SIOCGIFFLAGS, req))[1]
if not flags & IFF_UP:
return "IFF_UP did not stick on 'lo'"
finally:
s.close()
# end-to-end proof, cheap (~ms): bind + accept-connect over loopback
try:
with socket.socket() as srv:
srv.bind(("127.0.0.1", 0))
srv.listen(1)
with socket.create_connection(srv.getsockname(), timeout=2.0):
pass
except OSError as e:
return "loopback self-check (bind+connect 127.0.0.1) failed: %s" % e
return None
@pytest.fixture(scope="session", autouse=True)
def _loopback_up():
global _LO_ERROR
_LO_ERROR = _ensure_loopback_up()
yield
@pytest.fixture(scope="session", autouse=True)
def _suite_watchdog():
"""Last-resort guarantee the grinder never hangs: if the suite exceeds
SUITE_BUDGET, dump all thread stacks, SIGKILL every server process group,
and hard-exit. Individual waits below are all bounded; this only fires if
something truly unforeseen wedges."""
cancel = threading.Event()
def guard():
if not cancel.wait(SUITE_BUDGET):
msg = ("\n*** test_mneme_server watchdog: suite exceeded %.0fs — "
"killing servers, aborting (exit 70) ***\n" % SUITE_BUDGET)
# pytest's fd-capture swallows stderr when we os._exit, so also
# leave a postmortem in the working dir the runner preserves.
streams = [sys.stderr]
try:
streams.append(open("test_mneme_server.watchdog.log", "w"))
except OSError:
pass
for f in streams:
try:
f.write(msg)
faulthandler.dump_traceback(file=f)
f.flush()
except Exception:
pass
with _live_lock:
pgids = list(_live_pgids)
for pg in pgids:
_killpg(pg, signal.SIGKILL)
os._exit(70)
t = threading.Thread(target=guard, daemon=True)
t.start()
yield
cancel.set()
def _free_port():
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
class Server:
"""Launch `python3 -m mneme.server` black-box; context manager kills it."""
def __init__(self, tmp, *, token=TOKEN, token_file=None, interval=LONG):
if _LO_ERROR:
pytest.fail("loopback unusable, HTTP tests cannot run: %s"
% _LO_ERROR, pytrace=False)
self.tmp = Path(tmp)
self.db = str(self.tmp / "mneme.db")
self.port = _free_port()
self.home = self.tmp / "home"
self.home.mkdir(exist_ok=True)
if token_file is not None:
cfg = self.home / ".config" / "mneme"
cfg.mkdir(parents=True, exist_ok=True)
(cfg / "token").write_text(token_file)
env = os.environ.copy()
env.pop("MNEME_TOKEN", None)
env["HOME"] = str(self.home) # so a real ~/.config token never leaks in
env["PYTHONPATH"] = str(SRC) + os.pathsep + env.get("PYTHONPATH", "")
if token is not None:
env["MNEME_TOKEN"] = token
self.stderr_path = self.tmp / "server.stderr"
self._stderr_f = open(self.stderr_path, "wb")
# start_new_session: the server becomes its own process group leader,
# so teardown can killpg() it and any children it forks.
self.proc = subprocess.Popen(
[sys.executable, "-m", "mneme.server",
"--port", str(self.port), "--db", self.db,
"--drain-interval", interval],
env=env, stdout=subprocess.DEVNULL, stderr=self._stderr_f,
start_new_session=True,
)
self.pgid = self.proc.pid
with _live_lock:
_live_pgids.add(self.pgid)
try:
self._wait_up()
except BaseException:
self.stop()
raise
def _wait_up(self, deadline=STARTUP_DEADLINE):
"""Bounded startup wait: raises RuntimeError (after killing the
process group) if the server neither accepts nor exits in time."""
t0 = time.monotonic()
while time.monotonic() - t0 < deadline:
if self.proc.poll() is not None:
break
try:
with socket.create_connection(("127.0.0.1", self.port), 0.2):
return
except OSError:
time.sleep(0.05)
rc = self.proc.poll()
self.stop()
raise RuntimeError(
"mneme server did not come up within %.0fs: rc=%s stderr=%r"
% (deadline, rc, self.stderr())
)
def stderr(self):
if not self._stderr_f.closed:
self._stderr_f.flush()
return self.stderr_path.read_bytes().decode("utf-8", "replace")
def stop(self, sig=signal.SIGTERM, timeout=STOP_GRACE):
"""Signal the whole process group and wait, SIGKILL fallback; always
bounded, always reaps, idempotent. Returns the exit code (None only
if even SIGKILL failed to reap, which the caller asserts against)."""
try:
if self.proc.poll() is None:
_killpg(self.pgid, sig)
try:
self.proc.wait(timeout=timeout)
except subprocess.TimeoutExpired:
_killpg(self.pgid, signal.SIGKILL)
try:
self.proc.wait(timeout=STOP_GRACE)
except subprocess.TimeoutExpired:
pass
# nuke any surviving group members (forked workers etc.)
_killpg(self.pgid, signal.SIGKILL)
finally:
with _live_lock:
_live_pgids.discard(self.pgid)
if not self._stderr_f.closed:
self._stderr_f.close()
return self.proc.returncode
def __enter__(self):
return self
def __exit__(self, *exc):
self.stop()
# ---- HTTP helpers ------------------------------------------------
def req(self, method, path, body=None, headers=None):
con = http.client.HTTPConnection("127.0.0.1", self.port,
timeout=HTTP_TIMEOUT)
try:
con.request(method, path, body=body, headers=headers or {})
r = con.getresponse()
return r.status, r.read(), dict(r.getheaders())
finally:
con.close()
def get_json(self, path):
status, body, _ = self.req("GET", path)
assert status == 200, "GET %s -> %s (%r)" % (path, status, body[:200])
return json.loads(body)
def save(self, ns, text, token=TOKEN, ctype="application/json", raw=None):
body = raw if raw is not None else json.dumps({"ns": ns, "text": text}).encode()
headers = {"Content-Type": ctype}
if token is not None:
headers["Authorization"] = "Bearer " + token
return self.req("POST", "/save", body=body, headers=headers)
def query(self, q, k=None):
path = "/query?q=" + urllib.parse.quote(q, safe="")
if k is not None:
path += "&k=%s" % k
return self.get_json(path)
def wait_queryable(self, q, deadline=8.0):
"""Bounded wait for the background drain — never calls drain()."""
t0 = time.monotonic()
while time.monotonic() - t0 < deadline:
hits = self.query(q)
if hits:
return hits
time.sleep(0.1)
raise AssertionError("row never became queryable for q=%r" % q)
def _db_rows(db, sql):
con = sqlite3.connect(db)
try:
return con.execute(sql).fetchall()
finally:
con.close()
# ======================================================================
# (1) save is fast, queue-append only, not synchronously indexed
# ======================================================================
def test_save_is_queue_append_only_and_fast(tmp_path):
"""POST /save returns an int id quickly; the row is NOT immediately
queryable and NOT in memories — it sits in write_queue (status shows it)."""
with Server(tmp_path, interval=LONG) as s:
t0 = time.monotonic()
status, body, _ = s.save("/test/alpha", "the quick brown fox")
elapsed = time.monotonic() - t0
assert status == 200, body
assert elapsed < 1.0, "save took %.2fs — must be queue-append only" % elapsed
rid = json.loads(body)["id"]
assert isinstance(rid, int)
assert s.query("fox") == [] # not indexed yet
assert _db_rows(s.db, "SELECT COUNT(*) FROM memories")[0][0] == 0
st = s.get_json("/status")
assert st["queue_depth"] >= 1
# ======================================================================
# (2) automatic background drain — no manual drain() anywhere
# ======================================================================
def test_background_drain_makes_save_queryable(tmp_path):
"""With --drain-interval 0.2 a saved row becomes queryable by itself;
/status queue_depth returns to 0. This test never triggers a drain."""
with Server(tmp_path, interval=SHORT) as s:
status, body, _ = s.save("/test/beta", "aurora borealis localized")
assert status == 200, body
hits = s.wait_queryable("aurora")
assert hits[0]["text"] == "aurora borealis localized"
t0 = time.monotonic()
while time.monotonic() - t0 < 5:
if s.get_json("/status")["queue_depth"] == 0:
break
time.sleep(0.1)
assert s.get_json("/status")["queue_depth"] == 0
# ======================================================================
# (3) no overlapping / duplicate drains under concurrency
# ======================================================================
def test_concurrent_saves_no_duplicate_or_lost_drains(tmp_path):
"""40 saves from 8 threads while the drain loop fires every 50 ms:
afterwards memories holds exactly 40 rows with 40 distinct ids, the queue
is empty, and the server logged no Traceback/IntegrityError (overlapping
drains + plain INSERT would blow up exactly there)."""
with Server(tmp_path, interval="0.05") as s:
errors = []
def worker(i):
for j in range(5):
st, body, _ = s.save("/test/conc", "row %d-%d payload" % (i, j))
if st != 200:
errors.append((st, body))
threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
for t in threads:
t.start()
for t in threads:
t.join()
assert not errors, errors
t0 = time.monotonic()
while time.monotonic() - t0 < 10:
if s.get_json("/status")["queue_depth"] == 0:
break
time.sleep(0.1)
rows = _db_rows(s.db, "SELECT COUNT(*), COUNT(DISTINCT id) FROM memories")
assert rows[0] == (40, 40), rows
assert _db_rows(s.db, "SELECT COUNT(*) FROM write_queue")[0][0] == 0
log = s.stderr()
assert "Traceback" not in log and "IntegrityError" not in log, log
# ======================================================================
# (4) graceful shutdown performs one final drain (and exits promptly)
# ======================================================================
def test_sigterm_final_drain(tmp_path):
"""Interval is 3600 s, so ONLY a shutdown drain can move these rows.
SIGTERM must drain the queue, then exit well before the interval —
which also proves the loop waits on an interruptible Event, not a
busy-poll or an uninterruptible sleep."""
s = Server(tmp_path, interval=LONG)
try:
for i in range(3):
st, body, _ = s.save("/test/final", "shutdown row %d" % i)
assert st == 200, body
assert s.get_json("/status")["queue_depth"] == 3
finally:
rc = s.stop(signal.SIGTERM, timeout=10)
assert rc == 0, "SIGTERM exit code %s, stderr=%r" % (rc, s.stderr())
assert _db_rows(s.db, "SELECT COUNT(*) FROM memories")[0][0] == 3
assert _db_rows(s.db, "SELECT COUNT(*) FROM write_queue")[0][0] == 0
# ======================================================================
# (5) auth on POST /save — fail-closed, bearer, constant-time; reads open
# ======================================================================
def test_save_without_configured_token_is_503(tmp_path):
with Server(tmp_path, token=None) as s:
status, _, _ = s.save("/test/auth", "nope", token=TOKEN)
assert status == 503
def test_save_bad_token_is_401(tmp_path):
with Server(tmp_path) as s:
status, _, hdrs = s.save("/test/auth", "nope", token="wrong-token")
assert status == 401
assert "bearer" in (hdrs.get("WWW-Authenticate") or "").lower()
status, _, _ = s.save("/test/auth", "nope", token=None) # no header
assert status == 401
def test_save_good_token_from_env_is_200(tmp_path):
with Server(tmp_path) as s:
status, body, _ = s.save("/test/auth", "yes", token=TOKEN)
assert status == 200, body
assert isinstance(json.loads(body)["id"], int)
def test_save_token_from_file(tmp_path):
"""No MNEME_TOKEN env; token read from ~/.config/mneme/token (comments and
blank lines skipped), exactly like bullseye's _post_token."""
filetok = "file-t0ken-xyz"
with Server(tmp_path, token=None,
token_file="# comment\n\n%s\n" % filetok) as s:
assert s.save("/test/auth", "no", token=TOKEN)[0] == 401
assert s.save("/test/auth", "yes", token=filetok)[0] == 200
def test_reads_stay_open_without_token(tmp_path):
"""GET endpoints need no auth even when no token is configured."""
with Server(tmp_path, token=None) as s:
for path in ("/query?q=x", "/namespaces", "/recent?n=5",
"/keywords?ns=%2Ftest", "/status", "/healthz"):
status, body, _ = s.req("GET", path)
assert status == 200, "%s -> %s (%r)" % (path, status, body[:200])
def test_server_source_uses_constant_time_compare():
src = SERVER_PY.read_text()
assert "compare_digest" in src, \
"server.py must compare bearer tokens with hmac.compare_digest"
# ======================================================================
# (6) hostile FTS5 metacharacters via HTTP never 500
# ======================================================================
HOSTILE_Q = [
'"', 'a" OR "b', 'NEAR(a, 1)', 'text: foo', '*', '-term',
'(((', 'a AND', 'x OR', 'col:*', '"unterminated',
'éü ☃ " AND (SELECT 1)', 'a;DROP TABLE memories;--',
]
def test_hostile_query_params_do_not_500(tmp_path):
with Server(tmp_path, interval=SHORT) as s:
assert s.save("/test/fts", "benign haystack row")[0] == 200
s.wait_queryable("haystack")
for q in HOSTILE_Q:
path = "/query?q=" + urllib.parse.quote(q, safe="")
status, body, _ = s.req("GET", path)
assert status < 500, "q=%r -> %s (%r)" % (q, status, body[:200])
if status == 200:
json.loads(body) # must still be valid JSON
for k in ("abc", "-1", "999999", ""):
status, body, _ = s.req("GET", "/query?q=haystack&k=" + k)
assert status < 500, "k=%r -> %s" % (k, status)
# ======================================================================
# (7) malformed JSON body -> 4xx (8) oversized body -> 413
# ======================================================================
def test_malformed_json_body_is_4xx(tmp_path):
with Server(tmp_path) as s:
status, _, _ = s.save(None, None, raw=b'{"ns": "/x", txt')
assert 400 <= status < 500
status, _, _ = s.save(None, None, raw=b"")
assert 400 <= status < 500
# wrong content-type is refused too (bullseye pattern: 415)
status, _, _ = s.save("/test/x", "y", ctype="text/plain")
assert status == 415
def test_oversized_body_rejected_413(tmp_path):
"""Declared Content-Length far over the 1 MiB cap must be rejected from
the headers alone (413) — the server must not try to read it all."""
with Server(tmp_path) as s:
con = http.client.HTTPConnection("127.0.0.1", s.port,
timeout=HTTP_TIMEOUT)
try:
con.putrequest("POST", "/save")
con.putheader("Authorization", "Bearer " + TOKEN)
con.putheader("Content-Type", "application/json")
con.putheader("Content-Length", "5000000")
con.endheaders()
con.send(b"x" * 1024) # only a sliver actually sent
r = con.getresponse()
assert r.status == 413, r.status
finally:
con.close()
# ======================================================================
# (9) /healthz is cheap liveness — no DB write, no DB connection
# ======================================================================
def test_healthz_writes_nothing_to_db(tmp_path):
def snap(db):
out = {}
for suf in ("", "-wal", "-shm"):
p = Path(db + suf)
out[suf] = hashlib.sha256(p.read_bytes()).hexdigest() if p.exists() else None
return out
with Server(tmp_path, interval=LONG) as s:
time.sleep(0.3) # let startup fully settle
before = snap(s.db)
for _ in range(3):
status, _, _ = s.req("GET", "/healthz")
assert status == 200
assert snap(s.db) == before, "/healthz touched the database files"
# ======================================================================
# /status shape + read endpoints shape (incl. /similar)
# ======================================================================
def test_status_reports_degraded_and_queue(tmp_path):
with Server(tmp_path, interval=LONG) as s:
assert s.save("/test/st", "one")[0] == 200
assert s.save("/test/st", "two")[0] == 200
st = s.get_json("/status")
assert isinstance(st["degraded"], bool)
assert st["queue_depth"] == 2
assert isinstance(st["queue_age"], (int, float)) and st["queue_age"] >= 0
def test_read_endpoints_shapes(tmp_path):
with Server(tmp_path, interval=SHORT) as s:
status, body, _ = s.save("/test/shapes", "gamma delta epsilon")
assert status == 200
rid = json.loads(body)["id"]
s.wait_queryable("gamma")
ns = s.get_json("/namespaces")
assert isinstance(ns, dict) and "/test/shapes" in ns
assert ns["/test/shapes"]["count"] == 1
recent = s.get_json("/recent?n=5")
assert isinstance(recent, list) and recent[0]["id"] == rid
kw = s.get_json("/keywords?ns=%2Ftest%2Fshapes")
assert isinstance(kw, list) and all(isinstance(w, str) for w in kw)
sim = s.get_json("/similar?to=%d" % rid)
assert isinstance(sim, list)
# ======================================================================
# store.py v1.1 — UNINDEXED behaviour + source-level notes
# ======================================================================
def test_query_does_not_match_ns_tokens(tmp_path):
"""id/ns are UNINDEXED in memories_fts: tokens that appear only in the
namespace path must not be searchable."""
with Server(tmp_path, interval=SHORT) as s:
assert s.save("/zebra/notes", "hello world")[0] == 200
s.wait_queryable("hello") # proves the row IS indexed
assert s.query("zebra") == []
assert s.query("notes") == []
def test_store_drain_uses_plain_insert():
src = STORE_PY.read_text()
assert "INSERT OR REPLACE" not in src.upper().replace(" ", " "), \
"v1.1: drain must use plain INSERT so overlapping drains error loudly"
def test_store_fts_id_ns_unindexed():
src = STORE_PY.read_text()
assert re.search(r"\bid\s+UNINDEXED", src), "fts5: id must be UNINDEXED"
assert re.search(r"\bns\s+UNINDEXED", src), "fts5: ns must be UNINDEXED"
def test_store_connections_explicitly_closed():
"""`with sqlite3.connect(...)` commits but never closes; v1.1 requires an
explicit .close() (or contextlib.closing) for per-batch connections."""
src = STORE_PY.read_text()
assert ".close()" in src or "closing(" in src, \
"store.py never closes its sqlite connections"
Reference in New Issue View Git Blame Copy Permalink