forked from HAL9000/cleveragents-core
362 lines
12 KiB
Python
362 lines
12 KiB
Python
"""Step definitions for invariant_cli_new_coverage.feature.
|
|
|
|
Tests all CLI commands and helper functions in
|
|
cleveragents.cli.commands.invariant to bring coverage from 0% to high.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import typer
|
|
from behave import given, then, when
|
|
from typer.testing import CliRunner
|
|
|
|
from cleveragents.cli.commands.invariant import (
|
|
_get_service,
|
|
_invariant_dict,
|
|
_resolve_scope,
|
|
app,
|
|
)
|
|
from cleveragents.core.exceptions import CleverAgentsError, NotFoundError
|
|
from cleveragents.domain.models.core.invariant import Invariant, InvariantScope
|
|
|
|
_runner = CliRunner()
|
|
|
|
_ULID = "01JTEST0000000000000000001"
|
|
_ULID2 = "01JTEST0000000000000000002"
|
|
_NOW = datetime(2025, 7, 1, 12, 0, 0)
|
|
|
|
|
|
def _make_invariant(
|
|
*,
|
|
inv_id: str = _ULID,
|
|
text: str = "Never delete prod data",
|
|
scope: InvariantScope = InvariantScope.GLOBAL,
|
|
source_name: str = "system",
|
|
active: bool = True,
|
|
created_at: datetime = _NOW,
|
|
) -> Invariant:
|
|
"""Build a test Invariant with sensible defaults."""
|
|
return Invariant(
|
|
id=inv_id,
|
|
text=text,
|
|
scope=scope,
|
|
source_name=source_name,
|
|
active=active,
|
|
created_at=created_at,
|
|
)
|
|
|
|
|
|
def _patch_svc(context, svc):
|
|
"""Patch the module-level _service and register cleanup."""
|
|
patcher = patch("cleveragents.cli.commands.invariant._service", svc)
|
|
patcher.start()
|
|
context.add_cleanup(patcher.stop)
|
|
|
|
|
|
# ================================================================
|
|
# _resolve_scope helper steps
|
|
# ================================================================
|
|
|
|
|
|
@when("I resolve invariant scope with global flag set")
|
|
def step_resolve_scope_global(context):
|
|
context.resolved_scope, context.resolved_source = _resolve_scope(
|
|
is_global=True, project=None, plan=None, action=None
|
|
)
|
|
|
|
|
|
@when('I resolve invariant scope with project "{project}"')
|
|
def step_resolve_scope_project(context, project):
|
|
context.resolved_scope, context.resolved_source = _resolve_scope(
|
|
is_global=False, project=project, plan=None, action=None
|
|
)
|
|
|
|
|
|
@when('I resolve invariant scope with plan "{plan}"')
|
|
def step_resolve_scope_plan(context, plan):
|
|
context.resolved_scope, context.resolved_source = _resolve_scope(
|
|
is_global=False, project=None, plan=plan, action=None
|
|
)
|
|
|
|
|
|
@when('I resolve invariant scope with action "{action}"')
|
|
def step_resolve_scope_action(context, action):
|
|
context.resolved_scope, context.resolved_source = _resolve_scope(
|
|
is_global=False, project=None, plan=None, action=action
|
|
)
|
|
|
|
|
|
@when("I resolve invariant scope with no flags")
|
|
def step_resolve_scope_default(context):
|
|
context.resolved_scope, context.resolved_source = _resolve_scope(
|
|
is_global=False, project=None, plan=None, action=None
|
|
)
|
|
|
|
|
|
@when("I resolve invariant scope with global and project flags")
|
|
def step_resolve_scope_conflicting(context):
|
|
context.inv_bad_parameter_raised = False
|
|
try:
|
|
_resolve_scope(is_global=True, project="myapp", plan=None, action=None)
|
|
except typer.BadParameter:
|
|
context.inv_bad_parameter_raised = True
|
|
|
|
|
|
@then('the resolved invariant scope should be "{scope}"')
|
|
def step_check_resolved_scope(context, scope):
|
|
assert context.resolved_scope.value == scope, (
|
|
f"Expected scope '{scope}', got '{context.resolved_scope.value}'"
|
|
)
|
|
|
|
|
|
@then('the resolved invariant source name should be "{source}"')
|
|
def step_check_resolved_source(context, source):
|
|
assert context.resolved_source == source, (
|
|
f"Expected source '{source}', got '{context.resolved_source}'"
|
|
)
|
|
|
|
|
|
@then("a BadParameter error should be raised for invariant scope")
|
|
def step_check_bad_parameter(context):
|
|
assert context.inv_bad_parameter_raised, "Expected typer.BadParameter to be raised"
|
|
|
|
|
|
# ================================================================
|
|
# _invariant_dict helper steps
|
|
# ================================================================
|
|
|
|
|
|
@given("a sample invariant with known fields for dict test")
|
|
def step_create_sample_invariant(context):
|
|
context.sample_inv = _make_invariant()
|
|
|
|
|
|
@when("I call invariant_dict on the sample invariant")
|
|
def step_call_invariant_dict(context):
|
|
context.inv_dict = _invariant_dict(context.sample_inv)
|
|
|
|
|
|
@then("the invariant dict should contain the correct id and text")
|
|
def step_check_dict_id_text(context):
|
|
d = context.inv_dict
|
|
assert d["id"] == _ULID
|
|
assert d["text"] == "Never delete prod data"
|
|
|
|
|
|
@then("the invariant dict should contain scope and source_name")
|
|
def step_check_dict_scope_source(context):
|
|
d = context.inv_dict
|
|
assert d["scope"] == "global"
|
|
assert d["source_name"] == "system"
|
|
|
|
|
|
@then("the invariant dict should contain active and created_at ISO string")
|
|
def step_check_dict_active_created(context):
|
|
d = context.inv_dict
|
|
assert d["active"] is True
|
|
assert d["created_at"] == _NOW.isoformat()
|
|
|
|
|
|
# ================================================================
|
|
# _get_service singleton steps
|
|
# ================================================================
|
|
|
|
|
|
@when("I call get_service with invariant module service reset to None")
|
|
def step_get_service_reset(context):
|
|
import cleveragents.cli.commands.invariant as mod
|
|
|
|
# Save original and reset
|
|
context._orig_inv_service = mod._service
|
|
mod._service = None
|
|
context.add_cleanup(lambda: setattr(mod, "_service", context._orig_inv_service))
|
|
|
|
context.first_inv_service = _get_service()
|
|
|
|
|
|
@then("an InvariantService instance should be returned from get_service")
|
|
def step_check_service_instance(context):
|
|
from cleveragents.application.services.invariant_service import InvariantService
|
|
|
|
assert isinstance(context.first_inv_service, InvariantService)
|
|
|
|
|
|
@then("calling get_service again returns the same InvariantService instance")
|
|
def step_check_service_singleton(context):
|
|
second = _get_service()
|
|
assert second is context.first_inv_service
|
|
|
|
|
|
# ================================================================
|
|
# invariant add command steps
|
|
# ================================================================
|
|
|
|
|
|
@given("a mocked InvariantService for invariant CLI add")
|
|
def step_mock_service_for_add(context):
|
|
svc = MagicMock()
|
|
|
|
def _fake_add(text, scope, source_name):
|
|
return _make_invariant(text=text, scope=scope, source_name=source_name)
|
|
|
|
svc.add_invariant.side_effect = _fake_add
|
|
svc.list_invariants.return_value = []
|
|
context.inv_mock_svc = svc
|
|
_patch_svc(context, svc)
|
|
|
|
|
|
@when('I invoke invariant add with "{flags}" and text "{text}"')
|
|
def step_run_add(context, flags, text):
|
|
args = ["add", *flags.split(), text]
|
|
context.inv_result = _runner.invoke(app, args)
|
|
|
|
|
|
@then("the invariant CLI exit code should be 0")
|
|
def step_check_exit_zero(context):
|
|
assert context.inv_result.exit_code == 0, (
|
|
f"Expected exit code 0, got {context.inv_result.exit_code}.\n"
|
|
f"Output: {context.inv_result.output}"
|
|
)
|
|
|
|
|
|
@then("the invariant CLI exit code should be non-zero")
|
|
def step_check_exit_nonzero(context):
|
|
assert context.inv_result.exit_code != 0, (
|
|
f"Expected non-zero exit code, got {context.inv_result.exit_code}.\n"
|
|
f"Output: {context.inv_result.output}"
|
|
)
|
|
|
|
|
|
@then('the invariant CLI output should contain "{text}"')
|
|
def step_check_output_contains(context, text):
|
|
assert text in context.inv_result.output, (
|
|
f"Expected '{text}' in output, got:\n{context.inv_result.output}"
|
|
)
|
|
|
|
|
|
@given("a mocked InvariantService that raises CleverAgentsError on add")
|
|
def step_mock_service_error_add(context):
|
|
svc = MagicMock()
|
|
svc.add_invariant.side_effect = CleverAgentsError("Service failure")
|
|
_patch_svc(context, svc)
|
|
|
|
|
|
# ================================================================
|
|
# invariant list command steps
|
|
# ================================================================
|
|
|
|
|
|
@given("a mocked InvariantService that returns empty invariant list")
|
|
def step_mock_service_empty_list(context):
|
|
svc = MagicMock()
|
|
svc.list_invariants.return_value = []
|
|
context.inv_mock_svc = svc
|
|
_patch_svc(context, svc)
|
|
|
|
|
|
@given("a mocked InvariantService that returns two invariants for list")
|
|
def step_mock_service_two_invariants(context):
|
|
svc = MagicMock()
|
|
inv1 = _make_invariant(inv_id=_ULID, text="Never delete prod data")
|
|
inv2 = _make_invariant(
|
|
inv_id=_ULID2,
|
|
text="All changes need review",
|
|
scope=InvariantScope.PROJECT,
|
|
source_name="myapp",
|
|
)
|
|
svc.list_invariants.return_value = [inv1, inv2]
|
|
context.inv_mock_svc = svc
|
|
_patch_svc(context, svc)
|
|
|
|
|
|
@when("I invoke invariant list with no filters")
|
|
def step_run_list_no_filters(context):
|
|
context.inv_result = _runner.invoke(app, ["list"])
|
|
|
|
|
|
@when('I invoke invariant list with flags "{flags}"')
|
|
def step_run_list_with_flags(context, flags):
|
|
args = ["list", *flags.split()]
|
|
context.inv_result = _runner.invoke(app, args)
|
|
|
|
|
|
@when('I invoke invariant list with regex "{pattern}"')
|
|
def step_run_list_with_regex(context, pattern):
|
|
context.inv_result = _runner.invoke(app, ["list", pattern])
|
|
|
|
|
|
@then("the invariant service list was called with global scope")
|
|
def step_check_list_global(context):
|
|
call_kwargs = context.inv_mock_svc.list_invariants.call_args
|
|
assert call_kwargs is not None
|
|
kwargs = call_kwargs[1] if call_kwargs[1] else {}
|
|
assert kwargs.get("scope") == InvariantScope.GLOBAL
|
|
|
|
|
|
@then('the invariant service list was called with project scope and source "{source}"')
|
|
def step_check_list_project(context, source):
|
|
call_kwargs = context.inv_mock_svc.list_invariants.call_args
|
|
kwargs = call_kwargs[1] if call_kwargs[1] else {}
|
|
assert kwargs.get("scope") == InvariantScope.PROJECT
|
|
assert kwargs.get("source_name") == source
|
|
|
|
|
|
@then('the invariant service list was called with plan scope and source "{source}"')
|
|
def step_check_list_plan(context, source):
|
|
call_kwargs = context.inv_mock_svc.list_invariants.call_args
|
|
kwargs = call_kwargs[1] if call_kwargs[1] else {}
|
|
assert kwargs.get("scope") == InvariantScope.PLAN
|
|
assert kwargs.get("source_name") == source
|
|
|
|
|
|
@then('the invariant service list was called with action scope and source "{source}"')
|
|
def step_check_list_action(context, source):
|
|
call_kwargs = context.inv_mock_svc.list_invariants.call_args
|
|
kwargs = call_kwargs[1] if call_kwargs[1] else {}
|
|
assert kwargs.get("scope") == InvariantScope.ACTION
|
|
assert kwargs.get("source_name") == source
|
|
|
|
|
|
@then("the invariant service list was called with effective true")
|
|
def step_check_list_effective(context):
|
|
call_kwargs = context.inv_mock_svc.list_invariants.call_args
|
|
kwargs = call_kwargs[1] if call_kwargs[1] else {}
|
|
assert kwargs.get("effective") is True
|
|
|
|
|
|
# ================================================================
|
|
# invariant remove command steps
|
|
# ================================================================
|
|
|
|
|
|
@given("a mocked InvariantService for invariant CLI remove")
|
|
def step_mock_service_for_remove(context):
|
|
svc = MagicMock()
|
|
mock_inv = _make_invariant(active=False)
|
|
svc.remove_invariant.return_value = mock_inv
|
|
context.inv_mock_svc = svc
|
|
_patch_svc(context, svc)
|
|
|
|
|
|
@when('I invoke invariant remove with "{flags}" for id "{inv_id}"')
|
|
def step_run_remove_with_flags(context, flags, inv_id):
|
|
args = ["remove", *flags.split(), inv_id]
|
|
context.inv_result = _runner.invoke(app, args)
|
|
|
|
|
|
@when('I invoke invariant remove without --yes for id "{inv_id}" and answer no')
|
|
def step_run_remove_no_confirm(context, inv_id):
|
|
context.inv_result = _runner.invoke(app, ["remove", inv_id], input="n\n")
|
|
|
|
|
|
@given("a mocked InvariantService that raises NotFoundError on remove")
|
|
def step_mock_service_not_found_remove(context):
|
|
svc = MagicMock()
|
|
svc.remove_invariant.side_effect = NotFoundError(
|
|
resource_type="invariant", resource_id="MISSING_ID"
|
|
)
|
|
_patch_svc(context, svc)
|