forked from HAL9000/cleveragents-core
freemo
55aee7cf22
The step_register_skills_table step called add_skill in a loop but only committed once at the end. Because SkillRepository.create() obtains a new session per call and only flushes (never commits), the intermediate sessions could be garbage-collected before the final commit, rolling back their transactions on the shared SQLite :memory: connection. Moving _commit_pending inside the loop ensures each skill is durably committed before the next session is created. ISSUES CLOSED: #418
409 lines
14 KiB
Python
409 lines
14 KiB
Python
"""Step definitions for database models missing coverage tests.
|
|
|
|
Targets uncovered lines:
|
|
- Line 1666: ToolModel.from_domain() dict branch of _get helper
|
|
- Line 1921: SessionModel.from_domain() iterating over session.messages
|
|
- Lines 2131-2132: SkillModel column definitions (description, version)
|
|
"""
|
|
|
|
import json
|
|
from datetime import UTC, datetime
|
|
|
|
from behave import given, then, when
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from ulid import ULID
|
|
|
|
from cleveragents.domain.models.core.session import (
|
|
MessageRole,
|
|
Session,
|
|
SessionMessage,
|
|
SessionTokenUsage,
|
|
)
|
|
from cleveragents.infrastructure.database.models import (
|
|
Base,
|
|
SessionModel,
|
|
SkillModel,
|
|
ToolModel,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Reusable ULIDs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
ULID_SESSION = str(ULID())
|
|
ULID_MSG_1 = str(ULID())
|
|
ULID_MSG_2 = str(ULID())
|
|
|
|
NOW = datetime.now(tz=UTC)
|
|
NOW_ISO = NOW.isoformat()
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the missing coverage database is ready")
|
|
def step_missing_coverage_db_ready(context):
|
|
"""Set up an in-memory database with all tables."""
|
|
context.mc_engine = create_engine("sqlite:///:memory:")
|
|
context.McSessionLocal = sessionmaker(
|
|
bind=context.mc_engine, autoflush=False, autocommit=False
|
|
)
|
|
Base.metadata.create_all(context.mc_engine)
|
|
|
|
|
|
@given("a missing coverage database session is open")
|
|
def step_missing_coverage_session_open(context):
|
|
"""Open a database session for missing coverage tests."""
|
|
context.mc_session = context.McSessionLocal()
|
|
|
|
|
|
# ===================================================================
|
|
# ToolModel.from_domain() with dict input — line 1666
|
|
# ===================================================================
|
|
|
|
|
|
@given("a tool definition provided as a plain dict")
|
|
def step_tool_as_plain_dict(context):
|
|
"""Create a plain dict with tool fields."""
|
|
context.tool_dict = {
|
|
"name": "myns/my-tool",
|
|
"description": "A test tool from a dict",
|
|
"tool_type": "function",
|
|
"source": "custom",
|
|
"input_schema_json": json.dumps({"type": "object"}),
|
|
"output_schema_json": json.dumps({"type": "string"}),
|
|
"capability_json": None,
|
|
"lifecycle_json": None,
|
|
"code": "print('hello')",
|
|
"mcp_server": None,
|
|
"mcp_tool_name": None,
|
|
"agent_skill_path": None,
|
|
"timeout": 60,
|
|
"wraps": None,
|
|
"transform": None,
|
|
"mode": None,
|
|
"argument_mapping_json": None,
|
|
"resource_bindings": [],
|
|
}
|
|
|
|
|
|
@when("the tool dict is converted to a ToolModel via from_domain")
|
|
def step_convert_tool_dict(context):
|
|
"""Call ToolModel.from_domain with a plain dict."""
|
|
context.tool_model = ToolModel.from_domain(context.tool_dict)
|
|
|
|
|
|
@then("the ToolModel should have the correct name from the dict")
|
|
def step_tool_model_name(context):
|
|
assert context.tool_model.name == "myns/my-tool"
|
|
|
|
|
|
@then("the ToolModel should have the correct description from the dict")
|
|
def step_tool_model_description(context):
|
|
assert context.tool_model.description == "A test tool from a dict"
|
|
|
|
|
|
@then("the ToolModel should have the correct tool_type from the dict")
|
|
def step_tool_model_tool_type(context):
|
|
assert context.tool_model.tool_type == "function"
|
|
|
|
|
|
@then("the ToolModel should have the correct source from the dict")
|
|
def step_tool_model_source(context):
|
|
assert context.tool_model.source == "custom"
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# ToolModel.from_domain() with resource bindings as dicts
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@given("a tool definition provided as a plain dict with resource bindings")
|
|
def step_tool_dict_with_bindings(context):
|
|
"""Create a dict with resource_bindings also as dicts."""
|
|
context.tool_dict_bindings = {
|
|
"name": "myns/binding-tool",
|
|
"description": "Tool with bindings",
|
|
"tool_type": "function",
|
|
"source": "custom",
|
|
"resource_bindings": [
|
|
{
|
|
"slot_name": "db_conn",
|
|
"resource_type": "database/postgres",
|
|
"access_mode": "read_write",
|
|
"binding_mode": "static",
|
|
"static_resource": "main-db",
|
|
"required": True,
|
|
"description": "Primary database connection",
|
|
},
|
|
{
|
|
"slot_name": "cache",
|
|
"resource_type": "cache/redis",
|
|
"access_mode": "read_only",
|
|
"binding_mode": "contextual",
|
|
"static_resource": None,
|
|
"required": False,
|
|
"description": "Optional cache layer",
|
|
},
|
|
],
|
|
}
|
|
|
|
|
|
@when("the tool dict with bindings is converted to a ToolModel via from_domain")
|
|
def step_convert_tool_dict_bindings(context):
|
|
"""Call ToolModel.from_domain with dict resource bindings."""
|
|
context.tool_model_bindings = ToolModel.from_domain(context.tool_dict_bindings)
|
|
|
|
|
|
@then("the ToolModel should have resource binding child records")
|
|
def step_tool_model_has_bindings(context):
|
|
assert len(context.tool_model_bindings.resource_bindings_rel) == 2
|
|
|
|
|
|
@then("each resource binding should have the correct slot_name")
|
|
def step_binding_slot_names(context):
|
|
bindings = context.tool_model_bindings.resource_bindings_rel
|
|
assert bindings[0].slot_name == "db_conn"
|
|
assert bindings[1].slot_name == "cache"
|
|
|
|
|
|
@then("each resource binding should have the correct resource_type")
|
|
def step_binding_resource_types(context):
|
|
bindings = context.tool_model_bindings.resource_bindings_rel
|
|
assert bindings[0].resource_type == "database/postgres"
|
|
assert bindings[1].resource_type == "cache/redis"
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# ToolModel.from_domain() with namespaced name from dict
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@given("a tool definition dict with a namespaced name")
|
|
def step_tool_dict_namespaced(context):
|
|
"""Create a dict whose name includes a namespace."""
|
|
context.tool_dict_ns = {
|
|
"name": "acme/super-tool",
|
|
"description": "Namespaced tool",
|
|
"tool_type": "tool",
|
|
"source": "builtin",
|
|
}
|
|
|
|
|
|
@when("the namespaced tool dict is converted to a ToolModel via from_domain")
|
|
def step_convert_ns_tool_dict(context):
|
|
"""Call ToolModel.from_domain with a namespaced dict."""
|
|
context.tool_model_ns = ToolModel.from_domain(context.tool_dict_ns)
|
|
|
|
|
|
@then("the ToolModel namespace should be extracted from the dict name")
|
|
def step_tool_model_ns_namespace(context):
|
|
assert context.tool_model_ns.namespace == "acme"
|
|
|
|
|
|
@then("the ToolModel short_name should be extracted from the dict name")
|
|
def step_tool_model_ns_short_name(context):
|
|
assert context.tool_model_ns.short_name == "super-tool"
|
|
|
|
|
|
# ===================================================================
|
|
# SessionModel.from_domain() with messages — line 1921
|
|
# ===================================================================
|
|
|
|
|
|
def _make_session_message(message_id, role, content, sequence, tool_call_id=None):
|
|
"""Create a SessionMessage domain object."""
|
|
return SessionMessage(
|
|
message_id=message_id,
|
|
role=role,
|
|
content=content,
|
|
sequence=sequence,
|
|
timestamp=NOW,
|
|
metadata={"source": "test"},
|
|
tool_call_id=tool_call_id,
|
|
)
|
|
|
|
|
|
@given("a Session domain object with two messages")
|
|
def step_session_with_two_messages(context):
|
|
"""Create a Session domain object containing two messages."""
|
|
msg1 = _make_session_message(ULID_MSG_1, MessageRole.USER, "Hello agent", 0)
|
|
msg2 = _make_session_message(ULID_MSG_2, MessageRole.ASSISTANT, "Hello human", 1)
|
|
|
|
context.session_domain = Session(
|
|
session_id=ULID_SESSION,
|
|
actor_name="local/test-actor",
|
|
namespace="local",
|
|
linked_plan_ids=[],
|
|
token_usage=SessionTokenUsage(
|
|
input_tokens=100,
|
|
output_tokens=50,
|
|
estimated_cost=0.002,
|
|
),
|
|
metadata={"env": "test"},
|
|
messages=[msg1, msg2],
|
|
created_at=NOW,
|
|
updated_at=NOW,
|
|
)
|
|
|
|
|
|
@when("the Session domain object is converted to a SessionModel via from_domain")
|
|
def step_convert_session_domain(context):
|
|
"""Call SessionModel.from_domain with a Session that has messages."""
|
|
context.session_model = SessionModel.from_domain(context.session_domain)
|
|
|
|
|
|
@then("the SessionModel should have two message child records")
|
|
def step_session_model_two_messages(context):
|
|
assert len(context.session_model.messages_rel) == 2
|
|
|
|
|
|
@then('the first message record should have role "{role}"')
|
|
def step_first_msg_role(context, role):
|
|
assert context.session_model.messages_rel[0].role == role
|
|
|
|
|
|
@then('the second message record should have role "{role}"')
|
|
def step_second_msg_role(context, role):
|
|
assert context.session_model.messages_rel[1].role == role
|
|
|
|
|
|
@then("each message record should have the correct content")
|
|
def step_msg_contents(context):
|
|
assert context.session_model.messages_rel[0].content == "Hello agent"
|
|
assert context.session_model.messages_rel[1].content == "Hello human"
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# SessionModel.from_domain() with a single message
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@given("a Session domain object with one message")
|
|
def step_session_with_one_message(context):
|
|
"""Create a Session domain object containing one message."""
|
|
single_msg_id = str(ULID())
|
|
msg = _make_session_message(single_msg_id, MessageRole.USER, "Single message", 0)
|
|
|
|
context.single_msg_id = single_msg_id
|
|
context.session_domain_single = Session(
|
|
session_id=str(ULID()),
|
|
actor_name="local/solo-actor",
|
|
namespace="local",
|
|
linked_plan_ids=[],
|
|
token_usage=SessionTokenUsage(
|
|
input_tokens=10,
|
|
output_tokens=5,
|
|
estimated_cost=0.0001,
|
|
),
|
|
metadata={},
|
|
messages=[msg],
|
|
created_at=NOW,
|
|
updated_at=NOW,
|
|
)
|
|
|
|
|
|
@when("the single-message Session is converted to a SessionModel via from_domain")
|
|
def step_convert_single_msg_session(context):
|
|
"""Call SessionModel.from_domain with a single-message Session."""
|
|
context.session_model_single = SessionModel.from_domain(
|
|
context.session_domain_single
|
|
)
|
|
|
|
|
|
@then("the SessionModel should have one message child record")
|
|
def step_session_model_one_message(context):
|
|
assert len(context.session_model_single.messages_rel) == 1
|
|
|
|
|
|
@then("the single message record should have the correct message_id")
|
|
def step_single_msg_id(context):
|
|
assert (
|
|
context.session_model_single.messages_rel[0].message_id == context.single_msg_id
|
|
)
|
|
|
|
|
|
# ===================================================================
|
|
# SkillModel.from_domain() — lines 2131-2132 (description, version)
|
|
# ===================================================================
|
|
|
|
|
|
@given("a skill definition dict with description and version")
|
|
def step_skill_dict_with_version(context):
|
|
"""Create a dict representing a skill with description and version."""
|
|
context.skill_dict = {
|
|
"name": "local/code-tools",
|
|
"description": "A collection of code analysis tools",
|
|
"version": "1.2.3",
|
|
"tool_refs": ["local/lint", "local/format"],
|
|
"includes": [],
|
|
"anonymous_tools": [],
|
|
"mcp_servers": [],
|
|
"agent_skills": [],
|
|
"overrides": {},
|
|
}
|
|
|
|
|
|
@when("the skill dict is converted to a SkillModel via from_domain")
|
|
def step_convert_skill_dict(context):
|
|
"""Call SkillModel.from_domain with a dict."""
|
|
context.skill_model = SkillModel.from_domain(context.skill_dict)
|
|
|
|
|
|
@then("the SkillModel should have the correct description")
|
|
def step_skill_model_description(context):
|
|
assert context.skill_model.description == "A collection of code analysis tools"
|
|
|
|
|
|
@then("the SkillModel should have the correct version")
|
|
def step_skill_model_version(context):
|
|
assert context.skill_model.version == "1.2.3"
|
|
|
|
|
|
@then("the SkillModel should have the correct namespace")
|
|
def step_skill_model_namespace(context):
|
|
assert context.skill_model.namespace == "local"
|
|
|
|
|
|
@then("the SkillModel should have the correct short_name")
|
|
def step_skill_model_short_name(context):
|
|
assert context.skill_model.short_name == "code-tools"
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# SkillModel.from_domain() without version
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@given("a skill definition dict without a version field")
|
|
def step_skill_dict_no_version(context):
|
|
"""Create a skill dict without the version key."""
|
|
context.skill_dict_noversion = {
|
|
"name": "local/basic-skill",
|
|
"description": "A basic skill without version",
|
|
"tool_refs": [],
|
|
"includes": [],
|
|
"anonymous_tools": [],
|
|
"mcp_servers": [],
|
|
"agent_skills": [],
|
|
"overrides": {},
|
|
}
|
|
|
|
|
|
@when("the versionless skill dict is converted to a SkillModel via from_domain")
|
|
def step_convert_versionless_skill_dict(context):
|
|
"""Call SkillModel.from_domain with a dict missing version."""
|
|
context.skill_model_noversion = SkillModel.from_domain(context.skill_dict_noversion)
|
|
|
|
|
|
@then("the SkillModel version should be None")
|
|
def step_skill_model_version_none(context):
|
|
assert context.skill_model_noversion.version is None
|
|
|
|
|
|
@then("the SkillModel description should still be set")
|
|
def step_skill_model_description_still_set(context):
|
|
assert context.skill_model_noversion.description == "A basic skill without version"
|