feat(cli): tht session list --json + session show --json (Task 7)
- Add _list_sessions(sessions_root) pure helper: scans sessions dir, returns list[dict] with id/status/question/summary/created_at/ updated_at/author, sorted by created_at desc. - Add `tht session list` command: --json emits pristine JSON array, human mode prints one line per session. - Add --json flag to `tht session show`: emits manifest (model_dump by_alias) + phase (current_phase) + has_schema_linking. - Tests: 8 tests in test_session_list_json.py (TDD red→green). - Full suite: 230 passed (--ignore=tests/l2). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,162 @@
|
|||||||
|
"""Tests for `tht session list --json` and `tht session show --json` (Task 7)."""
|
||||||
|
import json
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from typer.testing import CliRunner
|
||||||
|
|
||||||
|
from tht.cli.session_cmd import _list_sessions, session_app
|
||||||
|
from tht.session.store import create_session
|
||||||
|
|
||||||
|
|
||||||
|
def _make_db():
|
||||||
|
from tht.config import DatabaseConfig
|
||||||
|
|
||||||
|
return DatabaseConfig(
|
||||||
|
database="testdb",
|
||||||
|
user="testuser",
|
||||||
|
password="testpass", # noqa: S106
|
||||||
|
**{"schema": "public"},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Unit: _list_sessions helper
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def test_list_sessions_returns_all_created(tmp_path):
|
||||||
|
db = _make_db()
|
||||||
|
m1 = create_session("prima domanda", db, tmp_path)
|
||||||
|
m2 = create_session("seconda domanda", db, tmp_path)
|
||||||
|
rows = _list_sessions(tmp_path)
|
||||||
|
ids = [r["id"] for r in rows]
|
||||||
|
assert m1.id in ids
|
||||||
|
assert m2.id in ids
|
||||||
|
|
||||||
|
|
||||||
|
def test_list_sessions_required_keys(tmp_path):
|
||||||
|
db = _make_db()
|
||||||
|
create_session("qualsiasi domanda", db, tmp_path)
|
||||||
|
rows = _list_sessions(tmp_path)
|
||||||
|
assert len(rows) >= 1
|
||||||
|
required = {"id", "status", "question", "created_at"}
|
||||||
|
assert required.issubset(rows[0].keys())
|
||||||
|
|
||||||
|
|
||||||
|
def test_list_sessions_sorted_desc(tmp_path):
|
||||||
|
db = _make_db()
|
||||||
|
create_session("prima", db, tmp_path)
|
||||||
|
create_session("seconda", db, tmp_path)
|
||||||
|
rows = _list_sessions(tmp_path)
|
||||||
|
dates = [r["created_at"] for r in rows]
|
||||||
|
assert dates == sorted(dates, reverse=True)
|
||||||
|
|
||||||
|
|
||||||
|
def test_list_sessions_empty_dir(tmp_path):
|
||||||
|
rows = _list_sessions(tmp_path)
|
||||||
|
assert rows == []
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# CLI: tht session list --json
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def test_cli_list_json_valid(tmp_path, monkeypatch):
|
||||||
|
db = _make_db()
|
||||||
|
m = create_session("test question", db, tmp_path)
|
||||||
|
|
||||||
|
# Patch _load_config_or_exit so we don't need a real config file
|
||||||
|
from tht.cli import session_cmd
|
||||||
|
|
||||||
|
class FakePaths:
|
||||||
|
sessions = tmp_path
|
||||||
|
|
||||||
|
class FakeCfg:
|
||||||
|
paths = FakePaths()
|
||||||
|
|
||||||
|
monkeypatch.setattr(session_cmd, "_load_config_or_exit", lambda _: FakeCfg())
|
||||||
|
|
||||||
|
runner = CliRunner()
|
||||||
|
result = runner.invoke(session_app, ["list", "--json"])
|
||||||
|
assert result.exit_code == 0, result.output
|
||||||
|
rows = json.loads(result.output)
|
||||||
|
assert isinstance(rows, list)
|
||||||
|
ids = [r["id"] for r in rows]
|
||||||
|
assert m.id in ids
|
||||||
|
|
||||||
|
|
||||||
|
def test_cli_list_human_no_json_noise(tmp_path, monkeypatch):
|
||||||
|
"""Human mode must not emit JSON."""
|
||||||
|
db = _make_db()
|
||||||
|
create_session("human question", db, tmp_path)
|
||||||
|
|
||||||
|
from tht.cli import session_cmd
|
||||||
|
|
||||||
|
class FakePaths:
|
||||||
|
sessions = tmp_path
|
||||||
|
|
||||||
|
class FakeCfg:
|
||||||
|
paths = FakePaths()
|
||||||
|
|
||||||
|
monkeypatch.setattr(session_cmd, "_load_config_or_exit", lambda _: FakeCfg())
|
||||||
|
|
||||||
|
runner = CliRunner()
|
||||||
|
result = runner.invoke(session_app, ["list"])
|
||||||
|
assert result.exit_code == 0
|
||||||
|
# Must not be parseable as JSON array
|
||||||
|
with pytest.raises(json.JSONDecodeError):
|
||||||
|
json.loads(result.output)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# CLI: tht session show --json
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def test_cli_show_json_valid(tmp_path, monkeypatch):
|
||||||
|
db = _make_db()
|
||||||
|
m = create_session("domanda show json", db, tmp_path)
|
||||||
|
|
||||||
|
from tht.cli import session_cmd
|
||||||
|
|
||||||
|
class FakePaths:
|
||||||
|
sessions = tmp_path
|
||||||
|
|
||||||
|
class FakeCfg:
|
||||||
|
paths = FakePaths()
|
||||||
|
database = db
|
||||||
|
|
||||||
|
monkeypatch.setattr(session_cmd, "_load_config_or_exit", lambda _: FakeCfg())
|
||||||
|
|
||||||
|
runner = CliRunner()
|
||||||
|
result = runner.invoke(session_app, ["show", m.id, "--json"])
|
||||||
|
assert result.exit_code == 0, result.output
|
||||||
|
data = json.loads(result.output)
|
||||||
|
assert data["id"] == m.id
|
||||||
|
assert "phase" in data
|
||||||
|
assert "has_schema_linking" in data
|
||||||
|
|
||||||
|
|
||||||
|
def test_cli_show_json_stdout_pristine(tmp_path, monkeypatch):
|
||||||
|
"""--json must produce only valid JSON on stdout (no ANSI color codes or extra lines)."""
|
||||||
|
db = _make_db()
|
||||||
|
m = create_session("pristine stdout test", db, tmp_path)
|
||||||
|
|
||||||
|
from tht.cli import session_cmd
|
||||||
|
|
||||||
|
class FakePaths:
|
||||||
|
sessions = tmp_path
|
||||||
|
|
||||||
|
class FakeCfg:
|
||||||
|
paths = FakePaths()
|
||||||
|
database = db
|
||||||
|
|
||||||
|
monkeypatch.setattr(session_cmd, "_load_config_or_exit", lambda _: FakeCfg())
|
||||||
|
|
||||||
|
runner = CliRunner()
|
||||||
|
result = runner.invoke(session_app, ["show", m.id, "--json"])
|
||||||
|
assert result.exit_code == 0, result.output
|
||||||
|
# stdout must parse cleanly as JSON
|
||||||
|
data = json.loads(result.output)
|
||||||
|
assert isinstance(data, dict)
|
||||||
@@ -1,3 +1,4 @@
|
|||||||
|
import json
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
import typer
|
import typer
|
||||||
@@ -22,6 +23,47 @@ def load_session_or_exit(cfg, session_id: str):
|
|||||||
raise typer.Exit(code=1)
|
raise typer.Exit(code=1)
|
||||||
|
|
||||||
|
|
||||||
|
def _list_sessions(sessions_root: Path) -> list[dict]:
|
||||||
|
"""Return session summary dicts sorted by created_at desc (pure helper, no I/O side-effects)."""
|
||||||
|
from tht.session.models import SessionManifest
|
||||||
|
|
||||||
|
MANIFEST_NAME = "session_manifest.yaml"
|
||||||
|
out = []
|
||||||
|
if not sessions_root.exists():
|
||||||
|
return out
|
||||||
|
for d in sessions_root.iterdir():
|
||||||
|
manifest_path = d / MANIFEST_NAME
|
||||||
|
if not manifest_path.exists():
|
||||||
|
continue
|
||||||
|
m = SessionManifest.from_yaml(manifest_path)
|
||||||
|
out.append({
|
||||||
|
"id": m.id,
|
||||||
|
"status": m.status,
|
||||||
|
"question": m.question,
|
||||||
|
"summary": m.summary,
|
||||||
|
"created_at": m.created_at.isoformat(),
|
||||||
|
"updated_at": m.updated_at.isoformat() if m.updated_at else None,
|
||||||
|
"author": m.author,
|
||||||
|
})
|
||||||
|
out.sort(key=lambda r: r["created_at"], reverse=True)
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
@session_app.command("list")
|
||||||
|
def list_cmd(
|
||||||
|
json_out: bool = typer.Option(False, "--json", help="Emetti JSON su stdout (pristine)."),
|
||||||
|
config: Path = CONFIG_OPT,
|
||||||
|
) -> None:
|
||||||
|
"""Elenca le sessioni esistenti."""
|
||||||
|
cfg = _load_config_or_exit(config)
|
||||||
|
rows = _list_sessions(cfg.paths.sessions)
|
||||||
|
if json_out:
|
||||||
|
typer.echo(json.dumps(rows, ensure_ascii=False, indent=2))
|
||||||
|
return
|
||||||
|
for r in rows:
|
||||||
|
typer.echo(f"{r['id']} [{r['status']}] {r['summary'] or r['question'][:60]}")
|
||||||
|
|
||||||
|
|
||||||
@session_app.command("new")
|
@session_app.command("new")
|
||||||
def new_cmd(
|
def new_cmd(
|
||||||
question: str = typer.Argument(..., help="La domanda in linguaggio naturale."),
|
question: str = typer.Argument(..., help="La domanda in linguaggio naturale."),
|
||||||
@@ -56,13 +98,29 @@ def set_question_cmd(
|
|||||||
|
|
||||||
|
|
||||||
@session_app.command("show")
|
@session_app.command("show")
|
||||||
def show_cmd(session_id: str = typer.Argument(...), config: Path = CONFIG_OPT) -> None:
|
def show_cmd(
|
||||||
|
session_id: str = typer.Argument(...),
|
||||||
|
json_out: bool = typer.Option(False, "--json", help="Emetti JSON su stdout (pristine)."),
|
||||||
|
config: Path = CONFIG_OPT,
|
||||||
|
) -> None:
|
||||||
"""Stato della sessione: manifest + decisioni registrate (per la ripresa)."""
|
"""Stato della sessione: manifest + decisioni registrate (per la ripresa)."""
|
||||||
from tht.decisions import list_decisions
|
from tht.decisions import list_decisions
|
||||||
|
from tht.phase import current_phase
|
||||||
|
|
||||||
cfg = _load_config_or_exit(config)
|
cfg = _load_config_or_exit(config)
|
||||||
manifest = load_session_or_exit(cfg, session_id)
|
manifest = load_session_or_exit(cfg, session_id)
|
||||||
decisions = list_decisions(session_dir(cfg, session_id))
|
sdir = session_dir(cfg, session_id)
|
||||||
|
|
||||||
|
if json_out:
|
||||||
|
has_schema_linking = (sdir / "schema_linking.json").exists()
|
||||||
|
phase = current_phase(sdir)
|
||||||
|
data = manifest.model_dump(mode="json", by_alias=True)
|
||||||
|
data["phase"] = phase
|
||||||
|
data["has_schema_linking"] = has_schema_linking
|
||||||
|
typer.echo(json.dumps(data, ensure_ascii=False, indent=2))
|
||||||
|
return
|
||||||
|
|
||||||
|
decisions = list_decisions(sdir)
|
||||||
typer.echo(f"id : {manifest.id}")
|
typer.echo(f"id : {manifest.id}")
|
||||||
typer.echo(f"stato : {manifest.status}")
|
typer.echo(f"stato : {manifest.status}")
|
||||||
typer.echo(f"domanda : {manifest.question}")
|
typer.echo(f"domanda : {manifest.question}")
|
||||||
@@ -70,7 +128,7 @@ def show_cmd(session_id: str = typer.Argument(...), config: Path = CONFIG_OPT) -
|
|||||||
typer.echo(f"decisioni: {len(decisions)}")
|
typer.echo(f"decisioni: {len(decisions)}")
|
||||||
for d in decisions[-10:]:
|
for d in decisions[-10:]:
|
||||||
typer.echo(f" [{d.seq}] {d.type}: {d.subject}" + (f" — {d.detail}" if d.detail else ""))
|
typer.echo(f" [{d.seq}] {d.type}: {d.subject}" + (f" — {d.detail}" if d.detail else ""))
|
||||||
linking = session_dir(cfg, session_id) / "schema_linking.json"
|
linking = sdir / "schema_linking.json"
|
||||||
typer.echo(f"schema_linking.json: {'presente' if linking.exists() else 'assente'}")
|
typer.echo(f"schema_linking.json: {'presente' if linking.exists() else 'assente'}")
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user