engineering/arborist
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
arborist/tests/test_cli_session.py
russell@unturf.com fea761c577
tests: unit + integration + functional coverage for this session's surface 
51 new tests across the three layers (unit / integration / functional)
for tickets #000014/#000015/#000017/#000020/#000021/#000023/#000024/
#000025/#000019.

NEW FILES

tests/test_cli_session.py (17 tests) — functional CLI coverage:
  arborist selfmodel snapshot|show|show --root|falsify|falsify-idempotent|list
  arborist memory     snapshot|show|branches|falsify
  arborist capital    summary|summary --op-type|op-cost|top|top --rejects-unknown-form
  + audit chain stays clean across all three CLI families

tests/test_session_migrations.py (7 tests) — schema migration semantics:
  fresh-db has all five new tables
  re-connect is idempotent
  explicit migration helpers re-apply without error
  PRAGMA table_info confirms expected columns
  CHECK constraints reject invalid state values
  audit chain re-verifies after writes from all three modules
  capital_ledger writes do not chain into audit_events (sibling invariant)

tests/test_session_integration.py (11 tests) — cross-module flows:
  ingest emits one capital_ledger row per batch tied to last event hash
  SelfModel.snapshot folds memory_root from memory_records when present
  SelfModel.snapshot returns memory_root=None on empty memory table
  π* registry rejects conflicting registration (name@version pinned)
  π* registry tolerates same-instance re-registration
  pi_star.get raises KeyError on unknown
  Battery runtime_digest fingerprint shifts when registry changes
  Full Dav1DPrometheus suite via runner --all returns 0; 312 fixtures
  _DEFAULT_FIXTURES sums to 312 deterministic tasks
  Phase 1a fixture digests stay byte-stable
  Full state-space round-trip: ingest → SelfModel + Memory + Capital

EXTENDED FILES

tests/test_pi_star.py (+6 tests):
  assert_round_trip passes on idempotent / raises on non-idempotent π*
  equivalence_class_id determinism + input sensitivity
  registry_key format
  domains() partitioning invariant

tests/test_bench_batteries.py (+10 tests):
  _eval_propositional parens nesting
  _eval_propositional rejects unknown variable + malformed
  _eval_propositional XOR/IMPL/IFF truth-table coverage
  _walk_relation_path: self-loop, cycles without infinite-loop, unreachable
  _walk_relation_path rejects non-whitelisted relation
  _content_tokens strips punctuation, handles unicode
  _capital_cost_delta handles missing/empty budget

Full suite: 1161 passed, 36 skipped. Up from 1110.
2026-05-07 21:15:14 -04:00

360 lines
12 KiB
Python
Raw Permalink Blame History

"""CLI subcommand integration tests for this session's new work.
Covers `arborist selfmodel|memory|capital` subcommand surfaces end-to-
end through ``build_parser`` + ``args.func(args)`` (matches existing
arborist CLI-test pattern in ``tests/test_burn.py``).
This file is the *functional* layer for tickets #000014 (SelfModel),
#000017 (Memory), #000020 (Capital). Unit-level tests for the
underlying modules live in ``tests/test_selfmodel.py``,
``tests/test_memory_root.py``, ``tests/test_capital.py``.
"""
from __future__ import annotations
import json
import pytest
from arborist.cli import build_parser
from arborist.store import append_audit, connect, transaction
def _run(parser, argv: list[str]) -> int:
"""Run an argv through the CLI parser and dispatch table."""
args = parser.parse_args(argv)
return args.func(args)
# ---------------------------------------------------------------------
# selfmodel CLI
# ---------------------------------------------------------------------
def test_selfmodel_snapshot_cli_writes_record(tmp_path, capsys):
db = tmp_path / "shard.db"
parser = build_parser()
rc = _run(parser, ["--db", str(db), "selfmodel", "snapshot"])
assert rc == 0
payload = json.loads(capsys.readouterr().out)
root = payload["selfmodel_root"]
assert len(root) == 64
# Record present in the database.
conn = connect(db)
try:
row = conn.execute(
"SELECT state FROM selfmodel_records WHERE selfmodel_root = ?",
(root,),
).fetchone()
assert row is not None
assert row["state"] == "live"
finally:
conn.close()
def test_selfmodel_show_cli_returns_full_record(tmp_path, capsys):
db = tmp_path / "shard.db"
parser = build_parser()
_run(parser, ["--db", str(db), "selfmodel", "snapshot"])
capsys.readouterr() # drain
rc = _run(parser, ["--db", str(db), "selfmodel", "show"])
assert rc == 0
payload = json.loads(capsys.readouterr().out)
assert "selfmodel_root" in payload
assert payload["state"] == "live"
assert payload["schema_version"] == "selfmodel-v1"
assert "claims" in payload # at minimum empty list
def test_selfmodel_show_cli_specific_root(tmp_path, capsys):
db = tmp_path / "shard.db"
parser = build_parser()
_run(parser, ["--db", str(db), "selfmodel", "snapshot"])
snapshot_payload = json.loads(capsys.readouterr().out)
root = snapshot_payload["selfmodel_root"]
rc = _run(parser, ["--db", str(db), "selfmodel", "show", "--root", root])
assert rc == 0
show_payload = json.loads(capsys.readouterr().out)
assert show_payload["selfmodel_root"] == root
def test_selfmodel_show_cli_returns_error_on_missing(tmp_path, capsys):
db = tmp_path / "shard.db"
parser = build_parser()
bogus = "00" * 32
rc = _run(parser, ["--db", str(db), "selfmodel", "show", "--root", bogus])
assert rc == 1
payload = json.loads(capsys.readouterr().out)
assert payload["error"] == "no SelfModel found"
def test_selfmodel_falsify_cli_flips_state(tmp_path, capsys):
db = tmp_path / "shard.db"
parser = build_parser()
_run(parser, ["--db", str(db), "selfmodel", "snapshot"])
snapshot_payload = json.loads(capsys.readouterr().out)
root = snapshot_payload["selfmodel_root"]
rc = _run(parser, [
"--db", str(db),
"selfmodel", "falsify", root,
"--reason", "verifier upgrade",
])
assert rc == 0
falsify_payload = json.loads(capsys.readouterr().out)
assert falsify_payload["selfmodel_root"] == root
assert falsify_payload["audit_event_hash"] is not None
assert falsify_payload["noop"] is False
# State row is now falsified.
conn = connect(db)
try:
row = conn.execute(
"SELECT state, falsified_reason FROM selfmodel_records "
"WHERE selfmodel_root = ?", (root,),
).fetchone()
assert row["state"] == "falsified"
assert row["falsified_reason"] == "verifier upgrade"
finally:
conn.close()
def test_selfmodel_falsify_cli_idempotent(tmp_path, capsys):
db = tmp_path / "shard.db"
parser = build_parser()
_run(parser, ["--db", str(db), "selfmodel", "snapshot"])
root = json.loads(capsys.readouterr().out)["selfmodel_root"]
_run(parser, ["--db", str(db), "selfmodel", "falsify", root, "--reason", "r1"])
capsys.readouterr()
rc = _run(parser, ["--db", str(db), "selfmodel", "falsify", root, "--reason", "r2"])
assert rc == 0
payload = json.loads(capsys.readouterr().out)
assert payload["noop"] is True
assert payload["audit_event_hash"] is None
def test_selfmodel_list_cli_returns_recent(tmp_path, capsys):
db = tmp_path / "shard.db"
parser = build_parser()
# Ensure at least one record exists.
_run(parser, ["--db", str(db), "selfmodel", "snapshot"])
capsys.readouterr()
rc = _run(parser, ["--db", str(db), "selfmodel", "list", "--limit", "5"])
assert rc == 0
payload = json.loads(capsys.readouterr().out)
assert isinstance(payload, list)
assert len(payload) >= 1
assert payload[0]["state"] == "live"
# ---------------------------------------------------------------------
# memory CLI
# ---------------------------------------------------------------------
def test_memory_snapshot_cli_writes_record(tmp_path, capsys):
db = tmp_path / "shard.db"
parser = build_parser()
rc = _run(parser, ["--db", str(db), "memory", "snapshot"])
assert rc == 0
payload = json.loads(capsys.readouterr().out)
root = payload["memory_root"]
assert len(root) == 64
conn = connect(db)
try:
row = conn.execute(
"SELECT state, schema_version FROM memory_records "
"WHERE memory_root = ?", (root,),
).fetchone()
assert row is not None
assert row["state"] == "live"
assert row["schema_version"] == "memory-v1"
finally:
conn.close()
def test_memory_show_cli_returns_branches(tmp_path, capsys):
db = tmp_path / "shard.db"
parser = build_parser()
_run(parser, ["--db", str(db), "memory", "snapshot"])
capsys.readouterr()
rc = _run(parser, ["--db", str(db), "memory", "show"])
assert rc == 0
payload = json.loads(capsys.readouterr().out)
assert payload["state"] == "live"
branch_ids = {b["branch_id"] for b in payload["branches"]}
# Default projection set per ticket #000017.
assert "audit-mode-distribution" in branch_ids
assert "falsification-state" in branch_ids
def test_memory_branches_cli_lists_summaries(tmp_path, capsys):
db = tmp_path / "shard.db"
parser = build_parser()
_run(parser, ["--db", str(db), "memory", "snapshot"])
capsys.readouterr()
rc = _run(parser, ["--db", str(db), "memory", "branches"])
assert rc == 0
payload = json.loads(capsys.readouterr().out)
assert isinstance(payload, list)
branch_ids = {b["branch_id"] for b in payload}
assert "audit-mode-distribution" in branch_ids
def test_memory_falsify_cli(tmp_path, capsys):
db = tmp_path / "shard.db"
parser = build_parser()
_run(parser, ["--db", str(db), "memory", "snapshot"])
snap = json.loads(capsys.readouterr().out)
root = snap["memory_root"]
rc = _run(parser, [
"--db", str(db), "memory", "falsify", root, "--reason", "drift detected",
])
assert rc == 0
payload = json.loads(capsys.readouterr().out)
assert payload["audit_event_hash"] is not None
assert payload["noop"] is False
conn = connect(db)
try:
row = conn.execute(
"SELECT state, falsified_reason FROM memory_records WHERE memory_root = ?",
(root,),
).fetchone()
assert row["state"] == "falsified"
assert row["falsified_reason"] == "drift detected"
finally:
conn.close()
# ---------------------------------------------------------------------
# capital CLI
# ---------------------------------------------------------------------
def _seed_capital_rows(db_path) -> None:
"""Write a few capital_ledger rows tied to real audit events."""
from arborist.capital import CapitalProfile, record
conn = connect(db_path)
try:
with transaction(conn):
ev_ingest = append_audit(conn, event_type="ingest",
subject_root=None, body={"k": "v"})
record(conn, audit_event_hash=ev_ingest, op_type="ingest",
profile=CapitalProfile(material=2.0, intellectual=0.05))
ev_qa = append_audit(conn, event_type="qa",
subject_root=None, body={"k": "v"})
record(conn, audit_event_hash=ev_qa, op_type="qa",
profile=CapitalProfile(financial=0.01, experiential=0.5))
ev_distill = append_audit(conn, event_type="distill",
subject_root=None, body={"k": "v"})
record(conn, audit_event_hash=ev_distill, op_type="distill",
profile=CapitalProfile(material=1.5, intellectual=0.02))
finally:
conn.close()
def test_capital_summary_cli_full(tmp_path, capsys):
db = tmp_path / "shard.db"
_seed_capital_rows(db)
parser = build_parser()
rc = _run(parser, ["--db", str(db), "capital", "summary"])
assert rc == 0
payload = json.loads(capsys.readouterr().out)
assert payload["row_count"] == 3
assert payload["material"] == pytest.approx(3.5)
assert payload["financial"] == pytest.approx(0.01)
def test_capital_summary_cli_filter_op_type(tmp_path, capsys):
db = tmp_path / "shard.db"
_seed_capital_rows(db)
parser = build_parser()
rc = _run(parser, [
"--db", str(db), "capital", "summary", "--op-type", "ingest",
])
assert rc == 0
payload = json.loads(capsys.readouterr().out)
assert payload["row_count"] == 1
assert payload["material"] == pytest.approx(2.0)
def test_capital_op_cost_cli(tmp_path, capsys):
db = tmp_path / "shard.db"
_seed_capital_rows(db)
parser = build_parser()
rc = _run(parser, ["--db", str(db), "capital", "op-cost", "qa"])
assert rc == 0
payload = json.loads(capsys.readouterr().out)
assert payload["row_count"] == 1
assert payload["financial"] == pytest.approx(0.01)
assert payload["experiential"] == pytest.approx(0.5)
def test_capital_top_cli(tmp_path, capsys):
db = tmp_path / "shard.db"
_seed_capital_rows(db)
parser = build_parser()
rc = _run(parser, [
"--db", str(db), "capital", "top", "--form", "material", "--limit", "10",
])
assert rc == 0
payload = json.loads(capsys.readouterr().out)
assert isinstance(payload, list)
op_types = [r["op_type"] for r in payload]
# ingest=2.0, distill=1.5, qa=0 → ingest first
assert op_types[0] == "ingest"
assert op_types[1] == "distill"
def test_capital_top_cli_rejects_unknown_form(tmp_path, capsys):
db = tmp_path / "shard.db"
parser = build_parser()
with pytest.raises(SystemExit):
# argparse rejects on the choices list before the handler runs.
_run(parser, [
"--db", str(db), "capital", "top", "--form", "imaginary_form",
])
# ---------------------------------------------------------------------
# Audit chain stays intact across all three CLI families
# ---------------------------------------------------------------------
def test_chain_stays_clean_across_session_clis(tmp_path, capsys):
"""All three CLI families chain through ``store.append_audit`` and
must leave the audit_events chain re-verifiable."""
import hashlib
db = tmp_path / "shard.db"
parser = build_parser()
# SelfModel + Memory + Capital all trigger audit events.
_run(parser, ["--db", str(db), "selfmodel", "snapshot"])
capsys.readouterr()
_run(parser, ["--db", str(db), "memory", "snapshot"])
capsys.readouterr()
_seed_capital_rows(db)
conn = connect(db)
try:
rows = conn.execute(
"SELECT event_hash, prev_event_hash, body FROM audit_events ORDER BY seq"
).fetchall()
prev = None
for row in rows:
h = hashlib.sha256()
if prev is not None:
h.update(bytes.fromhex(prev))
h.update(row["body"].encode("utf-8", errors="surrogatepass"))
assert h.hexdigest() == row["event_hash"], (
"audit chain break detected at row "
f"{row['event_hash']}"
)
assert row["prev_event_hash"] == prev
prev = row["event_hash"]
finally:
conn.close()
Reference in a new issue View git blame Copy permalink