145 lines
6.3 KiB
Python
145 lines
6.3 KiB
Python
"""Integration: every `tht ...` command the gate invokes must exist in the Typer CLI.
|
|
|
|
Root cause of the Blocco 1 critical bugs: the Pi gate and the Python CLI
|
|
were ported separately and never run together, so the gate called commands/flags that
|
|
did not exist (`phase advance --if-ready`, `cte plan` without `--name`, `set-question
|
|
--session` on a positional arg). This test extracts every `["group","sub",...,"--flag"]`
|
|
array literal from tht-gate.js, plus the subcommand arrays passed through domain-specific
|
|
CLI capabilities, and asserts via `tht <group> <sub> --help` that the subcommand exists
|
|
(exit 0) and that each long flag used is offered. It makes that whole class of drift
|
|
impossible to reintroduce silently.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from typer.main import get_command
|
|
|
|
from tht.cli import app
|
|
|
|
_ROOT = Path(__file__).resolve().parent.parent.parent
|
|
_GATE = _ROOT / ".pi" / "extensions" / "tht-gate.js"
|
|
_DISAMBIGUATION_GATE = _ROOT / ".pi" / "extensions" / "gate" / "disambiguation" / "index.js"
|
|
_THT = Path(sys.executable).parent / "tht"
|
|
|
|
# The command groups the gate drives. Anything else in an array literal is data, not a CLI call.
|
|
_GROUPS = {"phase", "session", "cte", "decision", "memory", "search", "schema", "sql", "vector"}
|
|
|
|
# Python's workflow CLI is also a contract for its non-gate consumers. Keep this
|
|
# list literal rather than deriving it from the current tree: a deleted live
|
|
# command must fail this test before a container or preprocessing job reaches it.
|
|
_LIVE_CONSUMER_COMMANDS = {
|
|
"Pi skill": {
|
|
("cte", "info"), ("cte", "save"), ("cte", "test"),
|
|
("datamart", "generate"), ("memory", "promote"),
|
|
("memory", "save-one"), ("memory", "search"),
|
|
("memory", "solved-index"), ("memory", "solved-search"),
|
|
("phase", "advance"), ("phase", "reopen"),
|
|
("schema", "introspect"), ("schema", "render"),
|
|
("search", "find"), ("search", "pack"),
|
|
("session", "check"), ("session", "documents"),
|
|
("session", "finalize"), ("session", "show"),
|
|
("sql", "preview"), ("sql", "set-final"), ("sql", "validate"),
|
|
},
|
|
"backend ThtRunner": {
|
|
("db", "ping"), ("ollama", "ensure"), ("search", "pack"),
|
|
("session", "archive"), ("session", "close"), ("session", "delete"),
|
|
("session", "documents"), ("session", "fail"), ("session", "list"),
|
|
("session", "new"), ("session", "reopen"), ("session", "set-group"),
|
|
("session", "set-name"), ("session", "show"), ("session", "unarchive"),
|
|
("session", "preferences", "get"), ("session", "preferences", "set"),
|
|
("sql", "export"), ("sql", "preview"),
|
|
},
|
|
"workspace preprocessing": {
|
|
("preprocess", "dwh"), ("preprocess", "evidence"),
|
|
("schema", "check"), ("schema", "suggest-fks"), ("vector", "index-schema"),
|
|
},
|
|
"deployment smoke": {("config", "check"), ("db", "ping")},
|
|
}
|
|
|
|
# Flags that are framework/JS artifacts, never real CLI options (skip from the check).
|
|
_SKIP_FLAGS: set[str] = set()
|
|
|
|
|
|
def _extract_invocations() -> list[tuple[str, str, list[str], str]]:
|
|
"""Returns (group, subcommand, long_flags, raw) for each gate CLI call site."""
|
|
src = _GATE.read_text()
|
|
# array literal opening with two string literals: ["group", "sub" ...
|
|
pattern = re.compile(r'\[\s*"([a-z-]+)"\s*,\s*"([a-z][a-z-]*)"((?:\s*,\s*[^\]\[]+?)?)\]')
|
|
out: list[tuple[str, str, list[str], str]] = []
|
|
for m in pattern.finditer(src):
|
|
group, sub, tail = m.group(1), m.group(2), m.group(3)
|
|
if group not in _GROUPS:
|
|
continue
|
|
flags = [f for f in re.findall(r'"(--[a-z][a-z-]*)"', tail) if f not in _SKIP_FLAGS]
|
|
out.append((group, sub, flags, m.group(0)))
|
|
|
|
# Disambiguation receives a session-scoped capability, so its arrays start at
|
|
# the subcommand while tht-gate.js supplies the fixed `session` group prefix.
|
|
domain_src = _DISAMBIGUATION_GATE.read_text()
|
|
domain_pattern = re.compile(r'\[\s*"([a-z][a-z-]*)"((?:\s*,\s*[^\]\[]+?)?)\]')
|
|
for m in domain_pattern.finditer(domain_src):
|
|
sub, tail = m.group(1), m.group(2)
|
|
flags = [f for f in re.findall(r'"(--[a-z][a-z-]*)"', tail) if f not in _SKIP_FLAGS]
|
|
out.append(("session", sub, flags, m.group(0)))
|
|
return out
|
|
|
|
|
|
_INVOCATIONS = _extract_invocations()
|
|
|
|
|
|
def test_gate_invokes_at_least_the_known_commands():
|
|
"""Guards the extractor itself: if it silently matches nothing, the test is useless."""
|
|
pairs = {(g, s) for g, s, _, _ in _INVOCATIONS}
|
|
assert ("phase", "advance") in pairs
|
|
assert ("cte", "plan") in pairs
|
|
assert ("session", "set-question") in pairs
|
|
assert ("decision", "add") in pairs
|
|
assert ("sql", "set-final") in pairs
|
|
|
|
|
|
def _leaf_paths(command, prefix: tuple[str, ...] = ()) -> set[tuple[str, ...]]:
|
|
children = getattr(command, "commands", {})
|
|
if not children:
|
|
return {prefix} if prefix else set()
|
|
paths: set[tuple[str, ...]] = set()
|
|
for name, child in children.items():
|
|
paths.update(_leaf_paths(child, prefix + (name,)))
|
|
return paths
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("consumer", "path"),
|
|
[
|
|
(consumer, path)
|
|
for consumer, paths in _LIVE_CONSUMER_COMMANDS.items()
|
|
for path in sorted(paths)
|
|
],
|
|
)
|
|
def test_live_workflow_consumers_keep_their_command_paths(consumer: str, path: tuple[str, ...]):
|
|
"""Live Pi, backend, preprocessing, and smoke paths must remain callable."""
|
|
assert path in _leaf_paths(get_command(app)), f"{consumer} requires `tht {' '.join(path)}`"
|
|
|
|
|
|
@pytest.mark.skipif(not _THT.exists(), reason="tht CLI not installed in this venv")
|
|
@pytest.mark.parametrize("group,sub,flags,raw", _INVOCATIONS, ids=lambda v: v if isinstance(v, str) else None)
|
|
def test_gate_call_site_matches_cli(group: str, sub: str, flags: list[str], raw: str):
|
|
res = subprocess.run(
|
|
[str(_THT), group, sub, "--help"],
|
|
capture_output=True, check=False, text=True, cwd=str(_ROOT),
|
|
)
|
|
assert res.returncode == 0, (
|
|
f"gate calls `tht {group} {sub}` but it does not exist in the CLI.\n"
|
|
f"call site: {raw}\nstderr: {res.stderr}"
|
|
)
|
|
help_text = res.stdout + res.stderr
|
|
for flag in flags:
|
|
assert flag in help_text, (
|
|
f"gate passes `{flag}` to `tht {group} {sub}` but the CLI does not offer it.\n"
|
|
f"call site: {raw}"
|
|
)
|