chore: import upstream snapshot with attribution
This commit is contained in:
wehub-resource-sync
@@ -0,0 +1,186 @@
|
||||
r"""GREEN regression guard — the UI directory picker enumerates all logical drives.
|
||||
|
||||
Guards the fix for issue #548 (landed on main). `handle_browse`
|
||||
(src/ui/http_server.c) appends a `"roots"` array to every `/api/browse`
|
||||
response; on Windows `append_roots_json` fills it from `GetLogicalDrives()` as
|
||||
`"C:/"`, `"D:/"`, … so the directory picker can reach every drive (POSIX emits
|
||||
a single `"/"`). This test asserts that user-level invariant against the running
|
||||
embedded HTTP UI.
|
||||
|
||||
Before #548 the picker did `opendir("/")`, listing only the current drive's
|
||||
subdirectories under `dirs` with no drive enumeration — non-system drives were
|
||||
unreachable. The original red test asserted drives appeared in `dirs`; the fix
|
||||
intentionally exposes them via the separate `roots` field, so this guard checks
|
||||
`roots` (and that each advertised drive is actually browsable).
|
||||
|
||||
Requires a UI build (`make -f Makefile.cbm cbm-with-ui`) because the HTTP server
|
||||
only starts when the frontend is embedded. Runs green with a single drive (C:/
|
||||
must be advertised and browsable); a machine with D:/E: exercises the multi-drive
|
||||
reach more fully.
|
||||
|
||||
Exit code: 0 == all drives advertised in roots and reachable (green),
|
||||
1 == a drive is missing from roots or not browsable (regression),
|
||||
2 == precondition not met (no UI build / server down).
|
||||
|
||||
Usage:
|
||||
python test_ui_drive_listing.py <path-to-codebase-memory-mcp-ui[.exe]> [port]
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
|
||||
def list_fixed_drives():
|
||||
# Python 3.12+: os.listdrives(). Fall back to scanning A:..Z:.
|
||||
listdrives = getattr(os, "listdrives", None)
|
||||
if listdrives:
|
||||
try:
|
||||
return [d for d in listdrives()]
|
||||
except Exception:
|
||||
pass
|
||||
found = []
|
||||
for ch in "CDEFGHIJKLMNOPQRSTUVWXYZ":
|
||||
root = "%s:\\" % ch
|
||||
if os.path.isdir(root):
|
||||
found.append(root)
|
||||
return found
|
||||
|
||||
|
||||
def free_port():
|
||||
s = socket.socket()
|
||||
s.bind(("127.0.0.1", 0))
|
||||
p = s.getsockname()[1]
|
||||
s.close()
|
||||
return p
|
||||
|
||||
|
||||
def http_get_json(url, timeout=5):
|
||||
with urllib.request.urlopen(url, timeout=timeout) as r:
|
||||
return json.loads(r.read().decode("utf-8", "replace"))
|
||||
|
||||
|
||||
def browse(port, path):
|
||||
return http_get_json("http://127.0.0.1:%d/api/browse?path=%s" %
|
||||
(port, urllib.parse.quote(path)))
|
||||
|
||||
|
||||
def wait_for_server(port, timeout=20):
|
||||
deadline = time.time() + timeout
|
||||
while time.time() < deadline:
|
||||
try:
|
||||
with socket.create_connection(("127.0.0.1", port), timeout=1):
|
||||
return True
|
||||
except OSError:
|
||||
time.sleep(0.3)
|
||||
return False
|
||||
|
||||
|
||||
def norm(s):
|
||||
"""Canonical drive key: 'D:\\', 'D:/', 'D:', 'D' -> 'D:'."""
|
||||
s = str(s).rstrip("\\/").upper()
|
||||
return s if s.endswith(":") else (s + ":" if len(s) == 1 else s)
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) < 2:
|
||||
print("usage: python test_ui_drive_listing.py <ui-binary> [port]")
|
||||
return 2
|
||||
binary = os.path.abspath(sys.argv[1])
|
||||
if not os.path.exists(binary):
|
||||
print("FAIL: binary not found: %s" % binary)
|
||||
return 2
|
||||
|
||||
drives = list_fixed_drives()
|
||||
print("fixed drives: %s" % drives)
|
||||
if not drives:
|
||||
print("PRECONDITION: no fixed drives detected.")
|
||||
return 2
|
||||
|
||||
work = tempfile.mkdtemp(prefix="cbm_win_uidrv_")
|
||||
port = int(sys.argv[2]) if len(sys.argv) > 2 else free_port()
|
||||
env = dict(os.environ)
|
||||
env["CBM_CACHE_DIR"] = os.path.join(work, "cache")
|
||||
os.makedirs(env["CBM_CACHE_DIR"], exist_ok=True)
|
||||
proc = subprocess.Popen([binary, "--ui=true", "--port=%d" % port],
|
||||
stdin=subprocess.PIPE, stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE, env=env)
|
||||
try:
|
||||
if not wait_for_server(port, timeout=25):
|
||||
print("PRECONDITION: HTTP server did not start on port %d. Is this a "
|
||||
"UI build (make -f Makefile.cbm cbm-with-ui)?" % port)
|
||||
return 2
|
||||
|
||||
# Control: browsing an explicit existing directory must return a payload
|
||||
# carrying the roots field, proving the endpoint works and isolating any
|
||||
# failure to drive enumeration itself. roots is appended to *every*
|
||||
# browse response, so any valid directory surfaces it.
|
||||
home = os.environ.get("USERPROFILE") or os.path.expanduser("~")
|
||||
home_fwd = home.replace("\\", "/")
|
||||
try:
|
||||
ctrl = browse(port, home_fwd)
|
||||
except Exception as ex:
|
||||
print("PRECONDITION: control /api/browse?path=%s failed: %r" %
|
||||
(home_fwd, ex))
|
||||
return 2
|
||||
roots = ctrl.get("roots")
|
||||
print("control browse(%r) -> dirs(%d) roots=%s" %
|
||||
(home_fwd, len(ctrl.get("dirs", [])), roots))
|
||||
if roots is None:
|
||||
print("PRECONDITION: response has no 'roots' field; build predates "
|
||||
"#548 or endpoint non-functional.")
|
||||
return 2
|
||||
|
||||
# Invariant 1: every fixed drive is advertised in roots.
|
||||
adv = {norm(r) for r in roots}
|
||||
missing = [d for d in drives if norm(d) not in adv]
|
||||
if missing:
|
||||
print("\nRED: drives %s are not advertised in the picker's roots "
|
||||
"array %s (handle_browse/append_roots_json did not enumerate "
|
||||
"them)." % (missing, roots))
|
||||
return 1
|
||||
|
||||
# Invariant 2: every advertised drive is actually browsable (a user can
|
||||
# select it). This is the user-level reach the fix promises.
|
||||
unreachable = []
|
||||
for d in drives:
|
||||
drive_root = norm(d) + "/" # "D:/"
|
||||
try:
|
||||
resp = browse(port, drive_root)
|
||||
if resp.get("roots") is None and "dirs" not in resp:
|
||||
unreachable.append(d)
|
||||
except Exception as ex:
|
||||
print(" browse(%r) failed: %r" % (drive_root, ex))
|
||||
unreachable.append(d)
|
||||
if unreachable:
|
||||
print("\nRED: drives advertised in roots but not browsable: %s" %
|
||||
unreachable)
|
||||
return 1
|
||||
|
||||
print("\nGREEN: all %d fixed drive(s) advertised in roots and reachable "
|
||||
"from the UI picker." % len(drives))
|
||||
return 0
|
||||
finally:
|
||||
try:
|
||||
proc.stdin.close()
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
proc.terminate()
|
||||
proc.wait(timeout=5)
|
||||
except Exception:
|
||||
try:
|
||||
proc.kill()
|
||||
except Exception:
|
||||
pass
|
||||
shutil.rmtree(work, ignore_errors=True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user