forked from HAL9000/cleveragents-core
freemo
a0df5a4cd0
Implements the spec-required JSON/YAML output envelope for all CLI commands
that use format_output(). The envelope structure is:
{
"command": "<command that was run>",
"status": "ok" | "warn" | "error",
"exit_code": 0,
"data": { ... command-specific payload ... },
"timing": { "duration_ms": 123 },
"messages": [{ "level": "ok", "text": "..." }]
}
Changes:
- Add _build_envelope() helper to construct the spec-required envelope
- Add optional command, status, exit_code, messages parameters to format_output()
- Wrap json/yaml output in the envelope; plain/table/rich/color unchanged
- Add timing measurement (duration_ms) to all json/yaml outputs
- Add new BDD feature file (cli_json_envelope.feature) with 14 scenarios
testing envelope field presence, values, and data payload
- Update 14 existing step files to unwrap the envelope when checking
specific data keys (backward-compatible via _unwrap_envelope() helper)
Closes #3431
736 lines
28 KiB
Python
736 lines
28 KiB
Python
"""Step definitions for config_cli_safety_net_coverage.feature.
|
|
|
|
Safety-net tests exercising every function in config.py to maintain
|
|
100% line and branch coverage. All step text uses a 'safety-net' prefix
|
|
to avoid collisions with config_cli_steps.py and
|
|
config_cli_uncovered_branches_steps.py.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
from io import StringIO
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
|
|
import typer
|
|
import yaml
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from rich.console import Console
|
|
from typer.testing import CliRunner
|
|
|
|
from cleveragents.cli.commands import config as config_mod
|
|
from cleveragents.cli.commands.config import (
|
|
_env_var_for_key,
|
|
_is_secret_key,
|
|
_mask_value,
|
|
_normalize_key,
|
|
_read_config_file,
|
|
_resolution_chain,
|
|
_resolve_source,
|
|
_settings_fields,
|
|
_validate_key,
|
|
_write_config_file,
|
|
)
|
|
from cleveragents.cli.commands.config import app as config_app
|
|
|
|
_runner = CliRunner()
|
|
|
|
_ENVELOPE_KEYS = {"command", "status", "exit_code", "data", "timing", "messages"}
|
|
|
|
|
|
def _unwrap_envelope(parsed: Any) -> Any:
|
|
"""Return the ``data`` field if *parsed* is a spec envelope, else *parsed* as-is."""
|
|
if isinstance(parsed, dict) and _ENVELOPE_KEYS.issubset(parsed.keys()):
|
|
return parsed["data"]
|
|
return parsed
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers - isolated temp directory for config file operations
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _setup_safety_net_temp(context: Context) -> None:
|
|
"""Redirect _CONFIG_DIR/_CONFIG_PATH to a fresh temp directory."""
|
|
if not hasattr(context, "_sn_tmpdir"):
|
|
context._sn_tmpdir = tempfile.mkdtemp(prefix="cfg_safety_net_")
|
|
tmp = Path(context._sn_tmpdir)
|
|
context._sn_dir_patch = patch.object(config_mod, "_CONFIG_DIR", tmp)
|
|
context._sn_path_patch = patch.object(
|
|
config_mod, "_CONFIG_PATH", tmp / "config.toml"
|
|
)
|
|
context._sn_dir_patch.start()
|
|
context._sn_path_patch.start()
|
|
|
|
if not hasattr(context, "_cleanup_handlers"):
|
|
context._cleanup_handlers: list[Any] = []
|
|
context._cleanup_handlers.append(lambda: _teardown_safety_net_temp(context))
|
|
|
|
|
|
def _teardown_safety_net_temp(context: Context) -> None:
|
|
for attr in ("_sn_dir_patch", "_sn_path_patch"):
|
|
patcher = getattr(context, attr, None)
|
|
if patcher is not None:
|
|
with contextlib.suppress(RuntimeError):
|
|
patcher.stop()
|
|
tmpdir = getattr(context, "_sn_tmpdir", None)
|
|
if tmpdir and os.path.isdir(tmpdir):
|
|
shutil.rmtree(tmpdir, ignore_errors=True)
|
|
|
|
|
|
# ===================================================================
|
|
# Background / Given
|
|
# ===================================================================
|
|
|
|
|
|
@given("a safety-net isolated temp config directory")
|
|
def step_sn_temp_dir(context: Context) -> None:
|
|
_setup_safety_net_temp(context)
|
|
|
|
|
|
@given('a safety-net toml config file containing key "{key}" with value "{value}"')
|
|
def step_sn_write_toml(context: Context, key: str, value: str) -> None:
|
|
import tomlkit
|
|
|
|
config_path: Path = config_mod._CONFIG_PATH # type: ignore[assignment]
|
|
config_path.parent.mkdir(parents=True, exist_ok=True)
|
|
doc = tomlkit.document()
|
|
# Coerce booleans
|
|
if value.lower() == "true":
|
|
doc[key] = True
|
|
elif value.lower() == "false":
|
|
doc[key] = False
|
|
else:
|
|
doc[key] = value
|
|
with open(config_path, "w") as fh:
|
|
tomlkit.dump(doc, fh)
|
|
|
|
|
|
@given('the safety-net CLI has previously set "{key}" to "{value}"')
|
|
def step_sn_pre_set(context: Context, key: str, value: str) -> None:
|
|
result = _runner.invoke(config_app, ["set", key, value])
|
|
assert result.exit_code == 0, f"Pre-set failed: {result.output}"
|
|
|
|
|
|
@given('the safety-net env var "{var}" is set to "{value}"')
|
|
def step_sn_set_env(context: Context, var: str, value: str) -> None:
|
|
os.environ[var] = value
|
|
if not hasattr(context, "_cleanup_handlers"):
|
|
context._cleanup_handlers = []
|
|
context._cleanup_handlers.append(lambda: os.environ.pop(var, None))
|
|
|
|
|
|
@given("safety-net settings fields include a Path-typed value")
|
|
def step_sn_mock_path_field(context: Context) -> None:
|
|
"""Patch _settings_fields to include a Path value that must be serialised."""
|
|
real_fields = _settings_fields()
|
|
# Inject a Path value into a known field
|
|
real_fields["log_level"] = Path("/mock/safety/net/path")
|
|
|
|
context._sn_fields_patch = patch.object(
|
|
config_mod, "_settings_fields", return_value=real_fields
|
|
)
|
|
context._sn_fields_patch.start()
|
|
context._cleanup_handlers.append(context._sn_fields_patch.stop)
|
|
|
|
|
|
@given("safety-net settings fields include a secret key with a non-pattern value")
|
|
def step_sn_mock_secret_field(context: Context) -> None:
|
|
"""Patch _settings_fields to include a field whose name matches _SECRET_PATTERNS.
|
|
|
|
Uses a value that does NOT match redact_value patterns (no sk-, tok_, etc.)
|
|
so that format_output's _redact_data won't double-mask it.
|
|
"""
|
|
# Use a plain secret value that won't match redact_value regex patterns
|
|
context._sn_actual_secret = "my-secret-value-here"
|
|
mock_fields = {
|
|
"log_level": "INFO",
|
|
"api_key": context._sn_actual_secret,
|
|
}
|
|
mock_defaults = {
|
|
"log_level": "INFO",
|
|
"api_key": None,
|
|
}
|
|
|
|
context._sn_fields_patch2 = patch.object(
|
|
config_mod, "_settings_fields", return_value=mock_fields
|
|
)
|
|
context._sn_defaults_patch2 = patch.object(
|
|
config_mod, "_settings_defaults", return_value=mock_defaults
|
|
)
|
|
context._sn_resolve_patch2 = patch.object(
|
|
config_mod, "_resolve_source", return_value="default"
|
|
)
|
|
context._sn_fields_patch2.start()
|
|
context._sn_defaults_patch2.start()
|
|
context._sn_resolve_patch2.start()
|
|
context._cleanup_handlers.append(context._sn_fields_patch2.stop)
|
|
context._cleanup_handlers.append(context._sn_defaults_patch2.stop)
|
|
context._cleanup_handlers.append(context._sn_resolve_patch2.stop)
|
|
|
|
|
|
@given("safety-net settings fields have a value different from default")
|
|
def step_sn_mock_modified_field(context: Context) -> None:
|
|
"""Patch fields so that a value differs from its default."""
|
|
mock_fields = {"log_level": "DEBUG"}
|
|
mock_defaults = {"log_level": "INFO"}
|
|
|
|
context._sn_fields_patch3 = patch.object(
|
|
config_mod, "_settings_fields", return_value=mock_fields
|
|
)
|
|
context._sn_defaults_patch3 = patch.object(
|
|
config_mod, "_settings_defaults", return_value=mock_defaults
|
|
)
|
|
context._sn_resolve_patch3 = patch.object(
|
|
config_mod, "_resolve_source", return_value="config_file"
|
|
)
|
|
context._sn_fields_patch3.start()
|
|
context._sn_defaults_patch3.start()
|
|
context._sn_resolve_patch3.start()
|
|
context._cleanup_handlers.append(context._sn_fields_patch3.stop)
|
|
context._cleanup_handlers.append(context._sn_defaults_patch3.stop)
|
|
context._cleanup_handlers.append(context._sn_resolve_patch3.stop)
|
|
|
|
|
|
# ===================================================================
|
|
# _normalize_key (L91-96)
|
|
# ===================================================================
|
|
|
|
|
|
@when('the safety-net normalizer processes key "{key}"')
|
|
def step_sn_normalize(context: Context, key: str) -> None:
|
|
context._sn_normalized = _normalize_key(key)
|
|
|
|
|
|
@then('the safety-net normalized result should be "{expected}"')
|
|
def step_sn_normalized_equals(context: Context, expected: str) -> None:
|
|
assert context._sn_normalized == expected, (
|
|
f"Expected '{expected}', got '{context._sn_normalized}'"
|
|
)
|
|
|
|
|
|
# ===================================================================
|
|
# _is_secret_key (L116-118)
|
|
# ===================================================================
|
|
|
|
|
|
@when('the safety-net secret checker inspects key "{key}"')
|
|
def step_sn_is_secret(context: Context, key: str) -> None:
|
|
context._sn_is_secret = _is_secret_key(key)
|
|
|
|
|
|
@then("the safety-net secret check result should be true")
|
|
def step_sn_secret_true(context: Context) -> None:
|
|
assert context._sn_is_secret is True, "Expected secret=True"
|
|
|
|
|
|
@then("the safety-net secret check result should be false")
|
|
def step_sn_secret_false(context: Context) -> None:
|
|
assert context._sn_is_secret is False, "Expected secret=False"
|
|
|
|
|
|
# ===================================================================
|
|
# _mask_value (L121-123)
|
|
# ===================================================================
|
|
|
|
|
|
@when('the safety-net masker masks value "{value}"')
|
|
def step_sn_mask(context: Context, value: str) -> None:
|
|
context._sn_masked = _mask_value(value)
|
|
|
|
|
|
@when("the safety-net masker masks an empty string value")
|
|
def step_sn_mask_empty(context: Context) -> None:
|
|
context._sn_masked = _mask_value("")
|
|
|
|
|
|
@then('the safety-net masked output should be "{expected}"')
|
|
def step_sn_masked_equals(context: Context, expected: str) -> None:
|
|
assert context._sn_masked == expected, (
|
|
f"Expected '{expected}', got '{context._sn_masked}'"
|
|
)
|
|
|
|
|
|
# ===================================================================
|
|
# _env_var_for_key (L158-160)
|
|
# ===================================================================
|
|
|
|
|
|
@when('the safety-net env var builder processes key "{key}"')
|
|
def step_sn_env_var(context: Context, key: str) -> None:
|
|
context._sn_env_var = _env_var_for_key(key)
|
|
|
|
|
|
@then('the safety-net env var name should be "{expected}"')
|
|
def step_sn_env_var_equals(context: Context, expected: str) -> None:
|
|
assert context._sn_env_var == expected, (
|
|
f"Expected '{expected}', got '{context._sn_env_var}'"
|
|
)
|
|
|
|
|
|
# ===================================================================
|
|
# _read_config_file (L126-131)
|
|
# ===================================================================
|
|
|
|
|
|
@when("the safety-net reader reads the config file")
|
|
def step_sn_read_config(context: Context) -> None:
|
|
context._sn_read_result = _read_config_file()
|
|
|
|
|
|
@then("the safety-net read result should be an empty dict")
|
|
def step_sn_read_empty(context: Context) -> None:
|
|
assert context._sn_read_result == {}, (
|
|
f"Expected empty dict, got {context._sn_read_result}"
|
|
)
|
|
|
|
|
|
@then('the safety-net read result should contain key "{key}"')
|
|
def step_sn_read_has_key(context: Context, key: str) -> None:
|
|
assert key in context._sn_read_result, (
|
|
f"Key '{key}' not in {context._sn_read_result}"
|
|
)
|
|
|
|
|
|
# ===================================================================
|
|
# _write_config_file (L134-155) - create new
|
|
# ===================================================================
|
|
|
|
|
|
@when('the safety-net writer writes key "{key}" with value "{value}"')
|
|
def step_sn_write_config(context: Context, key: str, value: str) -> None:
|
|
_write_config_file({key: value})
|
|
|
|
|
|
@then('the safety-net config file should exist and contain key "{key}"')
|
|
def step_sn_file_has_key(context: Context, key: str) -> None:
|
|
import tomllib
|
|
|
|
config_path: Path = config_mod._CONFIG_PATH # type: ignore[assignment]
|
|
assert config_path.exists(), f"Config file does not exist at {config_path}"
|
|
with open(config_path, "rb") as fh:
|
|
data = tomllib.load(fh)
|
|
assert key in data, f"Key '{key}' not in file data: {data}"
|
|
|
|
|
|
# ===================================================================
|
|
# _settings_fields (L62-70)
|
|
# ===================================================================
|
|
|
|
|
|
@when("the safety-net fields loader retrieves all settings fields")
|
|
def step_sn_settings_fields(context: Context) -> None:
|
|
context._sn_fields = _settings_fields()
|
|
|
|
|
|
@then("the safety-net fields result should be a non-empty dict")
|
|
def step_sn_fields_nonempty(context: Context) -> None:
|
|
assert isinstance(context._sn_fields, dict), (
|
|
f"Expected dict, got {type(context._sn_fields)}"
|
|
)
|
|
assert len(context._sn_fields) > 0, "Expected non-empty dict of fields"
|
|
|
|
|
|
# ===================================================================
|
|
# _validate_key (L99-113) - unknown key
|
|
# ===================================================================
|
|
|
|
|
|
@when('the safety-net validator checks unknown key "{key}"')
|
|
def step_sn_validate_unknown(context: Context, key: str) -> None:
|
|
context._sn_validate_exc = None
|
|
try:
|
|
_validate_key(key)
|
|
except typer.BadParameter as exc:
|
|
context._sn_validate_exc = exc
|
|
|
|
|
|
@then('the safety-net validator should raise BadParameter with "{msg}"')
|
|
def step_sn_validate_bad_param(context: Context, msg: str) -> None:
|
|
assert context._sn_validate_exc is not None, (
|
|
"Expected typer.BadParameter but no exception raised"
|
|
)
|
|
assert msg.lower() in str(context._sn_validate_exc).lower(), (
|
|
f"Expected '{msg}' in: {context._sn_validate_exc}"
|
|
)
|
|
|
|
|
|
@when('the safety-net validator checks valid key "{key}"')
|
|
def step_sn_validate_valid(context: Context, key: str) -> None:
|
|
context._sn_validate_result = _validate_key(key)
|
|
|
|
|
|
@then('the safety-net validator should return "{expected}"')
|
|
def step_sn_validate_returns(context: Context, expected: str) -> None:
|
|
assert context._sn_validate_result == expected, (
|
|
f"Expected '{expected}', got '{context._sn_validate_result}'"
|
|
)
|
|
|
|
|
|
# ===================================================================
|
|
# _resolve_source (L163-174)
|
|
# ===================================================================
|
|
|
|
|
|
@when('the safety-net source resolver checks key "{key}"')
|
|
def step_sn_resolve_source(context: Context, key: str) -> None:
|
|
context._sn_source = _resolve_source(key)
|
|
|
|
|
|
@then('the safety-net resolved source should be "{expected}"')
|
|
def step_sn_source_equals(context: Context, expected: str) -> None:
|
|
assert context._sn_source == expected, (
|
|
f"Expected '{expected}', got '{context._sn_source}'"
|
|
)
|
|
|
|
|
|
# ===================================================================
|
|
# _resolution_chain (L177-209)
|
|
# ===================================================================
|
|
|
|
|
|
@when('the safety-net chain builder builds chain for key "{key}"')
|
|
def step_sn_resolution_chain(context: Context, key: str) -> None:
|
|
context._sn_chain = _resolution_chain(key)
|
|
|
|
|
|
@then("the safety-net chain should have exactly {count:d} entries")
|
|
def step_sn_chain_count(context: Context, count: int) -> None:
|
|
assert len(context._sn_chain) == count, (
|
|
f"Expected {count} entries, got {len(context._sn_chain)}"
|
|
)
|
|
|
|
|
|
@then('the safety-net chain sources should be "{sources}"')
|
|
def step_sn_chain_sources(context: Context, sources: str) -> None:
|
|
expected = [s.strip() for s in sources.split(",")]
|
|
actual = [entry["source"] for entry in context._sn_chain]
|
|
assert actual == expected, f"Expected sources {expected}, got {actual}"
|
|
|
|
|
|
# ===================================================================
|
|
# config_set - type coercion (L241-248)
|
|
# ===================================================================
|
|
|
|
|
|
@when('the safety-net CLI sets key "{key}" to value "{value}" with format "{fmt}"')
|
|
def step_sn_cli_set(context: Context, key: str, value: str, fmt: str) -> None:
|
|
context._sn_result = _runner.invoke(
|
|
config_app, ["set", key, value, "--format", fmt]
|
|
)
|
|
|
|
|
|
@then("the safety-net set output should be valid JSON")
|
|
def step_sn_set_valid_json(context: Context) -> None:
|
|
result = context._sn_result
|
|
assert result.exit_code == 0, f"Exit {result.exit_code}: {result.output}"
|
|
parsed = json.loads(result.output.strip())
|
|
assert isinstance(parsed, dict), f"Expected dict, got {type(parsed)}"
|
|
context._sn_set_json = _unwrap_envelope(parsed)
|
|
|
|
|
|
@then('the safety-net set JSON field "value" should be boolean true')
|
|
def step_sn_set_bool_true(context: Context) -> None:
|
|
assert context._sn_set_json["value"] is True, (
|
|
f"Expected True, got {context._sn_set_json['value']!r}"
|
|
)
|
|
|
|
|
|
@then('the safety-net set JSON field "value" should be boolean false')
|
|
def step_sn_set_bool_false(context: Context) -> None:
|
|
assert context._sn_set_json["value"] is False, (
|
|
f"Expected False, got {context._sn_set_json['value']!r}"
|
|
)
|
|
|
|
|
|
@then('the safety-net set JSON field "value" should be integer {expected:d}')
|
|
def step_sn_set_int(context: Context, expected: int) -> None:
|
|
val = context._sn_set_json["value"]
|
|
assert val == expected and isinstance(val, int), (
|
|
f"Expected int {expected}, got {val!r} ({type(val).__name__})"
|
|
)
|
|
|
|
|
|
@then('the safety-net set JSON field "value" should be float {expected:g}')
|
|
def step_sn_set_float(context: Context, expected: float) -> None:
|
|
val = context._sn_set_json["value"]
|
|
assert isinstance(val, float) and abs(val - expected) < 1e-9, (
|
|
f"Expected float {expected}, got {val!r} ({type(val).__name__})"
|
|
)
|
|
|
|
|
|
@then('the safety-net set JSON field "value" should be string "{expected}"')
|
|
def step_sn_set_string(context: Context, expected: str) -> None:
|
|
val = context._sn_set_json["value"]
|
|
assert val == expected and isinstance(val, str), (
|
|
f"Expected string '{expected}', got {val!r} ({type(val).__name__})"
|
|
)
|
|
|
|
|
|
@then('the safety-net set JSON field "previous_value" should be string "{expected}"')
|
|
def step_sn_set_previous(context: Context, expected: str) -> None:
|
|
val = context._sn_set_json["previous_value"]
|
|
assert val == expected, f"Expected previous '{expected}', got {val!r}"
|
|
|
|
|
|
@then('the safety-net set JSON should contain a "previous_value" field')
|
|
def step_sn_set_has_previous(context: Context) -> None:
|
|
assert "previous_value" in context._sn_set_json, (
|
|
f"Expected 'previous_value' field in: {list(context._sn_set_json.keys())}"
|
|
)
|
|
|
|
|
|
@then('the safety-net set rich output should contain "{text}"')
|
|
def step_sn_set_rich_contains(context: Context, text: str) -> None:
|
|
result = context._sn_result
|
|
assert result.exit_code == 0, f"Exit {result.exit_code}: {result.output}"
|
|
assert text in result.output, f"Expected '{text}' in output: {result.output[:500]}"
|
|
|
|
|
|
# ===================================================================
|
|
# config_get - rich and non-rich formats (L279-335)
|
|
# ===================================================================
|
|
|
|
|
|
@when('the safety-net CLI gets key "{key}" with format "{fmt}"')
|
|
def step_sn_cli_get(context: Context, key: str, fmt: str) -> None:
|
|
context._sn_result = _runner.invoke(config_app, ["get", key, "--format", fmt])
|
|
|
|
|
|
@when('the safety-net CLI gets key "{key}" with format "{fmt}" and verbose')
|
|
def step_sn_cli_get_verbose(context: Context, key: str, fmt: str) -> None:
|
|
context._sn_result = _runner.invoke(
|
|
config_app, ["get", key, "--format", fmt, "--verbose"]
|
|
)
|
|
|
|
|
|
@then('the safety-net get rich output should contain "{text}"')
|
|
def step_sn_get_rich_contains(context: Context, text: str) -> None:
|
|
result = context._sn_result
|
|
assert result.exit_code == 0, f"Exit {result.exit_code}: {result.output}"
|
|
assert text in result.output, f"Expected '{text}' in output: {result.output[:500]}"
|
|
|
|
|
|
@then("the safety-net get output should be valid YAML")
|
|
def step_sn_get_valid_yaml(context: Context) -> None:
|
|
result = context._sn_result
|
|
assert result.exit_code == 0, f"Exit {result.exit_code}: {result.output}"
|
|
parsed = yaml.safe_load(result.output.strip())
|
|
assert parsed is not None, "YAML parsed to None"
|
|
context._sn_get_yaml = parsed
|
|
|
|
|
|
@then('the safety-net get YAML should contain key "{key}"')
|
|
def step_sn_get_yaml_key(context: Context, key: str) -> None:
|
|
assert key in context._sn_get_yaml, (
|
|
f"Key '{key}' not in YAML: {list(context._sn_get_yaml.keys())}"
|
|
)
|
|
|
|
|
|
@then("the safety-net get output should be valid JSON with type field")
|
|
def step_sn_get_json_type(context: Context) -> None:
|
|
result = context._sn_result
|
|
assert result.exit_code == 0, f"Exit {result.exit_code}: {result.output}"
|
|
parsed = json.loads(result.output.strip())
|
|
data = _unwrap_envelope(parsed)
|
|
assert "type" in data, f"'type' not in JSON keys: {list(data.keys())}"
|
|
assert "key" in data, f"'key' not in JSON keys: {list(data.keys())}"
|
|
assert "resolution_chain" in data, (
|
|
f"'resolution_chain' not in JSON: {list(data.keys())}"
|
|
)
|
|
|
|
|
|
# ===================================================================
|
|
# config_list - formats (L426-449)
|
|
# ===================================================================
|
|
|
|
|
|
@when('the safety-net CLI lists config with format "{fmt}"')
|
|
def step_sn_cli_list_format(context: Context, fmt: str) -> None:
|
|
context._sn_result = _runner.invoke(config_app, ["list", "--format", fmt])
|
|
|
|
|
|
@then("the safety-net list output should be valid YAML list")
|
|
def step_sn_list_valid_yaml(context: Context) -> None:
|
|
result = context._sn_result
|
|
assert result.exit_code == 0, f"Exit {result.exit_code}: {result.output}"
|
|
parsed = yaml.safe_load(result.output.strip())
|
|
assert isinstance(parsed, list), f"Expected list, got {type(parsed)}"
|
|
assert len(parsed) > 0, "Expected non-empty YAML list"
|
|
|
|
|
|
@then("the safety-net list plain output should contain key-value lines")
|
|
def step_sn_list_plain(context: Context) -> None:
|
|
result = context._sn_result
|
|
assert result.exit_code == 0, f"Exit {result.exit_code}: {result.output}"
|
|
# Plain format should have key: value style lines
|
|
assert "key:" in result.output or "log_level" in result.output, (
|
|
f"Expected plain key-value in output: {result.output[:500]}"
|
|
)
|
|
|
|
|
|
@then("the safety-net list table output should contain column headers")
|
|
def step_sn_list_table(context: Context) -> None:
|
|
result = context._sn_result
|
|
assert result.exit_code == 0, f"Exit {result.exit_code}: {result.output}"
|
|
# Table format should have column header text
|
|
output = result.output.lower()
|
|
assert "key" in output or "value" in output, (
|
|
f"Expected table headers in output: {result.output[:500]}"
|
|
)
|
|
|
|
|
|
# ===================================================================
|
|
# config_list - combined key + value filter (L391-399)
|
|
# ===================================================================
|
|
|
|
|
|
@when(
|
|
'the safety-net CLI lists config with key pattern "{pattern}" and value filter "{vfilter}"'
|
|
)
|
|
def step_sn_cli_list_combined(context: Context, pattern: str, vfilter: str) -> None:
|
|
context._sn_result = _runner.invoke(
|
|
config_app, ["list", pattern, "--filter-values", vfilter]
|
|
)
|
|
|
|
|
|
@then("the safety-net combined filter result should succeed")
|
|
def step_sn_combined_ok(context: Context) -> None:
|
|
result = context._sn_result
|
|
assert result.exit_code == 0, f"Exit {result.exit_code}: {result.output}"
|
|
|
|
|
|
# ===================================================================
|
|
# config_list - Path serialisation (L428-429)
|
|
# ===================================================================
|
|
|
|
|
|
@then('the safety-net list JSON output should not contain "PosixPath" or "WindowsPath"')
|
|
def step_sn_list_no_path_obj(context: Context) -> None:
|
|
result = context._sn_result
|
|
assert result.exit_code == 0, f"Exit {result.exit_code}: {result.output}"
|
|
output = result.output
|
|
assert "PosixPath" not in output, f"Found PosixPath in output: {output[:500]}"
|
|
assert "WindowsPath" not in output, f"Found WindowsPath in output: {output[:500]}"
|
|
|
|
|
|
# ===================================================================
|
|
# config_list - secret masking (L408-411)
|
|
# ===================================================================
|
|
|
|
|
|
@when("the safety-net CLI lists all config in json format")
|
|
def step_sn_cli_list_json(context: Context) -> None:
|
|
context._sn_result = _runner.invoke(config_app, ["list", "--format", "json"])
|
|
|
|
|
|
@when("the safety-net CLI lists all config in json format with show-secrets")
|
|
def step_sn_cli_list_json_secrets(context: Context) -> None:
|
|
context._sn_result = _runner.invoke(
|
|
config_app, ["list", "--format", "json", "--show-secrets"]
|
|
)
|
|
|
|
|
|
@then('the safety-net list JSON should contain masked value "****" for the secret key')
|
|
def step_sn_list_masked(context: Context) -> None:
|
|
result = context._sn_result
|
|
assert result.exit_code == 0, f"Exit {result.exit_code}: {result.output}"
|
|
parsed = json.loads(result.output.strip())
|
|
data = _unwrap_envelope(parsed)
|
|
api_entries = [e for e in data if e.get("key") == "api_key"]
|
|
assert len(api_entries) > 0, f"No api_key entry found in: {data}"
|
|
assert api_entries[0]["value"] == "****", (
|
|
f"Expected masked ****, got {api_entries[0]['value']!r}"
|
|
)
|
|
|
|
|
|
@then("the safety-net list JSON should contain the actual secret value")
|
|
def step_sn_list_unmasked(context: Context) -> None:
|
|
result = context._sn_result
|
|
assert result.exit_code == 0, f"Exit {result.exit_code}: {result.output}"
|
|
parsed = json.loads(result.output.strip())
|
|
data = _unwrap_envelope(parsed)
|
|
api_entries = [e for e in data if e.get("key") == "api_key"]
|
|
assert len(api_entries) > 0, f"No api_key entry found in: {data}"
|
|
expected = context._sn_actual_secret
|
|
assert api_entries[0]["value"] == expected, (
|
|
f"Expected '{expected}', got {api_entries[0]['value']!r}"
|
|
)
|
|
|
|
|
|
@then('the safety-net list JSON should contain masked "****" for key "{key}"')
|
|
def step_sn_list_masked_by_key(context: Context, key: str) -> None:
|
|
result = context._sn_result
|
|
assert result.exit_code == 0, f"Exit {result.exit_code}: {result.output}"
|
|
parsed = json.loads(result.output.strip())
|
|
data = _unwrap_envelope(parsed)
|
|
entries = [e for e in data if e.get("key") == key]
|
|
assert len(entries) > 0, f"No '{key}' entry found in list output"
|
|
assert entries[0]["value"] == "****", (
|
|
f"Expected masked ****, got {entries[0]['value']!r}"
|
|
)
|
|
|
|
|
|
@then('the safety-net list JSON should contain value "{value}" for key "{key}"')
|
|
def step_sn_list_value_by_key(context: Context, value: str, key: str) -> None:
|
|
result = context._sn_result
|
|
assert result.exit_code == 0, f"Exit {result.exit_code}: {result.output}"
|
|
parsed = json.loads(result.output.strip())
|
|
data = _unwrap_envelope(parsed)
|
|
entries = [e for e in data if e.get("key") == key]
|
|
assert len(entries) > 0, f"No '{key}' entry found in list output"
|
|
assert entries[0]["value"] == value, (
|
|
f"Expected '{value}', got {entries[0]['value']!r}"
|
|
)
|
|
|
|
|
|
# ===================================================================
|
|
# config_list - modified flag (L405-407)
|
|
# ===================================================================
|
|
|
|
|
|
@then("the safety-net list JSON should include a modified flag set to true")
|
|
def step_sn_list_modified(context: Context) -> None:
|
|
result = context._sn_result
|
|
assert result.exit_code == 0, f"Exit {result.exit_code}: {result.output}"
|
|
parsed = json.loads(result.output.strip())
|
|
data = _unwrap_envelope(parsed)
|
|
modified_entries = [e for e in data if e.get("modified") is True]
|
|
assert len(modified_entries) > 0, f"No entries with modified=True found in: {data}"
|
|
|
|
|
|
# ===================================================================
|
|
# config_get - active marker via patched console (L333)
|
|
# ===================================================================
|
|
|
|
|
|
@given("a safety-net patched console for capturing rich output")
|
|
def step_sn_patch_console(context: Context) -> None:
|
|
"""Patch the module-level console to capture rich output in a StringIO buffer."""
|
|
context._sn_console_buf = StringIO()
|
|
context._sn_test_console = Console(
|
|
file=context._sn_console_buf, no_color=True, width=200
|
|
)
|
|
context._sn_console_patch = patch.object(
|
|
config_mod, "console", context._sn_test_console
|
|
)
|
|
context._sn_console_patch.start()
|
|
context._cleanup_handlers.append(context._sn_console_patch.stop)
|
|
|
|
|
|
@then('the safety-net captured console output should contain "{text}"')
|
|
def step_sn_console_contains(context: Context, text: str) -> None:
|
|
result = context._sn_result
|
|
assert result.exit_code == 0, f"Exit {result.exit_code}: {result.output}"
|
|
console_output = context._sn_console_buf.getvalue()
|
|
assert text in console_output, (
|
|
f"Expected '{text}' in captured console output: {console_output[:500]}"
|
|
)
|