russell@unturf.com
fea761c577
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.
360 lines
12 KiB
Python
360 lines
12 KiB
Python
"""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()
|