ab911dbdc4
The format_output() function returned a string that callers passed to Rich console.print(), which wraps long lines at terminal width. This injected literal newline characters into JSON string values (e.g. in definition_of_done fields), producing invalid JSON that downstream parsers could not decode (JSONDecodeError: Invalid control character). For machine-readable formats (json, yaml, plain), format_output() now writes the rendered output directly to sys.stdout and returns an empty string. This preserves the exact serialization from json.dumps/ yaml.dump without Rich text processing artifacts. Refs: #746
256 lines
9.4 KiB
Python
256 lines
9.4 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()
|
|
# Machine-readable formats write to sys.stdout; capture it.
|
|
from contextlib import redirect_stdout
|
|
from io import StringIO
|
|
|
|
buf = StringIO()
|
|
with redirect_stdout(buf):
|
|
result = format_output(data, fmt)
|
|
return result or buf.getvalue().rstrip("\n")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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)}"
|
|
)
|