3f14cbbf7e
CI / lint (pull_request) Successful in 40s
CI / security (pull_request) Successful in 32s
CI / quality (pull_request) Successful in 42s
CI / typecheck (pull_request) Successful in 1m1s
CI / benchmark-publish (pull_request) Has been skipped
CI / build (pull_request) Successful in 14s
CI / unit_tests (pull_request) Successful in 3m58s
CI / integration_tests (pull_request) Successful in 4m34s
CI / docker (pull_request) Successful in 1m6s
CI / coverage (pull_request) Successful in 6m21s
CI / quality (push) Successful in 16s
CI / lint (push) Successful in 21s
CI / security (push) Successful in 32s
CI / typecheck (push) Successful in 35s
CI / benchmark-regression (push) Has been skipped
CI / build (push) Successful in 39s
CI / unit_tests (push) Successful in 2m38s
CI / integration_tests (push) Successful in 3m2s
CI / docker (push) Successful in 40s
CI / coverage (push) Successful in 4m9s
CI / benchmark-publish (push) Successful in 14m42s
CI / benchmark-regression (pull_request) Successful in 26m29s
Add LLMTrace Pydantic v2 domain model with all required fields (trace_id, plan_id, decision_id, actor, provider, model, prompt_tokens, completion_tokens, cost_usd, latency_ms, tool_calls, context_hash, streaming, retry_count, error). Define 14 OperationalMetricKey values (PLAN_DURATION_MS, PLAN_TOTAL_COST_USD, PLAN_DECISION_COUNT, ACTOR_INVOCATION_COUNT, ACTOR_LATENCY_MS, TOOL_INVOCATION_COUNT, TOOL_ERROR_RATE, CONTEXT_BUILD_TIME_MS, CONTEXT_TOKEN_COUNT, LLM_CALL_COUNT, LLM_TOTAL_TOKENS, LLM_TOTAL_COST_USD, LLM_AVG_LATENCY_MS, SUBPLAN_COUNT) with MetricEntry model and MetricCollector. Add llm_traces database table (LLMTraceModel) with LLMTraceRepository for persistence. TraceService provides recording, querying, metric computation, plan lifecycle hooks, and optional LangSmith forwarding when LANGCHAIN_TRACING_V2=true. Wired into DI container as trace_service. Includes 28 Behave BDD scenarios, 6 Robot Framework smoke tests, 3 ASV benchmark suites, and reference documentation. Fix pre-existing cli_core server_mode test flake by mocking resolve_server_mode in test steps to avoid stale config file interference. Closes #500
249 lines
9.2 KiB
Python
249 lines
9.2 KiB
Python
"""Step definitions for core system commands (version, info, diagnostics).
|
|
|
|
Covers features/cli_core.feature — the enhanced CLI0.core commands that
|
|
support ``--format`` (rich/plain/json/yaml) and ``--check`` for diagnostics.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from unittest.mock import patch as _patch
|
|
|
|
from behave import then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.cli.commands.system import (
|
|
build_diagnostics_data,
|
|
build_info_data,
|
|
build_version_data,
|
|
)
|
|
from cleveragents.cli.formatting import format_output
|
|
|
|
|
|
def _build_info_data_stable() -> dict[str, Any]:
|
|
"""Build info data with deterministic server_mode for tests."""
|
|
with _patch(
|
|
"cleveragents.cli.commands.server.resolve_server_mode",
|
|
return_value="disabled",
|
|
):
|
|
return build_info_data()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _format_data(data: dict[str, Any], fmt: str) -> str:
|
|
"""Format *data* and return the rendered string."""
|
|
if fmt == "rich":
|
|
from io import StringIO
|
|
from unittest.mock import patch
|
|
|
|
from rich.console import Console
|
|
|
|
buf = StringIO()
|
|
console = Console(file=buf, width=200, no_color=True)
|
|
|
|
with patch("cleveragents.cli.main.get_console", return_value=console):
|
|
if data.get("checks") is not None:
|
|
from cleveragents.cli.commands.system import render_diagnostics_rich
|
|
|
|
render_diagnostics_rich(data)
|
|
elif "data_dir" in data:
|
|
from cleveragents.cli.commands.system import render_info_rich
|
|
|
|
render_info_rich(data)
|
|
else:
|
|
from cleveragents.cli.commands.system import render_version_rich
|
|
|
|
render_version_rich(data)
|
|
|
|
return buf.getvalue()
|
|
return format_output(data, fmt)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# VERSION — When steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I run the system version command")
|
|
def step_run_system_version(context: Context) -> None:
|
|
"""Run version with default (rich) format."""
|
|
data = build_version_data()
|
|
context.sys_version_data = data
|
|
context.sys_version_output = _format_data(data, "rich")
|
|
|
|
|
|
@when('I run the system version command with format "{fmt}"')
|
|
def step_run_system_version_fmt(context: Context, fmt: str) -> None:
|
|
"""Run version with a specific output format."""
|
|
data = build_version_data()
|
|
context.sys_version_data = data
|
|
context.sys_version_output = _format_data(data, fmt)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# VERSION — Then steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the system version output should contain "{text}"')
|
|
def step_system_version_output_contains(context: Context, text: str) -> None:
|
|
output = context.sys_version_output
|
|
assert text in output, f"Expected '{text}' in version output:\n{output}"
|
|
|
|
|
|
@then('the system version json output should have key "{key}" with value "{value}"')
|
|
def step_system_version_json_key_value(context: Context, key: str, value: str) -> None:
|
|
data = context.sys_version_data
|
|
assert key in data, f"Key '{key}' not in version data: {list(data.keys())}"
|
|
assert str(data[key]) == value, f"Expected {key}={value!r}, got {data[key]!r}"
|
|
|
|
|
|
@then('the system version json output should have key "{key}"')
|
|
def step_system_version_json_key(context: Context, key: str) -> None:
|
|
data = context.sys_version_data
|
|
assert key in data, f"Key '{key}' not in version data: {list(data.keys())}"
|
|
|
|
|
|
@then('the system version json output should have nested key "{key}"')
|
|
def step_system_version_json_nested_key(context: Context, key: str) -> None:
|
|
data = context.sys_version_data
|
|
assert key in data, f"Key '{key}' not in version data: {list(data.keys())}"
|
|
assert isinstance(data[key], dict), (
|
|
f"Expected '{key}' to be a dict, got {type(data[key])}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# INFO — When steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I run the system info command")
|
|
def step_run_system_info(context: Context) -> None:
|
|
"""Run info in default (rich) format."""
|
|
data = _build_info_data_stable()
|
|
context.sys_info_data = data
|
|
context.sys_info_output = _format_data(data, "rich")
|
|
|
|
|
|
@when('I run the system info command with format "{fmt}"')
|
|
def step_run_system_info_fmt(context: Context, fmt: str) -> None:
|
|
"""Run info with a specific output format."""
|
|
data = _build_info_data_stable()
|
|
context.sys_info_data = data
|
|
context.sys_info_output = _format_data(data, fmt)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# INFO — Then steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the system info output should contain "{text}"')
|
|
def step_system_info_output_contains(context: Context, text: str) -> None:
|
|
output = context.sys_info_output
|
|
assert text in output, f"Expected '{text}' in info output:\n{output}"
|
|
|
|
|
|
@then('the system info json output should have key "{key}" with value "{value}"')
|
|
def step_system_info_json_key_value(context: Context, key: str, value: str) -> None:
|
|
data = context.sys_info_data
|
|
assert key in data, f"Key '{key}' not in info data: {list(data.keys())}"
|
|
assert str(data[key]) == value, f"Expected {key}={value!r}, got {data[key]!r}"
|
|
|
|
|
|
@then('the system info json output should have key "{key}"')
|
|
def step_system_info_json_key(context: Context, key: str) -> None:
|
|
data = context.sys_info_data
|
|
assert key in data, f"Key '{key}' not in info data: {list(data.keys())}"
|
|
|
|
|
|
@then('the system info json output should have nested key "{key}"')
|
|
def step_system_info_json_nested_key(context: Context, key: str) -> None:
|
|
data = context.sys_info_data
|
|
assert key in data, f"Key '{key}' not in info data: {list(data.keys())}"
|
|
assert isinstance(data[key], dict), (
|
|
f"Expected '{key}' to be a dict, got {type(data[key])}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DIAGNOSTICS — When steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I run the system diagnostics command")
|
|
def step_run_system_diagnostics(context: Context) -> None:
|
|
"""Run diagnostics with default (rich) format."""
|
|
data = build_diagnostics_data()
|
|
context.sys_diag_data = data
|
|
context.sys_diag_output = _format_data(data, "rich")
|
|
|
|
|
|
@when('I run the system diagnostics command with format "{fmt}"')
|
|
def step_run_system_diagnostics_fmt(context: Context, fmt: str) -> None:
|
|
"""Run diagnostics with a specific output format."""
|
|
data = build_diagnostics_data()
|
|
context.sys_diag_data = data
|
|
context.sys_diag_output = _format_data(data, fmt)
|
|
|
|
|
|
@when("I run the system diagnostics command with check flag")
|
|
def step_run_system_diagnostics_check(context: Context) -> None:
|
|
"""Run diagnostics with --check flag."""
|
|
data = build_diagnostics_data()
|
|
context.sys_diag_data = data
|
|
context.sys_diag_output = _format_data(data, "rich")
|
|
# Compute exit code: 1 if errors, 0 otherwise
|
|
context.sys_diag_exit_code = 1 if data["has_errors"] else 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DIAGNOSTICS — Then steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the system diagnostics output should contain "{text}"')
|
|
def step_system_diag_output_contains(context: Context, text: str) -> None:
|
|
output = context.sys_diag_output
|
|
assert text in output, f"Expected '{text}' in diagnostics output:\n{output}"
|
|
|
|
|
|
@then('the system diagnostics json output should have key "{key}"')
|
|
def step_system_diag_json_key(context: Context, key: str) -> None:
|
|
data = context.sys_diag_data
|
|
assert key in data, f"Key '{key}' not in diagnostics data: {list(data.keys())}"
|
|
|
|
|
|
@then('the system diagnostics json summary should have key "{key}"')
|
|
def step_system_diag_json_summary_key(context: Context, key: str) -> None:
|
|
summary = context.sys_diag_data.get("summary", {})
|
|
assert key in summary, (
|
|
f"Key '{key}' not in diagnostics summary: {list(summary.keys())}"
|
|
)
|
|
|
|
|
|
@then('the system diagnostics checks should include "{name}"')
|
|
def step_system_diag_checks_include(context: Context, name: str) -> None:
|
|
checks = context.sys_diag_data.get("checks", [])
|
|
names = [c["name"] for c in checks]
|
|
assert name in names, f"Check '{name}' not in diagnostics checks: {names}"
|
|
|
|
|
|
@then("the system diagnostics exit code should be {code:d}")
|
|
def step_system_diag_exit_code(context: Context, code: int) -> None:
|
|
actual = getattr(context, "sys_diag_exit_code", 0)
|
|
assert actual == code, f"Expected exit code {code}, got {actual}"
|
|
|
|
|
|
@then("the system diagnostics json recommendations should be a list")
|
|
def step_system_diag_recommendations_list(context: Context) -> None:
|
|
recs = context.sys_diag_data.get("recommendations")
|
|
assert isinstance(recs, list), (
|
|
f"Expected recommendations to be a list, got {type(recs)}"
|
|
)
|