jason
113 lines
5.4 KiB
Python
113 lines
5.4 KiB
Python
#!/usr/bin/env python3
|
|
"""test_v1_scaffold.py — H4 interface guard for the v1.0 scaffolds. [offline, no creds]
|
|
|
|
Asserts the scaffold contract holds so the skeleton can't silently rot while 1.0 is
|
|
built: every planned module imports, the parts that ARE implemented work (BM25 ranking,
|
|
queue/cache round-trip, slug helpers, secret resolution order), and the parts that are
|
|
TODO raise NotImplementedError (so a stub can't masquerade as done).
|
|
|
|
Run: python test_v1_scaffold.py
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
|
|
failures: list[str] = []
|
|
|
|
|
|
def check(name: str, cond: bool, detail: str = "") -> None:
|
|
print(f"{'ok ' if cond else 'FAIL'} {name}" + (f" -- {detail}" if not cond else ""))
|
|
if not cond:
|
|
failures.append(name)
|
|
|
|
|
|
def main() -> int:
|
|
# H1 — BM25 core is implemented; assert it actually ranks a relevant doc first.
|
|
import echo_recall as r
|
|
ix = r.Bm25Index()
|
|
ix.add("notes/mqtt.md", "the mqtt broker config and tls certificate rotation")
|
|
ix.add("notes/payroll.md", "quarterly payroll run and employee deductions")
|
|
top = ix.score("certificate rotation broker", limit=2)
|
|
check("H1 BM25 ranks relevant doc first", bool(top) and top[0][0] == "notes/mqtt.md", str(top))
|
|
check("H1 BM25 survives json round-trip",
|
|
r.Bm25Index.from_json(ix.to_json()).score("payroll")[0][0] == "notes/payroll.md")
|
|
ix.remove("notes/payroll.md")
|
|
check("H1 BM25 remove() drops a doc's postings", ix.score("payroll") == [])
|
|
check("H1 recall path is wired (not a stub)",
|
|
all(callable(getattr(r, n, None))
|
|
for n in ("recall", "rebuild", "update_note", "load_index", "save_index")))
|
|
|
|
# H2 — cache round-trips, enqueue dedups, flush is a no-op on an empty queue (offline-safe).
|
|
import echo_queue as q
|
|
with tempfile.TemporaryDirectory() as d:
|
|
os.environ["ECHO_STATE_DIR"] = d
|
|
q.cache_put("a/b.md", b"hello")
|
|
check("H2 cache round-trip", q.cache_get("a/b.md") == b"hello")
|
|
q.enqueue("PUT", "http://x/vault/a.md", b"body", {}, idem_key="k1")
|
|
check("H2 enqueue persists a record", len(q.pending()) == 1)
|
|
q.enqueue("PUT", "http://x/vault/a.md", b"body", {}, idem_key="k1")
|
|
check("H2 enqueue dedups by idem_key", len(q.pending()) == 1)
|
|
with tempfile.TemporaryDirectory() as d2:
|
|
os.environ["ECHO_STATE_DIR"] = d2
|
|
check("H2 flush() returns 0 on an empty queue (no network)", q.flush() == 0)
|
|
|
|
# H3 — lock CM + atomic index update are stubs.
|
|
import echo_concurrency as c
|
|
check("H3 auto_owner is stable", c.auto_owner() == c.auto_owner())
|
|
check("H3 vault_lock returns a context manager (no network until entered)",
|
|
hasattr(c.vault_lock(), "__enter__") and hasattr(c.vault_lock(), "__exit__"))
|
|
check("H3 atomic_index_update is wired", callable(c.atomic_index_update))
|
|
|
|
# H5 — proposal validation is pure (offline): keeps valid (incl. inbox), drops the rest.
|
|
import echo_reflect as rf
|
|
v, errs = rf.validate([
|
|
{"title": "Ok", "kind": "person", "confidence": 0.9},
|
|
{"title": "", "kind": "person"}, # missing title
|
|
{"title": "Bad", "kind": "bogus"}, # unknown kind
|
|
{"title": "Weak", "kind": "person", "confidence": 0.2}, # below confidence floor
|
|
{"title": "Note", "inbox": True}, # inbox needs no kind
|
|
])
|
|
check("H5 validate keeps valid (incl. inbox), drops invalid", len(v) == 2 and len(errs) == 3, (len(v), errs))
|
|
check("H5 classify/preview/apply wired",
|
|
all(callable(getattr(rf, n, None)) for n in ("classify", "preview", "apply")))
|
|
|
|
# M1 — secret resolution order is implemented; env wins, then the credentials file.
|
|
import echo_secrets as s
|
|
os.environ["ECHO_KEY"] = "env-key-123"
|
|
check("M1 env ECHO_KEY wins", s.resolve_key("legacy-fallback") == "env-key-123")
|
|
with tempfile.TemporaryDirectory() as d:
|
|
os.environ["ECHO_STATE_DIR"] = d
|
|
os.environ.pop("ECHO_KEY", None)
|
|
s.write_key("file-key-xyz")
|
|
check("M1 write_key -> credentials -> resolve_key reads it", s.resolve_key() == "file-key-xyz")
|
|
os.environ["ECHO_KEY_LEGACY_OK"] = "1" # silence the warning for the legacy-path check
|
|
os.environ["ECHO_STATE_DIR"] = os.path.join(d, "empty") # an empty dir: no creds file
|
|
check("M1 legacy fallback when env+file absent", s.resolve_key("legacy-fallback") == "legacy-fallback")
|
|
os.environ["ECHO_KEY"] = "env-key-123" # restore for any later checks
|
|
|
|
# M2 — slug collision + link-confidence helpers are implemented.
|
|
import echo_quality as qual
|
|
check("M2 safe_slug disambiguates a collision",
|
|
qual.safe_slug({"foo"}, "foo") != "foo")
|
|
check("M2 short/common tokens are not confident links",
|
|
qual.is_confident_link("API", "we built an api") is False)
|
|
|
|
# M3 — doctor module exposes run(); M4 — output envelope is implemented.
|
|
import echo_doctor as doc
|
|
check("M3 doctor exposes run()", hasattr(doc, "run"))
|
|
import echo_output as out
|
|
env = out.envelope("created", {"path": "p.md"})
|
|
check("M4 envelope shape", env.get("ok") is True and env.get("action") == "created")
|
|
|
|
print(f"\n{len(failures)} failure(s)" if failures else "\nall scaffold interface checks passed")
|
|
return 1 if failures else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|