Luis Mendes
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
668 lines
23 KiB
Python
668 lines
23 KiB
Python
"""Step definitions for context_cli_wiring.feature.
|
|
|
|
Tests the ``agents project context set/show/inspect/simulate``
|
|
CLI commands wired to the ACMS pipeline using an in-memory
|
|
SQLite database and a real ContextTierService instance.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from io import StringIO
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when # type: ignore[import-untyped]
|
|
from sqlalchemy import create_engine # type: ignore[import-untyped]
|
|
from sqlalchemy.orm import Session, sessionmaker # type: ignore[import-untyped]
|
|
|
|
from cleveragents.application.services.context_tiers import ContextTierService
|
|
from cleveragents.domain.models.acms.tiers import (
|
|
ContextTier,
|
|
TieredFragment,
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Session wrapper (no-op close)
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
class _UnclosableSession:
|
|
"""Wraps a SQLAlchemy Session; ``close()`` is a no-op."""
|
|
|
|
def __init__(self, real_session: Session) -> None:
|
|
object.__setattr__(self, "_real", real_session)
|
|
|
|
def close(self) -> None:
|
|
"""No-op."""
|
|
|
|
def __getattr__(self, name: str) -> Any:
|
|
return getattr(object.__getattribute__(self, "_real"), name)
|
|
|
|
def __setattr__(self, name: str, value: Any) -> None:
|
|
setattr(object.__getattribute__(self, "_real"), name, value)
|
|
|
|
|
|
def _make_wiring_session_factory(
|
|
context: Any,
|
|
) -> tuple[Any, Any]:
|
|
"""In-memory SQLite with all tables."""
|
|
from cleveragents.infrastructure.database.models import Base
|
|
|
|
engine = create_engine(
|
|
"sqlite:///:memory:",
|
|
echo=False,
|
|
connect_args={"check_same_thread": False},
|
|
)
|
|
Base.metadata.create_all(engine)
|
|
real_session = sessionmaker(
|
|
bind=engine,
|
|
expire_on_commit=False,
|
|
autoflush=True,
|
|
autocommit=False,
|
|
)()
|
|
wrapper = _UnclosableSession(real_session)
|
|
|
|
def _factory() -> Any:
|
|
return wrapper
|
|
|
|
return engine, _factory
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Background
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@given("a context wiring in-memory database is initialized")
|
|
def step_init_wiring_db(context: Any) -> None:
|
|
from cleveragents.infrastructure.database.repositories import (
|
|
NamespacedProjectRepository,
|
|
)
|
|
|
|
engine, session_factory = _make_wiring_session_factory(context)
|
|
context.w_engine = engine
|
|
context.w_session_factory = session_factory
|
|
context.w_project_repo = NamespacedProjectRepository(
|
|
session_factory=session_factory
|
|
)
|
|
context.w_tier_service = ContextTierService()
|
|
context.w_output = ""
|
|
context.w_exit_code = 0
|
|
context.w_error = None
|
|
|
|
|
|
@given('a project "{name}" exists for context wiring')
|
|
def step_create_project_for_wiring(context: Any, name: str) -> None:
|
|
from cleveragents.domain.models.core.project import (
|
|
NamespacedProject,
|
|
parse_namespaced_name,
|
|
)
|
|
|
|
parsed = parse_namespaced_name(name)
|
|
proj = NamespacedProject(
|
|
name=parsed.name,
|
|
namespace=parsed.namespace,
|
|
)
|
|
context.w_project_repo.create(proj)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
def _mock_wiring_container(context: Any) -> MagicMock:
|
|
"""Build a mock container wired to the test DB and tier service."""
|
|
mock_container = MagicMock()
|
|
mock_container.namespaced_project_repo.return_value = context.w_project_repo
|
|
mock_container.session_factory.return_value = context.w_session_factory
|
|
mock_container.context_tier_service.return_value = context.w_tier_service
|
|
return mock_container
|
|
|
|
|
|
def _test_format_output(data, format_type):
|
|
"""Test-friendly format_output: captures stdout writes and returns the string."""
|
|
import sys
|
|
from io import StringIO
|
|
|
|
from cleveragents.cli.formatting import format_output
|
|
|
|
buf = StringIO()
|
|
old = sys.stdout
|
|
sys.stdout = buf
|
|
try:
|
|
result = format_output(data, format_type)
|
|
finally:
|
|
sys.stdout = old
|
|
return result or buf.getvalue().rstrip("\n")
|
|
|
|
|
|
def _run_wired(context: Any, func: Any, *args: Any, **kwargs: Any) -> None:
|
|
"""Execute a CLI function with a mocked DI container."""
|
|
import typer
|
|
from rich.console import Console as RichConsole
|
|
|
|
mock_cont = _mock_wiring_container(context)
|
|
buf = StringIO()
|
|
test_console = RichConsole(
|
|
file=buf,
|
|
no_color=True,
|
|
highlight=False,
|
|
force_terminal=False,
|
|
width=500,
|
|
soft_wrap=True,
|
|
)
|
|
|
|
with (
|
|
patch(
|
|
"cleveragents.application.container.get_container",
|
|
return_value=mock_cont,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.project_context.console",
|
|
test_console,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.project_context.err_console",
|
|
test_console,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.project_context.format_output",
|
|
_test_format_output,
|
|
),
|
|
):
|
|
try:
|
|
func(*args, **kwargs)
|
|
context.w_exit_code = 0
|
|
except typer.Exit as exc:
|
|
context.w_exit_code = exc.exit_code
|
|
except typer.Abort:
|
|
context.w_exit_code = 1
|
|
|
|
context.w_output = buf.getvalue()
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# context set with ACMS options
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@when('I run wired context set on "{project}" with hot-max-tokens {val:d}')
|
|
def step_wired_set_hot_max_tokens(context: Any, project: str, val: int) -> None:
|
|
from cleveragents.cli.commands.project_context import context_set
|
|
|
|
_run_wired(context, context_set, project=project, hot_max_tokens=val)
|
|
|
|
|
|
@when('I run wired context set on "{project}" with warm-max-decisions {val:d}')
|
|
def step_wired_set_warm_max_decisions(context: Any, project: str, val: int) -> None:
|
|
from cleveragents.cli.commands.project_context import context_set
|
|
|
|
_run_wired(context, context_set, project=project, warm_max_decisions=val)
|
|
|
|
|
|
@when('I run wired context set on "{project}" with cold-max-decisions {val:d}')
|
|
def step_wired_set_cold_max_decisions(context: Any, project: str, val: int) -> None:
|
|
from cleveragents.cli.commands.project_context import context_set
|
|
|
|
_run_wired(context, context_set, project=project, cold_max_decisions=val)
|
|
|
|
|
|
@when('I run wired context set on "{project}" with temporal-scope "{scope}"')
|
|
def step_wired_set_temporal_scope(context: Any, project: str, scope: str) -> None:
|
|
from cleveragents.cli.commands.project_context import context_set
|
|
|
|
_run_wired(context, context_set, project=project, temporal_scope=scope)
|
|
|
|
|
|
@when('I run wired context set on "{project}" with no-auto-refresh')
|
|
def step_wired_set_no_auto_refresh(context: Any, project: str) -> None:
|
|
from cleveragents.cli.commands.project_context import context_set
|
|
|
|
_run_wired(context, context_set, project=project, auto_refresh=False)
|
|
|
|
|
|
@when('I run wired context set on "{project}" with no-summarize')
|
|
def step_wired_set_no_summarize(context: Any, project: str) -> None:
|
|
from cleveragents.cli.commands.project_context import context_set
|
|
|
|
_run_wired(context, context_set, project=project, summarize=False)
|
|
|
|
|
|
@when('I run wired context set on "{project}" with strategy "{strategy}"')
|
|
def step_wired_set_strategy(context: Any, project: str, strategy: str) -> None:
|
|
from cleveragents.cli.commands.project_context import context_set
|
|
|
|
_run_wired(context, context_set, project=project, strategy=[strategy])
|
|
|
|
|
|
@when('I run wired context set on "{project}" with skeleton-ratio {val:g}')
|
|
def step_wired_set_skeleton_ratio(context: Any, project: str, val: float) -> None:
|
|
from cleveragents.cli.commands.project_context import context_set
|
|
|
|
_run_wired(context, context_set, project=project, skeleton_ratio=val)
|
|
|
|
|
|
@when('I run wired context set on "{project}" with default-breadth {val:d}')
|
|
def step_wired_set_default_breadth(context: Any, project: str, val: int) -> None:
|
|
from cleveragents.cli.commands.project_context import context_set
|
|
|
|
_run_wired(context, context_set, project=project, default_breadth=val)
|
|
|
|
|
|
@when('I run wired context set on "{project}" with default-depth {val:d}')
|
|
def step_wired_set_default_depth(context: Any, project: str, val: int) -> None:
|
|
from cleveragents.cli.commands.project_context import context_set
|
|
|
|
_run_wired(context, context_set, project=project, default_depth=str(val))
|
|
|
|
|
|
@when('I run wired context set on "{project}" with query-limit {val:d}')
|
|
def step_wired_set_query_limit(context: Any, project: str, val: int) -> None:
|
|
from cleveragents.cli.commands.project_context import context_set
|
|
|
|
_run_wired(context, context_set, project=project, query_limit=val)
|
|
|
|
|
|
@when('I run wired context set on "{project}" with summary-max-tokens {val:d}')
|
|
def step_wired_set_summary_max_tokens(context: Any, project: str, val: int) -> None:
|
|
from cleveragents.cli.commands.project_context import context_set
|
|
|
|
_run_wired(context, context_set, project=project, summary_max_tokens=val)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# context show with ACMS config
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@given('I have set ACMS config on "{project}" with hot-max-tokens {val:d}')
|
|
def step_given_acms_config(context: Any, project: str, val: int) -> None:
|
|
from cleveragents.cli.commands.project_context import context_set
|
|
|
|
_run_wired(context, context_set, project=project, hot_max_tokens=val)
|
|
|
|
|
|
@when('I run wired context show on "{project}" with format "{fmt}"')
|
|
def step_wired_show_format(context: Any, project: str, fmt: str) -> None:
|
|
from cleveragents.cli.commands.project_context import context_show
|
|
|
|
_run_wired(context, context_show, project=project, output_format=fmt)
|
|
|
|
|
|
@when('I run wired context show on "{project}" with view "{view}" and format "{fmt}"')
|
|
def step_wired_show_view_format(
|
|
context: Any, project: str, view: str, fmt: str
|
|
) -> None:
|
|
from cleveragents.cli.commands.project_context import context_show
|
|
|
|
_run_wired(context, context_show, project=project, view=view, output_format=fmt)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# context inspect
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@when('I run wired inspect for project "{project}" using defaults')
|
|
def step_wired_inspect(context: Any, project: str) -> None:
|
|
from cleveragents.cli.commands.project_context import context_inspect
|
|
|
|
_run_wired(context, context_inspect, project=project)
|
|
|
|
|
|
@when('I run wired inspect for project "{project}" using format "{fmt}"')
|
|
def step_wired_inspect_format(context: Any, project: str, fmt: str) -> None:
|
|
from cleveragents.cli.commands.project_context import context_inspect
|
|
|
|
_run_wired(context, context_inspect, project=project, output_format=fmt)
|
|
|
|
|
|
@when('I run wired inspect for project "{project}" using view "{view}"')
|
|
def step_wired_inspect_view(context: Any, project: str, view: str) -> None:
|
|
from cleveragents.cli.commands.project_context import context_inspect
|
|
|
|
_run_wired(context, context_inspect, project=project, view=view)
|
|
|
|
|
|
@given('the tier service has a hot fragment "{fid}" for project "{project}"')
|
|
def step_given_hot_fragment(context: Any, fid: str, project: str) -> None:
|
|
frag = TieredFragment(
|
|
fragment_id=fid,
|
|
content="test content for " + fid,
|
|
tier=ContextTier.HOT,
|
|
project_name=project,
|
|
token_count=50,
|
|
)
|
|
context.w_tier_service.store(frag)
|
|
|
|
|
|
@given('the tier service has a warm fragment "{fid}" for project "{project}"')
|
|
def step_given_warm_fragment(context: Any, fid: str, project: str) -> None:
|
|
frag = TieredFragment(
|
|
fragment_id=fid,
|
|
content="test content for " + fid,
|
|
tier=ContextTier.WARM,
|
|
project_name=project,
|
|
token_count=30,
|
|
)
|
|
context.w_tier_service.store(frag)
|
|
|
|
|
|
@given(
|
|
'the tier service has a hot fragment "{fid}" for project'
|
|
' "{project}" sized at {tokens:d} tokens'
|
|
)
|
|
def step_given_hot_fragment_tokens(
|
|
context: Any, fid: str, project: str, tokens: int
|
|
) -> None:
|
|
frag = TieredFragment(
|
|
fragment_id=fid,
|
|
content="test content for simulation " + fid,
|
|
tier=ContextTier.HOT,
|
|
project_name=project,
|
|
token_count=tokens,
|
|
)
|
|
context.w_tier_service.store(frag)
|
|
|
|
|
|
@given(
|
|
'a focused hot fragment "{fid}" with resource "{resource}"'
|
|
" and {tokens:d} tokens exists"
|
|
' for project "{project}"'
|
|
)
|
|
def step_given_focused_hot_fragment(
|
|
context: Any, fid: str, resource: str, tokens: int, project: str
|
|
) -> None:
|
|
frag = TieredFragment(
|
|
fragment_id=fid,
|
|
content="test content for " + fid,
|
|
tier=ContextTier.HOT,
|
|
project_name=project,
|
|
resource_id=resource,
|
|
token_count=tokens,
|
|
)
|
|
context.w_tier_service.store(frag)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# context simulate
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@when('I run wired simulate for project "{project}" using defaults')
|
|
def step_wired_simulate(context: Any, project: str) -> None:
|
|
from cleveragents.cli.commands.project_context import context_simulate
|
|
|
|
_run_wired(context, context_simulate, project=project)
|
|
|
|
|
|
@when('I run wired simulate for project "{project}" using format "{fmt}"')
|
|
def step_wired_simulate_format(context: Any, project: str, fmt: str) -> None:
|
|
from cleveragents.cli.commands.project_context import context_simulate
|
|
|
|
_run_wired(context, context_simulate, project=project, output_format=fmt)
|
|
|
|
|
|
@when('I run wired simulate for project "{project}" using budget {val:d}')
|
|
def step_wired_simulate_budget(context: Any, project: str, val: int) -> None:
|
|
from cleveragents.cli.commands.project_context import context_simulate
|
|
|
|
_run_wired(context, context_simulate, project=project, budget=val)
|
|
|
|
|
|
@when('I run wired simulate for project "{project}" using view "{view}"')
|
|
def step_wired_simulate_view(context: Any, project: str, view: str) -> None:
|
|
from cleveragents.cli.commands.project_context import context_simulate
|
|
|
|
_run_wired(context, context_simulate, project=project, view=view)
|
|
|
|
|
|
@when('I run wired simulate for project "{project}" using strategy "{strat}"')
|
|
def step_wired_simulate_strategy(context: Any, project: str, strat: str) -> None:
|
|
from cleveragents.cli.commands.project_context import context_simulate
|
|
|
|
_run_wired(context, context_simulate, project=project, strategy_hint=[strat])
|
|
|
|
|
|
@when(
|
|
'I run wired simulate for project "{project}"'
|
|
' focusing on "{focus}" in format "{fmt}"'
|
|
)
|
|
def step_wired_simulate_focus_format(
|
|
context: Any, project: str, focus: str, fmt: str
|
|
) -> None:
|
|
from cleveragents.cli.commands.project_context import context_simulate
|
|
|
|
_run_wired(
|
|
context,
|
|
context_simulate,
|
|
project=project,
|
|
focus=[focus],
|
|
output_format=fmt,
|
|
)
|
|
|
|
|
|
@then("the wired context set command should succeed")
|
|
def step_wired_set_ok(context: Any) -> None:
|
|
assert context.w_exit_code == 0, (
|
|
f"Expected exit 0, got {context.w_exit_code}. Output: {context.w_output}"
|
|
)
|
|
|
|
|
|
@then("the wired context set command should fail")
|
|
def step_wired_set_fail(context: Any) -> None:
|
|
assert context.w_exit_code != 0, (
|
|
f"Expected non-zero exit, got {context.w_exit_code}"
|
|
)
|
|
|
|
|
|
@then("the wired context show command should succeed")
|
|
def step_wired_show_ok(context: Any) -> None:
|
|
assert context.w_exit_code == 0, (
|
|
f"Expected exit 0, got {context.w_exit_code}. Output: {context.w_output}"
|
|
)
|
|
|
|
|
|
@then("the wired context inspect command should succeed")
|
|
def step_wired_inspect_ok(context: Any) -> None:
|
|
assert context.w_exit_code == 0, (
|
|
f"Expected exit 0, got {context.w_exit_code}. Output: {context.w_output}"
|
|
)
|
|
|
|
|
|
@then("the wired context inspect command should fail")
|
|
def step_wired_inspect_fail(context: Any) -> None:
|
|
assert context.w_exit_code != 0, (
|
|
f"Expected non-zero exit, got {context.w_exit_code}"
|
|
)
|
|
|
|
|
|
@then("the wired context simulate command should succeed")
|
|
def step_wired_simulate_ok(context: Any) -> None:
|
|
assert context.w_exit_code == 0, (
|
|
f"Expected exit 0, got {context.w_exit_code}. Output: {context.w_output}"
|
|
)
|
|
|
|
|
|
@then("the wired context simulate command should fail")
|
|
def step_wired_simulate_fail(context: Any) -> None:
|
|
assert context.w_exit_code != 0, (
|
|
f"Expected non-zero exit, got {context.w_exit_code}"
|
|
)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Then assertions — ACMS config
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
def _read_stored_acms_config(context: Any) -> dict:
|
|
"""Read the ACMS config from the test DB."""
|
|
from cleveragents.cli.commands.project_context import _read_acms_config
|
|
|
|
return _read_acms_config(context.w_session_factory, "local/wire-app")
|
|
|
|
|
|
@then("the stored ACMS config hot_max_tokens should be {val:d}")
|
|
def step_acms_hot_max_tokens(context: Any, val: int) -> None:
|
|
acms = _read_stored_acms_config(context)
|
|
assert acms["hot_max_tokens"] == val, (
|
|
f"Expected {val}, got {acms['hot_max_tokens']}"
|
|
)
|
|
|
|
|
|
@then("the stored ACMS config warm_max_decisions should be {val:d}")
|
|
def step_acms_warm_max_decisions(context: Any, val: int) -> None:
|
|
acms = _read_stored_acms_config(context)
|
|
assert acms["warm_max_decisions"] == val, (
|
|
f"Expected {val}, got {acms['warm_max_decisions']}"
|
|
)
|
|
|
|
|
|
@then("the stored ACMS config cold_max_decisions should be {val:d}")
|
|
def step_acms_cold_max_decisions(context: Any, val: int) -> None:
|
|
acms = _read_stored_acms_config(context)
|
|
assert acms["cold_max_decisions"] == val, (
|
|
f"Expected {val}, got {acms['cold_max_decisions']}"
|
|
)
|
|
|
|
|
|
@then('the stored ACMS config temporal_scope should be "{val}"')
|
|
def step_acms_temporal_scope(context: Any, val: str) -> None:
|
|
acms = _read_stored_acms_config(context)
|
|
assert acms["temporal_scope"] == val, (
|
|
f"Expected {val}, got {acms['temporal_scope']}"
|
|
)
|
|
|
|
|
|
@then("the stored ACMS config auto_refresh should be false")
|
|
def step_acms_auto_refresh_false(context: Any) -> None:
|
|
acms = _read_stored_acms_config(context)
|
|
assert acms["auto_refresh"] is False, f"Expected False, got {acms['auto_refresh']}"
|
|
|
|
|
|
@then("the stored ACMS config summarize should be false")
|
|
def step_acms_summarize_false(context: Any) -> None:
|
|
acms = _read_stored_acms_config(context)
|
|
assert acms["summarize"] is False, f"Expected False, got {acms['summarize']}"
|
|
|
|
|
|
@then('the stored ACMS config strategies should contain "{strat}"')
|
|
def step_acms_strategies_contain(context: Any, strat: str) -> None:
|
|
acms = _read_stored_acms_config(context)
|
|
assert strat in acms.get("strategies", []), (
|
|
f"Expected {strat} in {acms.get('strategies', [])}"
|
|
)
|
|
|
|
|
|
@then("the stored ACMS config skeleton_ratio should be {val:g}")
|
|
def step_acms_skeleton_ratio(context: Any, val: float) -> None:
|
|
acms = _read_stored_acms_config(context)
|
|
assert abs(acms["skeleton_ratio"] - val) < 1e-9, (
|
|
f"Expected {val}, got {acms['skeleton_ratio']}"
|
|
)
|
|
|
|
|
|
@then("the stored ACMS config default_breadth should be {val:d}")
|
|
def step_acms_default_breadth(context: Any, val: int) -> None:
|
|
acms = _read_stored_acms_config(context)
|
|
assert acms["default_breadth"] == val, (
|
|
f"Expected {val}, got {acms['default_breadth']}"
|
|
)
|
|
|
|
|
|
@then("the stored ACMS config default_depth should be {val:d}")
|
|
def step_acms_default_depth(context: Any, val: int) -> None:
|
|
acms = _read_stored_acms_config(context)
|
|
assert acms["default_depth"] == val, f"Expected {val}, got {acms['default_depth']}"
|
|
|
|
|
|
@then("the stored ACMS config query_limit should be {val:d}")
|
|
def step_acms_query_limit(context: Any, val: int) -> None:
|
|
acms = _read_stored_acms_config(context)
|
|
assert acms["query_limit"] == val, f"Expected {val}, got {acms['query_limit']}"
|
|
|
|
|
|
@then("the stored ACMS config summary_max_tokens should be {val:d}")
|
|
def step_acms_summary_max_tokens(context: Any, val: int) -> None:
|
|
acms = _read_stored_acms_config(context)
|
|
assert acms["summary_max_tokens"] == val, (
|
|
f"Expected {val}, got {acms['summary_max_tokens']}"
|
|
)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Then assertions — JSON output
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@then("the wired output should be valid JSON")
|
|
def step_wired_output_valid_json(context: Any) -> None:
|
|
try:
|
|
context.w_parsed_json = json.loads(context.w_output.strip())
|
|
except json.JSONDecodeError as exc:
|
|
raise AssertionError(
|
|
f"Output is not valid JSON: {exc}\nOutput: {context.w_output!r}"
|
|
) from exc
|
|
|
|
|
|
@then('the wired JSON output should contain key "{key}"')
|
|
def step_wired_json_has_key(context: Any, key: str) -> None:
|
|
data = getattr(context, "w_parsed_json", None)
|
|
if data is None:
|
|
data = json.loads(context.w_output.strip())
|
|
context.w_parsed_json = data
|
|
assert key in data, f"Key '{key}' not found in JSON output: {list(data.keys())}"
|
|
|
|
|
|
@then("the wired JSON project_fragments total should be {val:d}")
|
|
def step_wired_json_fragments_total(context: Any, val: int) -> None:
|
|
data = getattr(context, "w_parsed_json", None)
|
|
if data is None:
|
|
data = json.loads(context.w_output.strip())
|
|
context.w_parsed_json = data
|
|
total = data.get("project_fragments", {}).get("total", 0)
|
|
assert total == val, f"Expected {val} fragments, got {total}"
|
|
|
|
|
|
@then("the wired JSON total_tokens should be greater than 0")
|
|
def step_wired_json_total_tokens_positive(context: Any) -> None:
|
|
data = getattr(context, "w_parsed_json", None)
|
|
if data is None:
|
|
data = json.loads(context.w_output.strip())
|
|
context.w_parsed_json = data
|
|
total = data.get("total_tokens", 0)
|
|
assert total > 0, f"Expected total_tokens > 0, got {total}"
|
|
|
|
|
|
@then("the wired JSON tier_metrics hot_count should be {val:d}")
|
|
def step_wired_json_tier_metrics_hot_count(context: Any, val: int) -> None:
|
|
data = getattr(context, "w_parsed_json", None)
|
|
if data is None:
|
|
data = json.loads(context.w_output.strip())
|
|
context.w_parsed_json = data
|
|
hot_count = data.get("tier_metrics", {}).get("hot_count", -1)
|
|
assert hot_count == val, f"Expected tier_metrics.hot_count={val}, got {hot_count}"
|
|
|
|
|
|
@then("the wired JSON fragment_count should be {val:d}")
|
|
def step_wired_json_fragment_count(context: Any, val: int) -> None:
|
|
data = getattr(context, "w_parsed_json", None)
|
|
if data is None:
|
|
data = json.loads(context.w_output.strip())
|
|
context.w_parsed_json = data
|
|
count = data.get("fragment_count", -1)
|
|
assert count == val, f"Expected fragment_count={val}, got {count}"
|
|
|
|
|
|
@then("the wired JSON total_tokens should be {val:d}")
|
|
def step_wired_json_total_tokens_exact(context: Any, val: int) -> None:
|
|
data = getattr(context, "w_parsed_json", None)
|
|
if data is None:
|
|
data = json.loads(context.w_output.strip())
|
|
context.w_parsed_json = data
|
|
total = data.get("total_tokens", -1)
|
|
assert total == val, f"Expected total_tokens={val}, got {total}"
|