forked from HAL9000/cleveragents-core
freemo
a808c395f9
Add 53 new .feature files and corresponding step definition files targeting uncovered lines identified in build/coverage.xml. Fix AmbiguousStep conflicts in 7 pre-existing step files by disambiguating step text. New tests cover: ACP clients/facade, actor CLI/config, application container, ACMS service/strategies, async worker, automation profile CLI, autonomy guardrail, bridge, change model, config CLI/service, context service, cross-plan correction, database models, decision service, decomposition clustering/service, discovery handler, langchain chat provider, langgraph nodes, materializers, multi-project service, plan apply/CLI/lifecycle/model/ preflight/resume/service, PostgreSQL analyzer, project CLI/context CLI, provider registry, reactive application/route, repositories, resolver handler, resource registry service, resume model, retry patterns, sandbox protocol, server CLI, skill CLI/service, skills registry, subplan execution/service, system CLI, UKO loader, UoW, and YAML template engine. Closes #645
612 lines
20 KiB
Python
612 lines
20 KiB
Python
"""Step definitions for database_models_coverage_r2.feature.
|
|
|
|
Targets uncovered lines in
|
|
src/cleveragents/infrastructure/database/models.py
|
|
identified from build/coverage.xml (round 2).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
from behave import given, then, when # type: ignore[import-untyped]
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
NOW = datetime(2025, 6, 1, 12, 0, 0, tzinfo=UTC)
|
|
NOW_ISO = NOW.isoformat()
|
|
|
|
|
|
def _make_ulid(seed: int = 1) -> str:
|
|
"""Return a fake but valid-length 26-char ULID."""
|
|
return f"01ABCDEFGH{seed:016d}"
|
|
|
|
|
|
def _minimal_plan_timestamps() -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
created_at=NOW,
|
|
updated_at=NOW,
|
|
strategize_started_at=None,
|
|
strategize_completed_at=None,
|
|
execute_started_at=None,
|
|
execute_completed_at=None,
|
|
apply_started_at=None,
|
|
applied_at=None,
|
|
)
|
|
|
|
|
|
def _minimal_plan_identity(plan_id: str | None = None) -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
plan_id=plan_id or _make_ulid(1),
|
|
parent_plan_id=None,
|
|
root_plan_id=None,
|
|
attempt=1,
|
|
)
|
|
|
|
|
|
def _minimal_namespaced_name() -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
server=None,
|
|
namespace="local",
|
|
name="test-action",
|
|
__str__=lambda self: f"{self.namespace}/{self.name}",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LifecycleActionModel — safety_profile round-trip (lines 376, 427)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a LifecycleActionModel with safety_profile_json populated")
|
|
def step_action_model_with_safety(context: Any) -> None:
|
|
from cleveragents.infrastructure.database.models import (
|
|
LifecycleActionModel,
|
|
)
|
|
|
|
safety_data = {
|
|
"require_sandbox": True,
|
|
"require_checkpoints": False,
|
|
"allow_unsafe_tools": False,
|
|
}
|
|
model = LifecycleActionModel(
|
|
namespaced_name="local/test-action",
|
|
namespace="local",
|
|
name="test-action",
|
|
description="desc",
|
|
definition_of_done="dod",
|
|
strategy_actor="strat",
|
|
execution_actor="exec",
|
|
reusable=True,
|
|
read_only=False,
|
|
state="available",
|
|
tags_json="[]",
|
|
created_at=NOW_ISO,
|
|
updated_at=NOW_ISO,
|
|
safety_profile_json=json.dumps(safety_data),
|
|
)
|
|
# Ensure relationship lists exist
|
|
model.arguments_rel = []
|
|
model.invariants_rel = []
|
|
context.action_model = model
|
|
|
|
|
|
@when("I convert the action model to domain via to_domain")
|
|
def step_action_to_domain(context: Any) -> None:
|
|
context.domain_action = context.action_model.to_domain()
|
|
|
|
|
|
@then("the domain action has a non-null safety_profile object")
|
|
def step_check_safety_profile_not_none(context: Any) -> None:
|
|
sp = context.domain_action.safety_profile
|
|
assert sp is not None, "safety_profile should be deserialized"
|
|
assert sp.require_sandbox is True
|
|
|
|
|
|
@given("an Action domain object with a SafetyProfile attached")
|
|
def step_action_domain_with_safety(context: Any) -> None:
|
|
from cleveragents.domain.models.core.safety_profile import SafetyProfile
|
|
|
|
context.domain_action_input = SimpleNamespace(
|
|
namespaced_name=_minimal_namespaced_name(),
|
|
description="desc",
|
|
long_description=None,
|
|
definition_of_done="dod",
|
|
strategy_actor="strat",
|
|
execution_actor="exec",
|
|
review_actor=None,
|
|
apply_actor=None,
|
|
estimation_actor=None,
|
|
invariant_actor=None,
|
|
automation_profile=None,
|
|
safety_profile=SafetyProfile(
|
|
require_sandbox=True,
|
|
require_checkpoints=False,
|
|
allow_unsafe_tools=True,
|
|
),
|
|
reusable=True,
|
|
read_only=False,
|
|
inputs_schema=None,
|
|
state=SimpleNamespace(value="available"),
|
|
created_by=None,
|
|
tags=[],
|
|
arguments=[],
|
|
invariants=[],
|
|
created_at=NOW,
|
|
updated_at=NOW,
|
|
)
|
|
|
|
|
|
@when("I convert the action to a model via from_domain")
|
|
def step_action_from_domain(context: Any) -> None:
|
|
from cleveragents.infrastructure.database.models import (
|
|
LifecycleActionModel,
|
|
)
|
|
|
|
context.action_model_out = LifecycleActionModel.from_domain(
|
|
context.domain_action_input,
|
|
)
|
|
|
|
|
|
@then("the model safety_profile_json contains serialized SafetyProfile data")
|
|
def step_check_safety_json(context: Any) -> None:
|
|
raw = context.action_model_out.safety_profile_json
|
|
assert raw is not None, "safety_profile_json must not be None"
|
|
parsed = json.loads(raw)
|
|
assert parsed["require_sandbox"] is True
|
|
assert parsed["allow_unsafe_tools"] is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LifecycleActionModel — string arg_type/requirement (lines 466, 469)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
"an Action domain object whose arguments have plain-string arg_type and requirement"
|
|
)
|
|
def step_action_plain_string_args(context: Any) -> None:
|
|
arg = SimpleNamespace(
|
|
name="my_arg",
|
|
arg_type="string", # plain string, no .value
|
|
requirement="required", # plain string, no .value
|
|
description="a test arg",
|
|
default_value=None,
|
|
min_value=None,
|
|
max_value=None,
|
|
validation_pattern=None,
|
|
)
|
|
context.domain_action_input = SimpleNamespace(
|
|
namespaced_name=_minimal_namespaced_name(),
|
|
description="desc",
|
|
long_description=None,
|
|
definition_of_done="dod",
|
|
strategy_actor="strat",
|
|
execution_actor="exec",
|
|
review_actor=None,
|
|
apply_actor=None,
|
|
estimation_actor=None,
|
|
invariant_actor=None,
|
|
automation_profile=None,
|
|
safety_profile=None,
|
|
reusable=True,
|
|
read_only=False,
|
|
inputs_schema=None,
|
|
state=SimpleNamespace(value="available"),
|
|
created_by=None,
|
|
tags=[],
|
|
arguments=[arg],
|
|
invariants=[],
|
|
created_at=NOW,
|
|
updated_at=NOW,
|
|
)
|
|
|
|
|
|
@when("I convert that action to a model via from_domain")
|
|
def step_that_action_from_domain(context: Any) -> None:
|
|
from cleveragents.infrastructure.database.models import (
|
|
LifecycleActionModel,
|
|
)
|
|
|
|
context.action_model_out = LifecycleActionModel.from_domain(
|
|
context.domain_action_input,
|
|
)
|
|
|
|
|
|
@then("the model argument rows use the plain-string values directly")
|
|
def step_check_arg_strings(context: Any) -> None:
|
|
args = context.action_model_out.arguments_rel
|
|
assert len(args) == 1
|
|
assert args[0].arg_type == "string"
|
|
assert args[0].requirement == "required"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LifecyclePlanModel — string processing_state (line 882)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _base_plan_ns(
|
|
*,
|
|
processing_state: Any = "queued",
|
|
automation_profile: Any = None,
|
|
namespaced_name: Any = None,
|
|
error_details: Any = None,
|
|
) -> SimpleNamespace:
|
|
"""Build a minimal Plan-like namespace for from_domain tests."""
|
|
ns_name = namespaced_name or _minimal_namespaced_name()
|
|
return SimpleNamespace(
|
|
identity=_minimal_plan_identity(),
|
|
namespaced_name=ns_name,
|
|
action_name="local/test-action",
|
|
description="plan desc",
|
|
definition_of_done="dod",
|
|
phase=SimpleNamespace(value="action"),
|
|
processing_state=processing_state,
|
|
automation_profile=automation_profile,
|
|
strategy_actor=None,
|
|
execution_actor=None,
|
|
review_actor=None,
|
|
apply_actor=None,
|
|
estimation_actor=None,
|
|
invariant_actor=None,
|
|
project_links=[],
|
|
invariants=[],
|
|
arguments={},
|
|
arguments_order=[],
|
|
changeset_id=None,
|
|
sandbox_refs=[],
|
|
validation_summary=None,
|
|
decision_root_id=None,
|
|
timestamps=_minimal_plan_timestamps(),
|
|
error_message=None,
|
|
error_details=error_details,
|
|
created_by=None,
|
|
tags=[],
|
|
reusable=True,
|
|
read_only=False,
|
|
)
|
|
|
|
|
|
@given("a Plan domain object whose processing_state is a plain string")
|
|
def step_plan_string_state(context: Any) -> None:
|
|
context.plan_domain = _base_plan_ns(processing_state="complete")
|
|
|
|
|
|
@when("I convert the plan to a model via from_domain")
|
|
def step_plan_from_domain(context: Any) -> None:
|
|
from cleveragents.infrastructure.database.models import (
|
|
LifecyclePlanModel,
|
|
)
|
|
|
|
context.plan_model_out = LifecyclePlanModel.from_domain(context.plan_domain)
|
|
|
|
|
|
@then("the model processing_state equals the plain string value")
|
|
def step_check_plan_state(context: Any) -> None:
|
|
assert context.plan_model_out.processing_state == "complete"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LifecyclePlanModel — automation_profile (line 890)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a Plan domain object with a non-null automation_profile")
|
|
def step_plan_with_automation_profile(context: Any) -> None:
|
|
from cleveragents.domain.models.core.plan import (
|
|
AutomationProfileProvenance,
|
|
AutomationProfileRef,
|
|
)
|
|
|
|
profile = AutomationProfileRef(
|
|
profile_name="strict",
|
|
provenance=AutomationProfileProvenance.ACTION,
|
|
)
|
|
context.plan_domain = _base_plan_ns(automation_profile=profile)
|
|
|
|
|
|
@then("the model automation_profile column contains JSON with profile_name")
|
|
def step_check_automation_profile_json(context: Any) -> None:
|
|
raw = context.plan_model_out.automation_profile
|
|
assert raw is not None
|
|
parsed = json.loads(raw)
|
|
assert parsed["profile_name"] == "strict"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LifecyclePlanModel — fallback namespace (line 906)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a Plan domain object whose namespaced_name has no namespace attribute")
|
|
def step_plan_no_namespace(context: Any) -> None:
|
|
# Use a plain string that has no .namespace attribute
|
|
context.plan_domain = _base_plan_ns(namespaced_name="local/test-plan")
|
|
|
|
|
|
@then('the model namespace equals "local"')
|
|
def step_check_namespace_local(context: Any) -> None:
|
|
assert context.plan_model_out.namespace == "local"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LifecyclePlanModel — error_details (line 923)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a Plan domain object with non-null error_details")
|
|
def step_plan_with_error_details(context: Any) -> None:
|
|
context.plan_domain = _base_plan_ns(
|
|
error_details={"code": "E001", "msg": "something failed"},
|
|
)
|
|
|
|
|
|
@then("the model error_details_json is a JSON string of the details dict")
|
|
def step_check_error_details_json(context: Any) -> None:
|
|
raw = context.plan_model_out.error_details_json
|
|
assert raw is not None
|
|
parsed = json.loads(raw)
|
|
assert parsed["code"] == "E001"
|
|
assert parsed["msg"] == "something failed"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SkillModel — include with overrides (line 2327)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a Skill domain object with an include that has overrides")
|
|
def step_skill_with_include_overrides(context: Any) -> None:
|
|
include = SimpleNamespace(
|
|
name="other-skill/base",
|
|
overrides={"timeout": 120},
|
|
)
|
|
context.skill_domain = SimpleNamespace(
|
|
name="local/my-skill",
|
|
description="a skill",
|
|
tool_refs=[],
|
|
includes=[include],
|
|
anonymous_tools=[],
|
|
mcp_servers=[],
|
|
agent_skills=[],
|
|
overrides={},
|
|
version=None,
|
|
)
|
|
|
|
|
|
@when("I convert the skill to a model via from_domain")
|
|
def step_skill_from_domain(context: Any) -> None:
|
|
from cleveragents.infrastructure.database.models import SkillModel
|
|
|
|
context.skill_model_out = SkillModel.from_domain(context.skill_domain)
|
|
|
|
|
|
@then("the include item_config contains the serialized overrides")
|
|
def step_check_include_overrides(context: Any) -> None:
|
|
items = context.skill_model_out.items_rel
|
|
include_items = [i for i in items if i.item_type == "include"]
|
|
assert len(include_items) == 1
|
|
config = json.loads(include_items[0].item_config)
|
|
assert config["overrides"]["timeout"] == 120
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DecisionModel — string decision_type (line 2698)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a Decision domain object whose decision_type is a plain string")
|
|
def step_decision_string_type(context: Any) -> None:
|
|
from cleveragents.domain.models.core.decision import (
|
|
ContextSnapshot,
|
|
)
|
|
|
|
context.decision_domain = SimpleNamespace(
|
|
decision_id=_make_ulid(10),
|
|
plan_id=_make_ulid(11),
|
|
parent_decision_id=None,
|
|
sequence_number=1,
|
|
decision_type="strategy_choice", # plain string, no .value
|
|
question="Which strategy?",
|
|
chosen_option="Option A",
|
|
alternatives_considered=["Option B"],
|
|
confidence_score=0.9,
|
|
context_snapshot=ContextSnapshot(
|
|
hot_context_hash="abc",
|
|
hot_context_ref="ref1",
|
|
relevant_resources=[],
|
|
actor_state_ref="actor1",
|
|
),
|
|
rationale="because",
|
|
actor_reasoning=None,
|
|
downstream_decision_ids=[],
|
|
downstream_plan_ids=[],
|
|
artifacts_produced=[],
|
|
created_at=NOW,
|
|
is_correction=False,
|
|
corrects_decision_id=None,
|
|
correction_reason=None,
|
|
superseded_by=None,
|
|
)
|
|
|
|
|
|
@when("I convert the decision to a model via from_domain")
|
|
def step_decision_from_domain(context: Any) -> None:
|
|
from cleveragents.infrastructure.database.models import DecisionModel
|
|
|
|
context.decision_model_out = DecisionModel.from_domain(
|
|
context.decision_domain,
|
|
)
|
|
|
|
|
|
@then("the model decision_type equals the plain string")
|
|
def step_check_decision_type_string(context: Any) -> None:
|
|
assert context.decision_model_out.decision_type == "strategy_choice"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CheckpointModel — to_domain (lines 2823-2852)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a CheckpointModel with metadata_json containing reason and phase")
|
|
def step_checkpoint_model_with_meta(context: Any) -> None:
|
|
from cleveragents.infrastructure.database.models import CheckpointModel
|
|
|
|
context.checkpoint_model = CheckpointModel(
|
|
checkpoint_id=_make_ulid(20),
|
|
plan_id=_make_ulid(21),
|
|
sandbox_ref="abc123commit",
|
|
decision_id=None,
|
|
checkpoint_type="pre_write",
|
|
resource_id=None,
|
|
filesystem_path="checkpoints/cp1",
|
|
size_bytes=1024,
|
|
created_at=NOW_ISO,
|
|
metadata_json=json.dumps(
|
|
{
|
|
"reason": "before write",
|
|
"source_tool": "file_write",
|
|
"phase": "execute",
|
|
}
|
|
),
|
|
)
|
|
|
|
|
|
@when("I convert the checkpoint model to domain via to_domain")
|
|
def step_checkpoint_to_domain(context: Any) -> None:
|
|
context.domain_checkpoint = context.checkpoint_model.to_domain()
|
|
|
|
|
|
@then("the domain checkpoint has the correct metadata fields")
|
|
def step_check_checkpoint_metadata(context: Any) -> None:
|
|
cp = context.domain_checkpoint
|
|
assert cp.checkpoint_id == _make_ulid(20)
|
|
assert cp.plan_id == _make_ulid(21)
|
|
assert cp.sandbox_ref == "abc123commit"
|
|
assert cp.checkpoint_type == "pre_write"
|
|
assert cp.filesystem_path == "checkpoints/cp1"
|
|
assert cp.size_bytes == 1024
|
|
assert cp.metadata.reason == "before write"
|
|
assert cp.metadata.source_tool == "file_write"
|
|
assert cp.metadata.phase == "execute"
|
|
|
|
|
|
@given("a CheckpointModel with invalid JSON in metadata_json")
|
|
def step_checkpoint_bad_json(context: Any) -> None:
|
|
from cleveragents.infrastructure.database.models import CheckpointModel
|
|
|
|
context.checkpoint_model = CheckpointModel(
|
|
checkpoint_id=_make_ulid(30),
|
|
plan_id=_make_ulid(31),
|
|
sandbox_ref="deadbeef",
|
|
checkpoint_type="manual",
|
|
filesystem_path="",
|
|
size_bytes=None,
|
|
created_at=NOW_ISO,
|
|
metadata_json="NOT VALID JSON {{{",
|
|
)
|
|
|
|
|
|
@then("the domain checkpoint metadata is empty defaults")
|
|
def step_check_empty_metadata(context: Any) -> None:
|
|
cp = context.domain_checkpoint
|
|
assert cp.metadata.reason == ""
|
|
assert cp.metadata.source_tool == ""
|
|
assert cp.metadata.phase == ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CheckpointModel — from_domain (lines 2865-2879)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a Checkpoint domain object with full metadata")
|
|
def step_checkpoint_domain(context: Any) -> None:
|
|
from cleveragents.domain.models.core.checkpoint import (
|
|
Checkpoint,
|
|
CheckpointMetadata,
|
|
)
|
|
|
|
context.checkpoint_domain = Checkpoint(
|
|
checkpoint_id=_make_ulid(40),
|
|
plan_id=_make_ulid(41),
|
|
sandbox_ref="commitsha",
|
|
decision_id=_make_ulid(42),
|
|
checkpoint_type="post_step",
|
|
resource_id=_make_ulid(43),
|
|
filesystem_path="cp/path",
|
|
size_bytes=2048,
|
|
created_at=NOW,
|
|
metadata=CheckpointMetadata(
|
|
reason="post step save",
|
|
source_tool="apply_tool",
|
|
phase="apply",
|
|
),
|
|
)
|
|
|
|
|
|
@when("I convert the checkpoint to a model via from_domain")
|
|
def step_checkpoint_from_domain(context: Any) -> None:
|
|
from cleveragents.infrastructure.database.models import CheckpointModel
|
|
|
|
context.checkpoint_model_out = CheckpointModel.from_domain(
|
|
context.checkpoint_domain,
|
|
)
|
|
|
|
|
|
@then("the model has correct checkpoint_id plan_id and metadata_json")
|
|
def step_check_checkpoint_model_fields(context: Any) -> None:
|
|
m = context.checkpoint_model_out
|
|
assert m.checkpoint_id == _make_ulid(40)
|
|
assert m.plan_id == _make_ulid(41)
|
|
assert m.sandbox_ref == "commitsha"
|
|
assert m.decision_id == _make_ulid(42)
|
|
assert m.checkpoint_type == "post_step"
|
|
assert m.resource_id == _make_ulid(43)
|
|
assert m.filesystem_path == "cp/path"
|
|
assert m.size_bytes == 2048
|
|
assert m.created_at == NOW_ISO
|
|
raw = m.metadata_json
|
|
assert raw is not None
|
|
parsed = json.loads(raw)
|
|
assert parsed["reason"] == "post step save"
|
|
assert parsed["source_tool"] == "apply_tool"
|
|
assert parsed["phase"] == "apply"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_session (lines 2907-2908)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("an in-memory SQLAlchemy engine")
|
|
def step_create_engine(context: Any) -> None:
|
|
from cleveragents.infrastructure.database.models import init_database
|
|
|
|
context.engine = init_database("sqlite:///:memory:")
|
|
|
|
|
|
@when("I call get_session with that engine")
|
|
def step_call_get_session(context: Any) -> None:
|
|
from cleveragents.infrastructure.database.models import get_session
|
|
|
|
context.session = get_session(context.engine)
|
|
|
|
|
|
@then("I receive a valid SQLAlchemy session object")
|
|
def step_check_session(context: Any) -> None:
|
|
from sqlalchemy.orm import Session
|
|
|
|
assert isinstance(context.session, Session)
|
|
# Verify it can execute a simple query
|
|
result = context.session.execute(__import__("sqlalchemy").text("SELECT 1"))
|
|
assert result.scalar() == 1
|
|
context.session.close()
|