"""Step definitions for resource_registry_service_coverage.feature. Covers: - bootstrap_builtin_types / register_type / register_resource exception rollback - _spec_to_db serialisation of auto_discovery and equivalence - _db_to_spec parsing of auto_discover_json and equivalence_json - Full round-trip through a real in-memory database """ from __future__ import annotations import json import tempfile from typing import Any from unittest.mock import MagicMock from behave import given, then, when # type: ignore[import-untyped] from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker from cleveragents.application.services.resource_registry_service import ( ResourceRegistryService, _db_to_spec, _spec_to_db, ) from cleveragents.domain.models.core.resource_type import ( ResourceKind, ResourceTypeSpec, SandboxStrategy, ) from cleveragents.infrastructure.database.models import ( Base, ResourceTypeModel, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _in_memory_session_factory() -> tuple[Any, Any]: """Create an in-memory SQLite engine + unclosable session factory.""" engine = create_engine( "sqlite:///:memory:", echo=False, connect_args={"check_same_thread": False}, ) Base.metadata.create_all(engine) real_session = sessionmaker( bind=engine, expire_on_commit=False, autoflush=True, autocommit=False, )() class _Unclosable: """Wraps a real session but makes close() a no-op.""" def __init__(self, s: Session) -> None: object.__setattr__(self, "_s", s) def close(self) -> None: pass def __getattr__(self, name: str) -> Any: return getattr(object.__getattribute__(self, "_s"), name) def __setattr__(self, name: str, value: Any) -> None: setattr(object.__getattribute__(self, "_s"), name, value) wrapper = _Unclosable(real_session) return engine, lambda: wrapper # --------------------------------------------------------------------------- # Scenario: bootstrap_builtin_types rolls back on unexpected exception # --------------------------------------------------------------------------- @given("a resource registry service with a faulty session for bootstrap") def step_faulty_session_bootstrap(context: Any) -> None: mock_session = MagicMock() # Make query().filter_by().first() succeed the first time, then blow up # on session.add() to trigger the generic Exception branch. mock_session.query.return_value.filter_by.return_value.first.return_value = None mock_session.add.side_effect = RuntimeError("unexpected DB error") context.faulty_session = mock_session context.cov_svc = ResourceRegistryService( session_factory=lambda: mock_session, ) @when("I call bootstrap_builtin_types and it fails") def step_call_bootstrap_fails(context: Any) -> None: context.bootstrap_exception = None try: context.cov_svc.bootstrap_builtin_types() except RuntimeError as exc: context.bootstrap_exception = exc @then("the faulty session should have been rolled back") def step_faulty_session_rolled_back(context: Any) -> None: context.faulty_session.rollback.assert_called() @then("the original exception should propagate from bootstrap") def step_bootstrap_exception_propagates(context: Any) -> None: assert context.bootstrap_exception is not None assert "unexpected DB error" in str(context.bootstrap_exception) # --------------------------------------------------------------------------- # Scenario: register_type rolls back on non-ValidationError exception # --------------------------------------------------------------------------- @given("a resource registry service with a faulty session for register_type") def step_faulty_session_register_type(context: Any) -> None: mock_session = MagicMock() # Make the duplicate-check query return None (no existing type) mock_session.query.return_value.filter_by.return_value.first.return_value = None # Blow up on session.add() with a non-ValidationError mock_session.add.side_effect = RuntimeError("disk full") context.regtype_faulty_session = mock_session context.cov_regtype_svc = ResourceRegistryService( session_factory=lambda: mock_session, ) @given("a temporary valid resource type YAML file exists") def step_create_temp_yaml(context: Any) -> None: yaml_content = ( "name: testns/cov-type\n" "description: Coverage test type\n" "resource_kind: physical\n" "sandbox_strategy: none\n" "user_addable: true\n" ) with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as tmp: tmp.write(yaml_content) tmp.flush() context.cov_yaml_path = tmp.name @when("I call register_type and a non-validation exception occurs") def step_call_register_type_fails(context: Any) -> None: context.regtype_exception = None try: context.cov_regtype_svc.register_type(context.cov_yaml_path) except RuntimeError as exc: context.regtype_exception = exc @then("the register_type faulty session should have been rolled back") def step_regtype_session_rolled_back(context: Any) -> None: context.regtype_faulty_session.rollback.assert_called() @then("the original non-validation exception should propagate from register_type") def step_regtype_exception_propagates(context: Any) -> None: assert context.regtype_exception is not None assert "disk full" in str(context.regtype_exception) # --------------------------------------------------------------------------- # Scenario: register_resource rolls back on generic exception # --------------------------------------------------------------------------- @given("a resource registry service with a faulty session for register_resource") def step_faulty_session_register_resource(context: Any) -> None: mock_session = MagicMock() # type_row lookup succeeds (returns a mock row with resource_kind) type_row = MagicMock() type_row.resource_kind = "physical" mock_session.query.return_value.filter_by.return_value.first.return_value = type_row # Blow up on session.add() with a generic Exception mock_session.add.side_effect = OSError("filesystem gone") context.regrsc_faulty_session = mock_session context.cov_regrsc_svc = ResourceRegistryService( session_factory=lambda: mock_session, ) @when("I call register_resource and a generic exception occurs") def step_call_register_resource_fails(context: Any) -> None: context.regrsc_exception = None try: context.cov_regrsc_svc.register_resource( type_name="git-checkout", name="test/res", location="/tmp/test", ) except OSError as exc: context.regrsc_exception = exc @then("the register_resource faulty session should have been rolled back") def step_regrsc_session_rolled_back(context: Any) -> None: context.regrsc_faulty_session.rollback.assert_called() @then("the original generic exception should propagate from register_resource") def step_regrsc_exception_propagates(context: Any) -> None: assert context.regrsc_exception is not None assert "filesystem gone" in str(context.regrsc_exception) # --------------------------------------------------------------------------- # Scenario: _spec_to_db serialises auto_discovery to JSON # --------------------------------------------------------------------------- @given("a ResourceTypeSpec with auto_discovery set") def step_spec_with_auto_discovery(context: Any) -> None: context.auto_discovery_data = { "glob_pattern": "*.py", "recursive": True, } context.cov_spec_ad = ResourceTypeSpec( name="git-checkout", description="Test type with auto-discovery", resource_kind=ResourceKind.PHYSICAL, sandbox_strategy=SandboxStrategy.GIT_WORKTREE, user_addable=True, auto_discovery=context.auto_discovery_data, built_in=True, ) @when("I convert the spec to a database model via _spec_to_db") def step_convert_spec_to_db_ad(context: Any) -> None: context.cov_db_model_ad = _spec_to_db(context.cov_spec_ad, source="test") @then("the database model auto_discover_json should contain the auto_discovery data") def step_verify_ad_json(context: Any) -> None: raw = context.cov_db_model_ad.auto_discover_json assert raw is not None, "auto_discover_json should not be None" parsed = json.loads(raw) assert parsed == context.auto_discovery_data # --------------------------------------------------------------------------- # Scenario: _spec_to_db serialises equivalence to JSON # --------------------------------------------------------------------------- @given("a ResourceTypeSpec with equivalence set") def step_spec_with_equivalence(context: Any) -> None: context.equivalence_data = { "criteria": ["content_hash", "location"], "description": "Hash-based equivalence comparison", } # Virtual types require equivalence and all-false capabilities context.cov_spec_eq = ResourceTypeSpec( name="testns/virt-type", description="Virtual type with equivalence", resource_kind=ResourceKind.VIRTUAL, sandbox_strategy=SandboxStrategy.NONE, user_addable=False, handler=None, capabilities={ "read": False, "write": False, "sandbox": False, "checkpoint": False, }, equivalence=context.equivalence_data, built_in=False, ) @when("I convert the equivalence spec to a database model via _spec_to_db") def step_convert_spec_to_db_eq(context: Any) -> None: context.cov_db_model_eq = _spec_to_db(context.cov_spec_eq, source="test") @then("the database model equivalence_json should contain the equivalence data") def step_verify_eq_json(context: Any) -> None: raw = context.cov_db_model_eq.equivalence_json assert raw is not None, "equivalence_json should not be None" parsed = json.loads(raw) assert parsed == context.equivalence_data # --------------------------------------------------------------------------- # Scenario: _db_to_spec parses auto_discover_json from a database row # --------------------------------------------------------------------------- @given("an in-memory database with a resource type row containing auto_discover_json") def step_db_row_with_auto_discover(context: Any) -> None: engine, factory = _in_memory_session_factory() context.cov_ad_engine = engine context.cov_ad_factory = factory context.ad_original = {"glob": "**/*.py", "depth": 3} session = factory() from datetime import UTC, datetime now = datetime.now(tz=UTC).isoformat() row = ResourceTypeModel( name="git-checkout", namespace="builtin", description="Test row with auto_discover_json", resource_kind="physical", sandbox_strategy="git_worktree", user_addable=True, handler_ref=None, args_schema_json=None, allowed_parent_types_json=None, allowed_child_types_json=None, auto_discover_json=json.dumps(context.ad_original), capabilities_json=json.dumps( {"read": True, "write": True, "sandbox": True, "checkpoint": False} ), equivalence_json=None, source="test", created_at=now, updated_at=now, ) session.add(row) session.commit() context.cov_ad_row = ( session.query(ResourceTypeModel).filter_by(name="git-checkout").first() ) @when("I convert the database row back to a spec via _db_to_spec") def step_convert_db_to_spec_ad(context: Any) -> None: context.cov_spec_from_db_ad = _db_to_spec(context.cov_ad_row) @then("the resulting spec auto_discovery should match the original data") def step_verify_spec_ad(context: Any) -> None: assert context.cov_spec_from_db_ad.auto_discovery is not None assert context.cov_spec_from_db_ad.auto_discovery == context.ad_original # --------------------------------------------------------------------------- # Scenario: _db_to_spec parses equivalence_json from a database row # --------------------------------------------------------------------------- @given("an in-memory database with a resource type row containing equivalence_json") def step_db_row_with_equivalence(context: Any) -> None: engine, factory = _in_memory_session_factory() context.cov_eq_engine = engine context.cov_eq_factory = factory context.eq_original = {"criteria": ["hash"], "description": "Exact match"} session = factory() from datetime import UTC, datetime now = datetime.now(tz=UTC).isoformat() row = ResourceTypeModel( name="testns/virt-eq", namespace="testns", description="Virtual type with equivalence in DB", resource_kind="virtual", sandbox_strategy="none", user_addable=False, handler_ref=None, args_schema_json=None, allowed_parent_types_json=None, allowed_child_types_json=None, auto_discover_json=None, capabilities_json=json.dumps( {"read": False, "write": False, "sandbox": False, "checkpoint": False} ), equivalence_json=json.dumps(context.eq_original), source="test", created_at=now, updated_at=now, ) session.add(row) session.commit() context.cov_eq_row = ( session.query(ResourceTypeModel).filter_by(name="testns/virt-eq").first() ) @when("I convert the equivalence database row back to a spec via _db_to_spec") def step_convert_db_to_spec_eq(context: Any) -> None: context.cov_spec_from_db_eq = _db_to_spec(context.cov_eq_row) @then("the resulting spec equivalence should match the original equivalence data") def step_verify_spec_eq(context: Any) -> None: assert context.cov_spec_from_db_eq.equivalence is not None assert context.cov_spec_from_db_eq.equivalence == context.eq_original # --------------------------------------------------------------------------- # Scenario: auto_discovery and equivalence survive a full round-trip # --------------------------------------------------------------------------- @given("a real in-memory resource registry service is initialised") def step_real_inmem_service(context: Any) -> None: engine, factory = _in_memory_session_factory() context.cov_rt_engine = engine context.cov_rt_factory = factory context.cov_rt_svc = ResourceRegistryService(session_factory=factory) @given("a YAML config with auto_discovery and equivalence fields exists") def step_yaml_with_ad_and_eq(context: Any) -> None: # Virtual type requires equivalence yaml_content = ( "name: covns/round-trip\n" "description: Round-trip test type\n" "resource_kind: virtual\n" "sandbox_strategy: none\n" "user_addable: false\n" "auto_discovery:\n" " glob_pattern: '*.md'\n" " max_depth: 5\n" "equivalence:\n" " criteria:\n" " - content_hash\n" " description: Hash-based equivalence\n" ) with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as tmp: tmp.write(yaml_content) tmp.flush() context.cov_rt_yaml_path = tmp.name @when("I register the type with auto_discovery and equivalence via the service") def step_register_rt_type(context: Any) -> None: context.cov_rt_spec = context.cov_rt_svc.register_type(context.cov_rt_yaml_path) @when("I show the registered type with auto_discovery and equivalence") def step_show_rt_type(context: Any) -> None: context.cov_rt_shown = context.cov_rt_svc.show_type("covns/round-trip") @then("the shown type should have the correct auto_discovery") def step_verify_rt_ad(context: Any) -> None: ad = context.cov_rt_shown.auto_discovery assert ad is not None, "auto_discovery should be preserved" assert ad["glob_pattern"] == "*.md" assert ad["max_depth"] == 5 @then("the shown type should have the correct equivalence") def step_verify_rt_eq(context: Any) -> None: eq = context.cov_rt_shown.equivalence assert eq is not None, "equivalence should be preserved" assert eq["criteria"] == ["content_hash"] assert eq["description"] == "Hash-based equivalence"