tests/_fake_opencode.py implements the two commands MAM calls (debug paths, session delete) as measured on 2.0.32; unit tests never run a real opencode (it would migrate the test database). test_real_opencode_2x_contract runs a real 2.x opencode, when installed, in a home and tmp dir of its own with --standalone. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff
65 lines
2.6 KiB
Python
65 lines
2.6 KiB
Python
"""A fake opencode 2.x CLI for tests (Issue #60): only what MAM calls.
|
|
|
|
- `debug paths` prints the `data` and `db` lines of the real command (the db
|
|
follows XDG_DATA_HOME, else HOME/.local/share; the database is not created);
|
|
- `session delete [--standalone] [--] <id>` deletes the session and its child
|
|
sessions from the 2.x tables (session_v2, session_message, session_inbox,
|
|
session_pending, event, event_sequence) and leaves the tables a 1.x
|
|
migration keeps (session, message, part, todo, session_share), as the real
|
|
2.0.32 command was measured to do; prints `Session <id> deleted`, or
|
|
`Session not found: <id>` with exit 1.
|
|
|
|
Tests never run a real opencode: a real one would migrate the test's
|
|
database. The contract against a real binary is checked by the opt-in test in
|
|
tests/test_issue60_opencode_v2.py.
|
|
"""
|
|
import os
|
|
import stat
|
|
|
|
SCRIPT = r'''#!/usr/bin/env python3
|
|
import os, sqlite3, sys
|
|
args = sys.argv[1:]
|
|
data = os.environ.get("XDG_DATA_HOME") or os.path.join(os.environ.get("HOME", ""), ".local", "share")
|
|
db = os.path.join(data, "opencode", "opencode.db")
|
|
if args == ["debug", "paths"]:
|
|
print("data " + os.path.dirname(db))
|
|
print("db " + db)
|
|
sys.exit(0)
|
|
if args[:2] == ["session", "delete"]:
|
|
sid = args[-1]
|
|
conn = sqlite3.connect(db)
|
|
try:
|
|
if not conn.execute("SELECT 1 FROM session_v2 WHERE id=?", (sid,)).fetchone():
|
|
print("Session not found: " + sid, file=sys.stderr)
|
|
sys.exit(1)
|
|
ids, todo = [], [sid]
|
|
while todo:
|
|
cur = todo.pop()
|
|
ids.append(cur)
|
|
todo += [r[0] for r in conn.execute("SELECT id FROM session_v2 WHERE parent_id=?", (cur,))]
|
|
tables = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")}
|
|
v2 = (("session_message", "session_id"), ("session_inbox", "session_id"),
|
|
("session_pending", "session_id"), ("event", "aggregate_id"),
|
|
("event_sequence", "aggregate_id"), ("session_v2", "id"))
|
|
for t, col in v2:
|
|
if t in tables:
|
|
for i in ids:
|
|
conn.execute('DELETE FROM "%s" WHERE "%s"=?' % (t, col), (i,))
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
print("Session " + sid + " deleted")
|
|
sys.exit(0)
|
|
sys.exit(0)
|
|
'''
|
|
|
|
|
|
def write(path) -> str:
|
|
"""Write the fake as an executable `path` and return it as a str."""
|
|
path = str(path)
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, "w") as f:
|
|
f.write(SCRIPT)
|
|
os.chmod(path, os.stat(path).st_mode | stat.S_IEXEC)
|
|
return path
|