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
269 lines
11 KiB
Python
269 lines
11 KiB
Python
"""Step definitions for cost_budget_model_coverage.feature.
|
|
|
|
These steps target specific uncovered lines in cost_budget.py:
|
|
- Line 97: SessionCostBudget.utilization() when max_cost_usd == 0.0
|
|
- Lines 132-134: SessionCostBudget.can_afford() (unlimited + capped paths)
|
|
- Line 155: SessionCostBudget.reset()
|
|
- Line 201: OrgCostAccumulator.utilization() when max_cost_usd == 0.0
|
|
- Lines 236-238: OrgCostAccumulator.can_afford() (unlimited + capped paths)
|
|
- Line 259: OrgCostAccumulator.reset()
|
|
- Lines 291-292: ThreadSafeOrgCostAccumulator.max_cost_usd setter
|
|
- Lines 301-302: ThreadSafeOrgCostAccumulator.org_id property
|
|
- Lines 321-322: ThreadSafeOrgCostAccumulator.can_afford()
|
|
- Lines 333-334: ThreadSafeOrgCostAccumulator.reset()
|
|
"""
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.domain.models.core.cost_budget import (
|
|
OrgCostAccumulator,
|
|
SessionCostBudget,
|
|
ThreadSafeOrgCostAccumulator,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
@given("the cost budget module is imported")
|
|
def step_cost_budget_module_imported(context):
|
|
"""Ensure all cost-budget classes are importable."""
|
|
assert SessionCostBudget is not None
|
|
assert OrgCostAccumulator is not None
|
|
assert ThreadSafeOrgCostAccumulator is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SessionCostBudget - setup helpers
|
|
# ---------------------------------------------------------------------------
|
|
@given("a session budget with max cost USD {amount:g}")
|
|
def step_session_budget_with_max(context, amount):
|
|
"""Create a SessionCostBudget with a specific cap."""
|
|
context.session_budget = SessionCostBudget(max_cost_usd=amount)
|
|
|
|
|
|
@given("a session budget with no spending cap")
|
|
def step_session_budget_unlimited(context):
|
|
"""Create a SessionCostBudget with no spending cap (unlimited)."""
|
|
context.session_budget = SessionCostBudget(max_cost_usd=None)
|
|
|
|
|
|
@given("I record a session cost of {amount:g}")
|
|
def step_record_session_cost(context, amount):
|
|
"""Record cost against the session budget."""
|
|
context.session_budget.record_cost(amount)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SessionCostBudget - utilization (line 97)
|
|
# ---------------------------------------------------------------------------
|
|
@when("I query the session utilization")
|
|
def step_query_session_utilization(context):
|
|
"""Query the utilization of the session budget."""
|
|
context.session_utilization = context.session_budget.utilization()
|
|
|
|
|
|
@then("the session utilization should be None")
|
|
def step_session_utilization_is_none(context):
|
|
"""Assert utilization is None (zero or unlimited budget)."""
|
|
assert context.session_utilization is None, (
|
|
f"Expected None, got {context.session_utilization!r}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SessionCostBudget - can_afford (lines 132-134)
|
|
# ---------------------------------------------------------------------------
|
|
@when("I check if the session can afford {amount:g}")
|
|
def step_session_can_afford(context, amount):
|
|
"""Check whether the session budget can afford the given cost."""
|
|
context.session_can_afford_result = context.session_budget.can_afford(amount)
|
|
|
|
|
|
@then("the session can_afford result should be True")
|
|
def step_session_can_afford_true(context):
|
|
assert context.session_can_afford_result is True
|
|
|
|
|
|
@then("the session can_afford result should be False")
|
|
def step_session_can_afford_false(context):
|
|
assert context.session_can_afford_result is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SessionCostBudget - reset (line 155)
|
|
# ---------------------------------------------------------------------------
|
|
@when("I reset the session budget")
|
|
def step_reset_session_budget(context):
|
|
"""Reset the session budget."""
|
|
context.session_budget.reset()
|
|
|
|
|
|
@then("the session total cost should be {amount:g}")
|
|
def step_session_total_cost(context, amount):
|
|
assert context.session_budget.total_cost == amount, (
|
|
f"Expected {amount}, got {context.session_budget.total_cost}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# OrgCostAccumulator - setup helpers
|
|
# ---------------------------------------------------------------------------
|
|
@given('an org accumulator for "{org_id}" with max cost USD {amount:g}')
|
|
def step_org_accumulator_with_max(context, org_id, amount):
|
|
"""Create an OrgCostAccumulator with a specific cap."""
|
|
context.org_accumulator = OrgCostAccumulator(org_id=org_id, max_cost_usd=amount)
|
|
|
|
|
|
@given('an org accumulator for "{org_id}" with no spending cap')
|
|
def step_org_accumulator_unlimited(context, org_id):
|
|
"""Create an OrgCostAccumulator with no spending cap."""
|
|
context.org_accumulator = OrgCostAccumulator(org_id=org_id, max_cost_usd=None)
|
|
|
|
|
|
@given("I record an org cost of {amount:g}")
|
|
def step_record_org_cost(context, amount):
|
|
"""Record cost against the org accumulator."""
|
|
context.org_accumulator.record_cost(amount)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# OrgCostAccumulator - utilization (line 201)
|
|
# ---------------------------------------------------------------------------
|
|
@when("I query the org utilization")
|
|
def step_query_org_utilization(context):
|
|
"""Query the utilization of the org accumulator."""
|
|
context.org_utilization = context.org_accumulator.utilization()
|
|
|
|
|
|
@then("the org utilization should be None")
|
|
def step_org_utilization_is_none(context):
|
|
assert context.org_utilization is None, (
|
|
f"Expected None, got {context.org_utilization!r}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# OrgCostAccumulator - can_afford (lines 236-238)
|
|
# ---------------------------------------------------------------------------
|
|
@when("I check if the org can afford {amount:g}")
|
|
def step_org_can_afford(context, amount):
|
|
"""Check whether the org accumulator can afford the given cost."""
|
|
context.org_can_afford_result = context.org_accumulator.can_afford(amount)
|
|
|
|
|
|
@then("the org can_afford result should be True")
|
|
def step_org_can_afford_true(context):
|
|
assert context.org_can_afford_result is True
|
|
|
|
|
|
@then("the org can_afford result should be False")
|
|
def step_org_can_afford_false(context):
|
|
assert context.org_can_afford_result is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# OrgCostAccumulator - reset (line 259)
|
|
# ---------------------------------------------------------------------------
|
|
@when("I reset the org accumulator")
|
|
def step_reset_org_accumulator(context):
|
|
"""Reset the org accumulator."""
|
|
context.org_accumulator.reset()
|
|
|
|
|
|
@then("the org total cost should be {amount:g}")
|
|
def step_org_total_cost(context, amount):
|
|
assert context.org_accumulator.total_cost == amount, (
|
|
f"Expected {amount}, got {context.org_accumulator.total_cost}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ThreadSafeOrgCostAccumulator - setup helpers
|
|
# ---------------------------------------------------------------------------
|
|
@given('a thread-safe org accumulator for "{org_id}"')
|
|
def step_thread_safe_accumulator(context, org_id):
|
|
"""Create a ThreadSafeOrgCostAccumulator with no cap."""
|
|
inner = OrgCostAccumulator(org_id=org_id, max_cost_usd=None)
|
|
context.ts_accumulator = ThreadSafeOrgCostAccumulator(inner)
|
|
|
|
|
|
@given('a thread-safe org accumulator for "{org_id}" with max cost USD {amount:g}')
|
|
def step_thread_safe_accumulator_with_max(context, org_id, amount):
|
|
"""Create a ThreadSafeOrgCostAccumulator with a specific cap."""
|
|
inner = OrgCostAccumulator(org_id=org_id, max_cost_usd=amount)
|
|
context.ts_accumulator = ThreadSafeOrgCostAccumulator(inner)
|
|
|
|
|
|
@given("I record a thread-safe cost of {amount:g}")
|
|
def step_record_thread_safe_cost(context, amount):
|
|
"""Record cost via the thread-safe wrapper."""
|
|
context.ts_accumulator.record_cost(amount)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ThreadSafeOrgCostAccumulator - org_id property (lines 301-302)
|
|
# ---------------------------------------------------------------------------
|
|
@when("I read the thread-safe org_id")
|
|
def step_read_ts_org_id(context):
|
|
"""Read the org_id through the thread-safe wrapper."""
|
|
context.ts_org_id = context.ts_accumulator.org_id
|
|
|
|
|
|
@then('the thread-safe org_id should be "{expected}"')
|
|
def step_verify_ts_org_id(context, expected):
|
|
assert context.ts_org_id == expected, (
|
|
f"Expected '{expected}', got '{context.ts_org_id}'"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ThreadSafeOrgCostAccumulator - max_cost_usd setter (lines 291-292)
|
|
# ---------------------------------------------------------------------------
|
|
@when("I set the thread-safe max_cost_usd to {amount:g}")
|
|
def step_set_ts_max_cost(context, amount):
|
|
"""Set the max_cost_usd through the thread-safe wrapper."""
|
|
context.ts_accumulator.max_cost_usd = amount
|
|
|
|
|
|
@then("the thread-safe max_cost_usd should be {amount:g}")
|
|
def step_verify_ts_max_cost(context, amount):
|
|
assert context.ts_accumulator.max_cost_usd == amount, (
|
|
f"Expected {amount}, got {context.ts_accumulator.max_cost_usd}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ThreadSafeOrgCostAccumulator - can_afford (lines 321-322)
|
|
# ---------------------------------------------------------------------------
|
|
@when("I check if the thread-safe accumulator can afford {amount:g}")
|
|
def step_ts_can_afford(context, amount):
|
|
"""Check can_afford through the thread-safe wrapper."""
|
|
context.ts_can_afford_result = context.ts_accumulator.can_afford(amount)
|
|
|
|
|
|
@then("the thread-safe can_afford result should be True")
|
|
def step_ts_can_afford_true(context):
|
|
assert context.ts_can_afford_result is True
|
|
|
|
|
|
@then("the thread-safe can_afford result should be False")
|
|
def step_ts_can_afford_false(context):
|
|
assert context.ts_can_afford_result is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ThreadSafeOrgCostAccumulator - reset (lines 333-334)
|
|
# ---------------------------------------------------------------------------
|
|
@when("I reset the thread-safe accumulator")
|
|
def step_reset_ts_accumulator(context):
|
|
"""Reset through the thread-safe wrapper."""
|
|
context.ts_accumulator.reset()
|
|
|
|
|
|
@then("the thread-safe total cost should be {amount:g}")
|
|
def step_ts_total_cost(context, amount):
|
|
assert context.ts_accumulator.total_cost == amount, (
|
|
f"Expected {amount}, got {context.ts_accumulator.total_cost}"
|
|
)
|