"""Step definitions for Skill Registry persistence coverage. Covers skills, skill_items tables and the SkillRegistryService integration layer. """ from __future__ import annotations from collections.abc import Callable from typing import Any from behave import given, then, when from behave.runner import Context from sqlalchemy import create_engine, event from sqlalchemy.orm import Session, sessionmaker from cleveragents.application.services.skill_registry_service import ( SkillRegistryService, ) from cleveragents.domain.models.core.skill import ( Skill, SkillAgentSource, SkillInclude, SkillInlineTool, SkillMcpSource, ) from cleveragents.domain.models.core.tool import ToolSource from cleveragents.infrastructure.database.models import Base from cleveragents.infrastructure.database.repositories import ( DuplicateSkillError, SkillNotFoundError, SkillRepository, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _get_session_factory(context: Context) -> Callable[[], Session]: """Return the session factory stored on *context*. The factory always returns the same shared session for the current scenario, mirroring the ``UnitOfWork`` behaviour used in production. """ return context.skill_session_factory # type: ignore[no-any-return] def _commit_pending(context: Context) -> None: """Commit the shared scenario session. Because the repository only *flushes* (never commits), the test must explicitly commit to make data durable. The session is kept open so that subsequent steps can continue using it. """ session = _get_session_factory(context)() session.commit() def _make_skill( name: str = "local/test-skill", description: str = "Test skill", tool_refs: list[str] | None = None, includes: list[SkillInclude] | None = None, anonymous_tools: list[SkillInlineTool] | None = None, mcp_servers: list[SkillMcpSource] | None = None, agent_skills: list[SkillAgentSource] | None = None, overrides: dict[str, dict[str, Any]] | None = None, ) -> Skill: """Create a minimal valid Skill domain object for testing.""" return Skill( name=name, description=description, tool_refs=tool_refs or [], includes=includes or [], anonymous_tools=anonymous_tools or [], mcp_servers=mcp_servers or [], agent_skills=agent_skills or [], overrides=overrides or {}, ) def _enable_fk_pragma(dbapi_conn: Any, _connection_record: Any) -> None: """Enable SQLite foreign key enforcement per connection (SEC-3).""" cursor = dbapi_conn.cursor() cursor.execute("PRAGMA foreign_keys=ON") cursor.close() # --------------------------------------------------------------------------- # Background / setup steps # --------------------------------------------------------------------------- @given("a clean in-memory database with the skill registry schema") def step_clean_db(context: Context) -> None: engine = create_engine("sqlite:///:memory:", echo=False) # SEC-3: Enable FK enforcement so CASCADE works at the DB level event.listen(engine, "connect", _enable_fk_pragma) Base.metadata.create_all(engine) # Use a single shared session so that flush() data is visible across # all repository calls within the same scenario. Without this, # short-lived sessions returned by sessionmaker() may be garbage- # collected before _commit_pending() runs, causing the StaticPool's # reset_on_return="rollback" to discard uncommitted INSERTs. _shared_session: Session = sessionmaker(bind=engine)() def _session_factory() -> Session: return _shared_session context.skill_engine = engine context.skill_session_factory = _session_factory context.skill_error = None context.skill_result = None context.skill_removal_result = None @given("a skill registry repository backed by the database") def step_repo(context: Context) -> None: factory = _get_session_factory(context) context.skill_repo = SkillRepository(session_factory=factory) @given("a skill registry service backed by the repository") def step_service(context: Context) -> None: context.skill_service = SkillRegistryService( skill_repo=context.skill_repo, ) # --------------------------------------------------------------------------- # Given: Skill construction # --------------------------------------------------------------------------- @given('a valid skill named "{name}" with description "{description}"') def step_valid_skill(context: Context, name: str, description: str) -> None: context.current_skill = _make_skill(name=name, description=description) @given('the skill has tool refs "{refs_csv}"') def step_skill_tool_refs(context: Context, refs_csv: str) -> None: refs = [r.strip() for r in refs_csv.split(",")] skill: Skill = context.current_skill context.current_skill = skill.model_copy(update={"tool_refs": refs}) @given('the skill includes "{includes_csv}"') def step_skill_includes(context: Context, includes_csv: str) -> None: includes = [SkillInclude(name=n.strip()) for n in includes_csv.split(",")] skill: Skill = context.current_skill context.current_skill = skill.model_copy(update={"includes": includes}) @given('the skill has an inline tool with description "{desc}"') def step_skill_inline_tool(context: Context, desc: str) -> None: tool = SkillInlineTool(description=desc, source=ToolSource.CUSTOM, timeout=300) skill: Skill = context.current_skill context.current_skill = skill.model_copy(update={"anonymous_tools": [tool]}) @given('the skill has an MCP source "{server}"') def step_skill_mcp(context: Context, server: str) -> None: mcp = SkillMcpSource(server=server) skill: Skill = context.current_skill context.current_skill = skill.model_copy(update={"mcp_servers": [mcp]}) @given('the skill has an agent source "{path}"') def step_skill_agent(context: Context, path: str) -> None: agent = SkillAgentSource(path=path) skill: Skill = context.current_skill context.current_skill = skill.model_copy(update={"agent_skills": [agent]}) @given("the skill has already been registered once in the skill registry") def step_skill_already_registered(context: Context) -> None: svc: SkillRegistryService = context.skill_service svc.add_skill(context.current_skill) _commit_pending(context) @given("the following skills are registered:") def step_register_skills_table(context: Context) -> None: svc: SkillRegistryService = context.skill_service for row in context.table: skill = _make_skill( name=row["name"], description=row["description"], ) svc.add_skill(skill) _commit_pending(context) @given('a valid skill named "{name}" with overrides for "{tool_name}"') def step_skill_with_overrides(context: Context, name: str, tool_name: str) -> None: context.current_skill = _make_skill( name=name, description="Skill with overrides", tool_refs=[tool_name], overrides={tool_name: {"timeout": 60, "retries": 3}}, ) @given('a valid skill named "{name}" with version "{version}"') def step_skill_with_version(context: Context, name: str, version: str) -> None: # Skill domain model has no version field; store in context for DB check context.current_skill = _make_skill(name=name, description="Versioned skill") context.skill_expected_version = version @given('a valid skill named "{name}" with mixed item types') def step_skill_mixed_items(context: Context, name: str) -> None: context.current_skill = _make_skill( name=name, description="Mixed items skill", tool_refs=["local/tool-a", "local/tool-b"], includes=[SkillInclude(name="local/base-skill")], agent_skills=[SkillAgentSource(path="/skills/my-agent")], ) @given('a valid skill named "{name}" with {count:d} tool refs') def step_skill_many_refs(context: Context, name: str, count: int) -> None: refs = [f"local/tool-{i:04d}" for i in range(count)] context.current_skill = _make_skill( name=name, description=f"Skill with {count} tool refs", tool_refs=refs, ) # --------------------------------------------------------------------------- # When: Actions # --------------------------------------------------------------------------- @when("the skill is registered through the skill registry service") @given("the skill is registered through the skill registry service") def step_register_skill(context: Context) -> None: svc: SkillRegistryService = context.skill_service try: context.skill_result = svc.add_skill(context.current_skill) _commit_pending(context) context.skill_error = None except Exception as exc: context.skill_error = exc @when('a second skill with the same name "{name}" is registered') def step_register_duplicate(context: Context, name: str) -> None: svc: SkillRegistryService = context.skill_service dup_skill = _make_skill(name=name, description="Duplicate") try: svc.add_skill(dup_skill) _commit_pending(context) context.skill_error = None except DuplicateSkillError as exc: context.skill_error = exc except Exception as exc: # tenacity.RetryError may wrap the real exception; unwrap. cause = exc.__cause__ or exc context.skill_error = cause @when('I retrieve skill "{name}" from the skill registry') @then('I retrieve skill "{name}" from the skill registry') def step_retrieve_skill(context: Context, name: str) -> None: svc: SkillRegistryService = context.skill_service context.skill_result = svc.get_skill(name) @when("I list all skills from the skill registry") def step_list_all_skills(context: Context) -> None: svc: SkillRegistryService = context.skill_service context.skill_list_result = svc.list_skills() @when('I list skills from the skill registry with namespace "{ns}"') def step_list_skills_ns(context: Context, ns: str) -> None: svc: SkillRegistryService = context.skill_service context.skill_list_result = svc.list_skills(namespace=ns) @when('I update skill "{name}" with description "{desc}"') def step_update_skill(context: Context, name: str, desc: str) -> None: svc: SkillRegistryService = context.skill_service updated_skill = _make_skill(name=name, description=desc) try: svc.update_skill(updated_skill) _commit_pending(context) context.skill_result = svc.get_skill(name) context.skill_error = None except Exception as exc: context.skill_error = exc @when('I try to update a non-existent skill "{name}"') def step_update_nonexistent(context: Context, name: str) -> None: svc: SkillRegistryService = context.skill_service fake_skill = _make_skill(name=name, description="Ghost") try: svc.update_skill(fake_skill) _commit_pending(context) context.skill_error = None except SkillNotFoundError as exc: context.skill_error = exc except Exception as exc: # tenacity.RetryError may wrap the real exception; unwrap. cause = exc.__cause__ or exc context.skill_error = cause @when('I remove skill "{name}" from the skill registry') def step_remove_skill(context: Context, name: str) -> None: svc: SkillRegistryService = context.skill_service try: context.skill_removal_result = svc.remove_skill(name) _commit_pending(context) except Exception as exc: context.skill_error = exc @when('I register a skill with invalid name "{name}" via dict bypass') def step_register_invalid_name(context: Context, name: str) -> None: """Attempt to register a skill with an invalid name through from_domain.""" from cleveragents.infrastructure.database.models import SkillModel try: SkillModel.from_domain({"name": name, "description": "Invalid"}) context.skill_error = None except ValueError as exc: context.skill_error = exc @when("I register a skill with empty name via dict bypass") def step_register_empty_name(context: Context) -> None: """Attempt to register a skill with an empty name through from_domain.""" from cleveragents.infrastructure.database.models import SkillModel try: SkillModel.from_domain({"name": "", "description": "Invalid"}) context.skill_error = None except ValueError as exc: context.skill_error = exc @when("the skill is registered and retrieved with overrides check") def step_register_and_check_overrides(context: Context) -> None: svc: SkillRegistryService = context.skill_service try: svc.add_skill(context.current_skill) _commit_pending(context) context.skill_result = svc.get_skill(context.current_skill.name) context.skill_error = None except Exception as exc: context.skill_error = exc @when("I retrieve the skill and check item ordering") def step_check_item_ordering(context: Context) -> None: svc: SkillRegistryService = context.skill_service context.skill_result = svc.get_skill(context.current_skill.name) @when("the skill is registered with a transient DB error then succeeds") def step_register_with_retry(context: Context) -> None: """Simulate a transient DB error on first attempt, success on second.""" from sqlalchemy.exc import OperationalError as SqlaOpError svc: SkillRegistryService = context.skill_service repo: SkillRepository = context.skill_repo call_count = 0 original_session = repo._session def flaky_session() -> Session: nonlocal call_count call_count += 1 session = original_session() if call_count == 1: # Simulate a transient DB error on first call raise SqlaOpError( "database is locked", params=None, orig=Exception("locked") ) return session repo._session = flaky_session # type: ignore[assignment] try: context.skill_result = svc.add_skill(context.current_skill) _commit_pending(context) context.skill_error = None except Exception as exc: context.skill_error = exc finally: repo._session = original_session # type: ignore[assignment] context.retry_call_count = call_count # --------------------------------------------------------------------------- # Then: Assertions # --------------------------------------------------------------------------- @then("the skill registry service should not raise an error") def step_no_error(context: Context) -> None: assert context.skill_error is None, f"Unexpected error: {context.skill_error}" @then('the persisted skill should have the name "{name}"') def step_skill_has_name(context: Context, name: str) -> None: result = context.skill_result # If result is None, re-fetch from service if result is None: svc: SkillRegistryService = context.skill_service result = svc.get_skill(name) context.skill_result = result assert result is not None, f"Skill '{name}' not found" assert result.name == name, f"Expected name {name}, got {result.name}" @then('the persisted skill description should be "{desc}"') def step_skill_description(context: Context, desc: str) -> None: result = context.skill_result assert result is not None, "No skill result available" assert result.description == desc, ( f"Expected description '{desc}', got '{result.description}'" ) @then('the persisted skill should have {count:d} items of type "{item_type}"') def step_skill_items_count(context: Context, count: int, item_type: str) -> None: # Re-fetch to get the full reconstructed Skill skill = context.current_skill svc: SkillRegistryService = context.skill_service result = svc.get_skill(skill.name) assert result is not None, f"Skill '{skill.name}' not found" # Map item_type to the corresponding Skill field type_to_field: dict[str, int] = { "tool_ref": len(result.tool_refs), "include": len(result.includes), "inline_tool": len(result.anonymous_tools), "mcp_source": len(result.mcp_servers), "agent_source": len(result.agent_skills), } actual = type_to_field.get(item_type, 0) assert actual == count, ( f"Expected {count} items of type '{item_type}', got {actual}" ) @then('a DuplicateSkillError should be raised mentioning "{name}"') def step_duplicate_error(context: Context, name: str) -> None: assert context.skill_error is not None, "Expected DuplicateSkillError" assert isinstance(context.skill_error, DuplicateSkillError), ( f"Expected DuplicateSkillError, got {type(context.skill_error)}" ) assert name in str(context.skill_error), ( f"Expected '{name}' in error message: {context.skill_error}" ) @then("the skill registry result should be None") def step_result_none(context: Context) -> None: assert context.skill_result is None, f"Expected None, got {context.skill_result}" @then("the skill list should contain {count:d} skills") def step_list_count(context: Context, count: int) -> None: result_list: list[Skill] = context.skill_list_result assert len(result_list) == count, f"Expected {count} skills, got {len(result_list)}" @then("a SkillNotFoundError should be raised") def step_not_found_error(context: Context) -> None: assert context.skill_error is not None, "Expected SkillNotFoundError" assert isinstance(context.skill_error, SkillNotFoundError), ( f"Expected SkillNotFoundError, got {type(context.skill_error)}" ) @then("the skill removal should return True") def step_removal_true(context: Context) -> None: assert context.skill_removal_result is True, ( f"Expected True, got {context.skill_removal_result}" ) @then("the skill removal should return False") def step_removal_false(context: Context) -> None: assert context.skill_removal_result is False, ( f"Expected False, got {context.skill_removal_result}" ) @then("a ValueError should be raised about invalid skill name") def step_invalid_name_error(context: Context) -> None: assert context.skill_error is not None, "Expected ValueError" assert isinstance(context.skill_error, ValueError), ( f"Expected ValueError, got {type(context.skill_error)}" ) assert ( "namespace/short_name" in str(context.skill_error).lower() or "pattern" in str(context.skill_error).lower() ), f"Error should mention name format: {context.skill_error}" @then('the retrieved skill should have overrides for "{tool_name}"') def step_check_overrides(context: Context, tool_name: str) -> None: result: Skill = context.skill_result assert result is not None, "No skill result available" assert tool_name in result.overrides, ( f"Expected overrides for '{tool_name}', got keys: {list(result.overrides.keys())}" ) assert result.overrides[tool_name]["timeout"] == 60 assert result.overrides[tool_name]["retries"] == 3 @then("the retrieved skill should have items in stable order") def step_check_item_order(context: Context) -> None: result: Skill = context.skill_result assert result is not None, "No skill result available" # Original was: 2 tool_refs, 1 include, 1 agent_source assert len(result.tool_refs) == 2, ( f"Expected 2 tool_refs, got {len(result.tool_refs)}" ) assert result.tool_refs[0] == "local/tool-a" assert result.tool_refs[1] == "local/tool-b" assert len(result.includes) == 1, f"Expected 1 include, got {len(result.includes)}" assert result.includes[0].name == "local/base-skill" assert len(result.agent_skills) == 1, ( f"Expected 1 agent_skill, got {len(result.agent_skills)}" ) assert result.agent_skills[0].path == "/skills/my-agent" @then("the skill list entries should all be Skill domain objects") def step_list_domain_objects(context: Context) -> None: result_list: list[Skill] = context.skill_list_result for item in result_list: assert isinstance(item, Skill), f"Expected Skill instance, got {type(item)}" @then("the persisted skill should have {count:d} tool refs") def step_skill_tool_ref_count(context: Context, count: int) -> None: skill = context.current_skill svc: SkillRegistryService = context.skill_service result = svc.get_skill(skill.name) assert result is not None, f"Skill '{skill.name}' not found" assert len(result.tool_refs) == count, ( f"Expected {count} tool refs, got {len(result.tool_refs)}" ) @then("the retry should have been attempted") def step_retry_attempted(context: Context) -> None: # The @database_retry decorator retries on OperationalError, # so the call count should be > 1 (first attempt raised, second succeeded) assert context.retry_call_count >= 1, ( f"Expected retry, but call_count was {context.retry_call_count}" )