forked from HAL9000/cleveragents-core
freemo
051ee7c290
Added 52 new .feature files and corresponding _steps.py files targeting previously uncovered code paths in the following areas: - TUI layer: app, commands, persona (state/schema/registry), widgets, input (shell_exec, reference_parser) - Application services: plan lifecycle/service/executor, session, project, repo indexing, correction, checkpoint, actor, llm_actors, strategy coordinator, resource file watcher, service retry wiring - CLI commands: session, resource, repl, plan, db, automation_profile - Domain models: retry_policy, resource_type, cost_budget, docker_compose_analyzer, detail_level, _sql_string_aware, _postgresql_helpers - Core: circuit_breaker, retry_service_patterns - Infrastructure: repositories, transaction_sandbox, strategy_registry, plugins/loader, container - Config: settings - Agents: plan_generation, context_analysis, auto_debug - A2A: facade All new tests follow the Behave/Gherkin BDD standard. Resolved step definition collisions with unique prefixes. Fixed Alembic fileConfig logger disabling issue (disable_existing_loggers=False). ISSUES CLOSED: #1068
674 lines
22 KiB
Python
674 lines
22 KiB
Python
"""Step definitions for session_cli_coverage_boost.feature.
|
|
|
|
These steps target specific uncovered lines in session.py:
|
|
- Lines 156-162: create command DatabaseError handler
|
|
- Lines 184-190: list_sessions command DatabaseError handler
|
|
- Lines 306-317: show command cost_budget display (with max_cost and unlimited)
|
|
- Lines 134-153: create command happy path (rich and json formats)
|
|
- Lines 192-200: list_sessions empty sessions path
|
|
- Lines 205-231: list_sessions with sessions (rich table + summary)
|
|
- Lines 254-303: show command rich output (messages, plans, token usage)
|
|
- Lines 327-336: show command error handlers
|
|
- Lines 362-382: delete command paths
|
|
- Lines 410-441: export command paths
|
|
- Lines 459-492: import command paths
|
|
- Lines 495-567: tell command paths (stream, no stream, actor override)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from typer.testing import CliRunner
|
|
|
|
from cleveragents.cli.commands.session import app as session_app
|
|
from cleveragents.core.exceptions import DatabaseError
|
|
from cleveragents.domain.models.core.cost_budget import SessionCostBudget
|
|
from cleveragents.domain.models.core.session import (
|
|
MessageRole,
|
|
Session,
|
|
SessionExportError,
|
|
SessionImportError,
|
|
SessionMessage,
|
|
SessionNotFoundError,
|
|
SessionService,
|
|
SessionTokenUsage,
|
|
)
|
|
|
|
_runner = CliRunner()
|
|
|
|
# Valid ULIDs for tests (Crockford base32, 26 chars)
|
|
_ULID1 = "01SCVBST000000000000000001"
|
|
_ULID2 = "01SCVBST000000000000000002"
|
|
_ULID3 = "01SCVBST000000000000000003"
|
|
_PLAN_ID1 = "01PLANID0000000000000000A1"
|
|
_PLAN_ID2 = "01PLANID0000000000000000A2"
|
|
_NOW = datetime(2025, 7, 1, 10, 30, 0)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_session(
|
|
*,
|
|
session_id: str = _ULID1,
|
|
actor_name: str | None = None,
|
|
namespace: str = "local",
|
|
messages: list[SessionMessage] | None = None,
|
|
linked_plan_ids: list[str] | None = None,
|
|
token_usage: SessionTokenUsage | None = None,
|
|
metadata: dict[str, Any] | None = None,
|
|
cost_budget: SessionCostBudget | None = None,
|
|
) -> Session:
|
|
return Session(
|
|
session_id=session_id,
|
|
actor_name=actor_name,
|
|
namespace=namespace,
|
|
messages=messages or [],
|
|
linked_plan_ids=linked_plan_ids or [],
|
|
token_usage=token_usage or SessionTokenUsage(),
|
|
created_at=_NOW,
|
|
updated_at=_NOW,
|
|
metadata=metadata or {},
|
|
cost_budget=cost_budget,
|
|
)
|
|
|
|
|
|
def _make_message(
|
|
content: str = "hello world",
|
|
role: MessageRole = MessageRole.USER,
|
|
sequence: int = 0,
|
|
) -> SessionMessage:
|
|
return SessionMessage(
|
|
message_id=_ULID2,
|
|
role=role,
|
|
content=content,
|
|
sequence=sequence,
|
|
timestamp=_NOW,
|
|
)
|
|
|
|
|
|
def _mock_service() -> MagicMock:
|
|
return MagicMock(spec=SessionService)
|
|
|
|
|
|
def _patch_service(context, svc):
|
|
patcher = patch("cleveragents.cli.commands.session._service", svc)
|
|
patcher.start()
|
|
context._scvbst_cleanups.append(patcher.stop)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the session CLI coverage boost module is set up")
|
|
def step_background(context):
|
|
context._scvbst_cleanups = []
|
|
|
|
def cleanup():
|
|
for fn in reversed(context._scvbst_cleanups):
|
|
fn()
|
|
|
|
context.add_cleanup(cleanup)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# create: happy path rich
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("session coverage boost a mock service that creates a session")
|
|
def step_create_service(context):
|
|
svc = _mock_service()
|
|
session = _make_session(actor_name="openai/gpt-4")
|
|
svc.create.return_value = session
|
|
_patch_service(context, svc)
|
|
|
|
|
|
@when("session coverage boost I invoke the create command with rich format")
|
|
def step_invoke_create_rich(context):
|
|
context.result = _runner.invoke(session_app, ["create", "--format", "rich"])
|
|
|
|
|
|
@when("session coverage boost I invoke the create command with json format")
|
|
def step_invoke_create_json(context):
|
|
context.result = _runner.invoke(session_app, ["create", "--format", "json"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# create: DatabaseError
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("session coverage boost a mock service that raises DatabaseError on create")
|
|
def step_create_db_error(context):
|
|
svc = _mock_service()
|
|
svc.create.side_effect = DatabaseError("connection refused")
|
|
_patch_service(context, svc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# list: happy path with sessions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("session coverage boost a mock service that lists sessions")
|
|
def step_list_service(context):
|
|
svc = _mock_service()
|
|
sessions = [
|
|
_make_session(session_id=_ULID1, actor_name="openai/gpt-4"),
|
|
_make_session(session_id=_ULID2, actor_name=None),
|
|
]
|
|
# Give the second session some messages for message_count variety
|
|
sessions[0].messages = [_make_message(content="msg1", sequence=0)]
|
|
svc.list.return_value = sessions
|
|
_patch_service(context, svc)
|
|
|
|
|
|
@when("session coverage boost I invoke the list command with rich format")
|
|
def step_invoke_list_rich(context):
|
|
context.result = _runner.invoke(session_app, ["list", "--format", "rich"])
|
|
|
|
|
|
@when("session coverage boost I invoke the list command with json format")
|
|
def step_invoke_list_json(context):
|
|
context.result = _runner.invoke(session_app, ["list", "--format", "json"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# list: empty sessions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("session coverage boost a mock service that lists empty sessions")
|
|
def step_list_empty_service(context):
|
|
svc = _mock_service()
|
|
svc.list.return_value = []
|
|
_patch_service(context, svc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# list: DatabaseError
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("session coverage boost a mock service that raises DatabaseError on list")
|
|
def step_list_db_error(context):
|
|
svc = _mock_service()
|
|
svc.list.side_effect = DatabaseError("disk full")
|
|
_patch_service(context, svc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# show: cost budget with max cost
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
"session coverage boost a mock service returning a session with cost budget max 10"
|
|
)
|
|
def step_show_cost_budget_max(context):
|
|
svc = _mock_service()
|
|
budget = SessionCostBudget(max_cost_usd=10.0, total_cost=3.5)
|
|
session = _make_session(
|
|
cost_budget=budget,
|
|
token_usage=SessionTokenUsage(
|
|
input_tokens=100, output_tokens=50, estimated_cost=0.05
|
|
),
|
|
)
|
|
svc.get.return_value = session
|
|
_patch_service(context, svc)
|
|
context.scvbst_session_id = _ULID1
|
|
|
|
|
|
@when("session coverage boost I invoke the show command")
|
|
def step_invoke_show(context):
|
|
sid = getattr(context, "scvbst_session_id", _ULID1)
|
|
context.result = _runner.invoke(session_app, ["show", sid])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# show: cost budget unlimited
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
"session coverage boost a mock service returning a session with unlimited cost budget"
|
|
)
|
|
def step_show_cost_budget_unlimited(context):
|
|
svc = _mock_service()
|
|
budget = SessionCostBudget(max_cost_usd=None, total_cost=1.23)
|
|
session = _make_session(
|
|
cost_budget=budget,
|
|
token_usage=SessionTokenUsage(
|
|
input_tokens=200, output_tokens=100, estimated_cost=0.1
|
|
),
|
|
)
|
|
svc.get.return_value = session
|
|
_patch_service(context, svc)
|
|
context.scvbst_session_id = _ULID1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# show: with messages and linked plans
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
"session coverage boost a mock service returning a session with messages and plans"
|
|
)
|
|
def step_show_messages_plans(context):
|
|
svc = _mock_service()
|
|
msgs = [
|
|
_make_message(content="Hello there", role=MessageRole.USER, sequence=0),
|
|
SessionMessage(
|
|
message_id=_ULID3,
|
|
role=MessageRole.ASSISTANT,
|
|
content="Hi! How can I help?",
|
|
sequence=1,
|
|
timestamp=_NOW,
|
|
),
|
|
]
|
|
session = _make_session(
|
|
messages=msgs,
|
|
linked_plan_ids=[_PLAN_ID1, _PLAN_ID2],
|
|
actor_name="openai/gpt-4",
|
|
token_usage=SessionTokenUsage(
|
|
input_tokens=500, output_tokens=300, estimated_cost=0.15
|
|
),
|
|
)
|
|
svc.get.return_value = session
|
|
_patch_service(context, svc)
|
|
context.scvbst_session_id = _ULID1
|
|
|
|
|
|
@when("session coverage boost I invoke the show command with json format")
|
|
def step_invoke_show_json(context):
|
|
sid = getattr(context, "scvbst_session_id", _ULID1)
|
|
context.result = _runner.invoke(session_app, ["show", sid, "--format", "json"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# show: SessionNotFoundError
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("session coverage boost a mock service that raises SessionNotFoundError on get")
|
|
def step_show_not_found(context):
|
|
svc = _mock_service()
|
|
svc.get.side_effect = SessionNotFoundError("no such session")
|
|
_patch_service(context, svc)
|
|
context.scvbst_session_id = _ULID1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# show: DatabaseError
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("session coverage boost a mock service that raises DatabaseError on get")
|
|
def step_show_db_error(context):
|
|
svc = _mock_service()
|
|
svc.get.side_effect = DatabaseError("connection timeout")
|
|
_patch_service(context, svc)
|
|
context.scvbst_session_id = _ULID1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# show: long message truncation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("session coverage boost a mock service returning a session with a long message")
|
|
def step_show_long_message(context):
|
|
svc = _mock_service()
|
|
long_content = "X" * 120 # > 80 chars, triggers truncation
|
|
msgs = [_make_message(content=long_content, sequence=0)]
|
|
session = _make_session(messages=msgs)
|
|
svc.get.return_value = session
|
|
_patch_service(context, svc)
|
|
context.scvbst_session_id = _ULID1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# delete: happy path with --yes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("session coverage boost a mock service for delete")
|
|
def step_delete_service(context):
|
|
svc = _mock_service()
|
|
session = _make_session()
|
|
svc.get.return_value = session
|
|
svc.delete.return_value = None
|
|
_patch_service(context, svc)
|
|
|
|
|
|
@when("session coverage boost I invoke the delete command with yes")
|
|
def step_invoke_delete_yes(context):
|
|
context.result = _runner.invoke(session_app, ["delete", _ULID1, "--yes"])
|
|
|
|
|
|
@when("session coverage boost I invoke the delete command without yes answering no")
|
|
def step_invoke_delete_no(context):
|
|
context.result = _runner.invoke(session_app, ["delete", _ULID1], input="n\n")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# delete: DatabaseError
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("session coverage boost a mock service that raises DatabaseError on delete")
|
|
def step_delete_db_error(context):
|
|
svc = _mock_service()
|
|
session = _make_session()
|
|
svc.get.return_value = session
|
|
svc.delete.side_effect = DatabaseError("lock timeout")
|
|
_patch_service(context, svc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# export: happy path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("session coverage boost a mock service for export")
|
|
def step_export_service(context):
|
|
svc = _mock_service()
|
|
svc.export_session.return_value = {
|
|
"schema_version": 1,
|
|
"session_id": _ULID1,
|
|
"actor_name": None,
|
|
"messages": [],
|
|
"checksum": "abc123",
|
|
}
|
|
_patch_service(context, svc)
|
|
|
|
|
|
@when("session coverage boost I invoke the export command to stdout")
|
|
def step_invoke_export_stdout(context):
|
|
context.result = _runner.invoke(session_app, ["export", _ULID1])
|
|
|
|
|
|
@given("session coverage boost a temporary directory for export")
|
|
def step_export_tmpdir(context):
|
|
tmpdir = tempfile.mkdtemp()
|
|
context.scvbst_export_path = Path(tmpdir) / "export.json"
|
|
context._scvbst_cleanups.append(
|
|
lambda: context.scvbst_export_path.unlink(missing_ok=True)
|
|
)
|
|
|
|
|
|
@when("session coverage boost I invoke the export command to a file")
|
|
def step_invoke_export_file(context):
|
|
context.result = _runner.invoke(
|
|
session_app,
|
|
["export", _ULID1, "--output", str(context.scvbst_export_path)],
|
|
)
|
|
|
|
|
|
@given("session coverage boost a temporary directory with existing file")
|
|
def step_export_existing_file(context):
|
|
tmpdir = tempfile.mkdtemp()
|
|
context.scvbst_export_path = Path(tmpdir) / "export.json"
|
|
context.scvbst_export_path.write_text("{}")
|
|
context._scvbst_cleanups.append(
|
|
lambda: context.scvbst_export_path.unlink(missing_ok=True)
|
|
)
|
|
|
|
|
|
@when(
|
|
"session coverage boost I invoke the export command to existing file without force"
|
|
)
|
|
def step_invoke_export_no_force(context):
|
|
context.result = _runner.invoke(
|
|
session_app,
|
|
["export", _ULID1, "--output", str(context.scvbst_export_path)],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# export: errors
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
"session coverage boost a mock service that raises SessionNotFoundError on export"
|
|
)
|
|
def step_export_not_found(context):
|
|
svc = _mock_service()
|
|
svc.export_session.side_effect = SessionNotFoundError("gone")
|
|
_patch_service(context, svc)
|
|
|
|
|
|
@given("session coverage boost a mock service that raises SessionExportError on export")
|
|
def step_export_error(context):
|
|
svc = _mock_service()
|
|
svc.export_session.side_effect = SessionExportError("checksum fail")
|
|
_patch_service(context, svc)
|
|
|
|
|
|
@given("session coverage boost a mock service that raises DatabaseError on export")
|
|
def step_export_db_error(context):
|
|
svc = _mock_service()
|
|
svc.export_session.side_effect = DatabaseError("db locked")
|
|
_patch_service(context, svc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# import: happy path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("session coverage boost a mock service for import")
|
|
def step_import_service(context):
|
|
svc = _mock_service()
|
|
session = _make_session(actor_name="test/actor")
|
|
svc.import_session.return_value = session
|
|
_patch_service(context, svc)
|
|
|
|
|
|
@given("session coverage boost a temporary import file with valid JSON")
|
|
def step_import_valid_json(context):
|
|
tmpdir = tempfile.mkdtemp()
|
|
context.scvbst_import_path = Path(tmpdir) / "import.json"
|
|
data = {
|
|
"schema_version": 1,
|
|
"session_id": _ULID1,
|
|
"actor_name": "test/actor",
|
|
"messages": [],
|
|
}
|
|
context.scvbst_import_path.write_text(json.dumps(data))
|
|
context._scvbst_cleanups.append(
|
|
lambda: context.scvbst_import_path.unlink(missing_ok=True)
|
|
)
|
|
|
|
|
|
@when("session coverage boost I invoke the import command")
|
|
def step_invoke_import(context):
|
|
context.result = _runner.invoke(
|
|
session_app,
|
|
["import", "--input", str(context.scvbst_import_path)],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# import: file not found
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("session coverage boost I invoke the import command with missing file")
|
|
def step_invoke_import_missing(context):
|
|
context.result = _runner.invoke(
|
|
session_app,
|
|
["import", "--input", "/nonexistent/path/session.json"],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# import: invalid JSON
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("session coverage boost a temporary import file with invalid JSON")
|
|
def step_import_invalid_json(context):
|
|
tmpdir = tempfile.mkdtemp()
|
|
context.scvbst_import_invalid_path = Path(tmpdir) / "bad.json"
|
|
context.scvbst_import_invalid_path.write_text("{not valid json!!!")
|
|
context._scvbst_cleanups.append(
|
|
lambda: context.scvbst_import_invalid_path.unlink(missing_ok=True)
|
|
)
|
|
|
|
|
|
@when("session coverage boost I invoke the import command with invalid json")
|
|
def step_invoke_import_invalid(context):
|
|
context.result = _runner.invoke(
|
|
session_app,
|
|
["import", "--input", str(context.scvbst_import_invalid_path)],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# import: SessionImportError
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("session coverage boost a mock service that raises SessionImportError on import")
|
|
def step_import_error(context):
|
|
svc = _mock_service()
|
|
svc.import_session.side_effect = SessionImportError("schema mismatch")
|
|
_patch_service(context, svc)
|
|
|
|
|
|
@when("session coverage boost I invoke the import command expecting import error")
|
|
def step_invoke_import_error(context):
|
|
context.result = _runner.invoke(
|
|
session_app,
|
|
["import", "--input", str(context.scvbst_import_path)],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# import: DatabaseError
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("session coverage boost a mock service that raises DatabaseError on import")
|
|
def step_import_db_error(context):
|
|
svc = _mock_service()
|
|
svc.import_session.side_effect = DatabaseError("import db fail")
|
|
_patch_service(context, svc)
|
|
|
|
|
|
@when("session coverage boost I invoke the import command expecting database error")
|
|
def step_invoke_import_db_error(context):
|
|
context.result = _runner.invoke(
|
|
session_app,
|
|
["import", "--input", str(context.scvbst_import_path)],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# tell: happy path without stream
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("session coverage boost a mock service for tell")
|
|
def step_tell_service(context):
|
|
svc = _mock_service()
|
|
svc.append_message.return_value = None
|
|
_patch_service(context, svc)
|
|
|
|
|
|
@when("session coverage boost I invoke the tell command without stream")
|
|
def step_invoke_tell_no_stream(context):
|
|
context.result = _runner.invoke(
|
|
session_app,
|
|
["tell", "--session", _ULID1, "Hello world"],
|
|
)
|
|
|
|
|
|
@when("session coverage boost I invoke the tell command with stream")
|
|
def step_invoke_tell_stream(context):
|
|
context.result = _runner.invoke(
|
|
session_app,
|
|
["tell", "--session", _ULID1, "--stream", "Hello stream"],
|
|
)
|
|
|
|
|
|
@when("session coverage boost I invoke the tell command with actor override")
|
|
def step_invoke_tell_actor(context):
|
|
context.result = _runner.invoke(
|
|
session_app,
|
|
[
|
|
"tell",
|
|
"--session",
|
|
_ULID1,
|
|
"--actor",
|
|
"openai/gpt-4",
|
|
"Plan a feature",
|
|
],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# tell: errors
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
"session coverage boost a mock service that raises SessionNotFoundError on append"
|
|
)
|
|
def step_tell_not_found(context):
|
|
svc = _mock_service()
|
|
svc.append_message.side_effect = SessionNotFoundError("no session")
|
|
_patch_service(context, svc)
|
|
|
|
|
|
@given("session coverage boost a mock service that raises DatabaseError on append")
|
|
def step_tell_db_error(context):
|
|
svc = _mock_service()
|
|
svc.append_message.side_effect = DatabaseError("tell db fail")
|
|
_patch_service(context, svc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shared assertions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("session coverage boost the exit code is {code:d}")
|
|
def step_assert_exit_code(context, code):
|
|
assert context.result.exit_code == code, (
|
|
f"Expected exit code {code}, got {context.result.exit_code}. "
|
|
f"Output:\n{context.result.output}"
|
|
)
|
|
|
|
|
|
@then('session coverage boost the output contains "{text}"')
|
|
def step_assert_output_contains(context, text):
|
|
combined = context.result.output
|
|
assert text in combined, f"Expected '{text}' in output. Got:\n{combined}"
|
|
|
|
|
|
@then("session coverage boost the output contains truncated message indicator")
|
|
def step_assert_truncated(context):
|
|
"""Check that long message content was truncated (by code or Rich)."""
|
|
combined = context.result.output
|
|
# The code truncates to content[:77] + "..." but Rich table may further
|
|
# truncate with a Unicode ellipsis "…". Either indicates truncation.
|
|
assert "..." in combined or "\u2026" in combined, (
|
|
f"Expected truncation indicator ('...' or '…') in output. Got:\n{combined}"
|
|
)
|