"""Step definitions for resource registry database table tests. Tests the ResourceTypeModel, ResourceModel, and ResourceEdgeModel ORM models and their corresponding database tables (resource_types, resources, resource_edges). Uses an in-memory SQLite database with schema created via ``Base.metadata.create_all``. """ from __future__ import annotations import json from datetime import UTC, datetime from typing import Any from behave import given, then, when # type: ignore[import-untyped] from sqlalchemy import create_engine, event, inspect, text from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session, sessionmaker from cleveragents.infrastructure.database.models import ( Base, ResourceEdgeModel, ResourceModel, ResourceTypeModel, ) def _now_iso() -> str: """Return current UTC time as ISO-8601 string.""" return datetime.now(tz=UTC).isoformat() def _make_ulid(suffix: str = "0") -> str: """Create a deterministic 26-char fake ULID for testing.""" return ("0" * (26 - len(suffix))) + suffix # --------------------------------------------------------------------------- # Background # --------------------------------------------------------------------------- def _set_sqlite_pragma(dbapi_conn: Any, _connection_record: Any) -> None: """Enable FK enforcement on every new SQLite connection.""" cursor = dbapi_conn.cursor() cursor.execute("PRAGMA foreign_keys = ON") cursor.close() @given("a resource registry in-memory database is initialized") def step_resource_registry_init_db(context: Any) -> None: """Create an in-memory SQLite database with all tables.""" engine = create_engine("sqlite:///:memory:") # Enable FK enforcement on every connection via event listener event.listen(engine, "connect", _set_sqlite_pragma) # Re-establish FK pragma on the initial connection used for metadata with engine.connect() as conn: conn.execute(text("PRAGMA foreign_keys = ON")) conn.commit() Base.metadata.create_all(engine) context.rr_engine = engine context.rr_session_factory = sessionmaker(bind=engine) context.rr_session = context.rr_session_factory() # Storage for cross-step references context.rr_resource_type = None context.rr_resource = None context.rr_parent_resource = None context.rr_parent_resource_2 = None context.rr_child_resource = None context.rr_child_resource_2 = None context.rr_edge = None context.rr_error = None # --------------------------------------------------------------------------- # Table existence # --------------------------------------------------------------------------- @then('the resource registry database should contain table "{table_name}"') def step_resource_registry_table_exists(context: Any, table_name: str) -> None: inspector = inspect(context.rr_engine) tables = inspector.get_table_names() assert table_name in tables, f"Table {table_name} not found. Tables: {tables}" # --------------------------------------------------------------------------- # ResourceTypeModel CRUD # --------------------------------------------------------------------------- @when('I create a resource type "{name}" with kind "{kind}" and user_addable {addable}') def step_resource_registry_create_type( context: Any, name: str, kind: str, addable: str ) -> None: session: Session = context.rr_session namespace = name.split("/")[0] rt = ResourceTypeModel() rt.name = name rt.namespace = namespace rt.resource_kind = kind rt.user_addable = addable.lower() == "true" rt.created_at = _now_iso() rt.updated_at = _now_iso() session.add(rt) session.commit() context.rr_resource_type = rt @then('I can retrieve resource type "{name}"') def step_resource_registry_get_type(context: Any, name: str) -> None: session: Session = context.rr_session rt = session.get(ResourceTypeModel, name) assert rt is not None, f"ResourceType {name} not found" context.rr_resource_type = rt @then('the retrieved resource type namespace is "{expected}"') def step_resource_registry_type_namespace(context: Any, expected: str) -> None: assert context.rr_resource_type.namespace == expected @then('the retrieved resource type kind is "{expected}"') def step_resource_registry_type_kind(context: Any, expected: str) -> None: assert context.rr_resource_type.resource_kind == expected @then("the retrieved resource type user_addable is {expected}") def step_resource_registry_type_user_addable(context: Any, expected: str) -> None: expected_bool = expected.lower() == "true" assert context.rr_resource_type.user_addable == expected_bool @when('I create a resource type "{name}" with JSON fields') def step_resource_registry_create_type_json(context: Any, name: str) -> None: session: Session = context.rr_session namespace = name.split("/")[0] rt = ResourceTypeModel() rt.name = name rt.namespace = namespace rt.resource_kind = "physical" rt.user_addable = True rt.args_schema_json = json.dumps( [{"flag": "--path", "help": "Path to repo", "required": True}] ) rt.allowed_parent_types_json = json.dumps([]) rt.allowed_child_types_json = json.dumps( [{"type": "builtin/fs-directory"}, {"type": "builtin/git"}] ) rt.auto_discover_json = json.dumps({"enabled": True, "max_depth": 10}) rt.capabilities_json = json.dumps( { "readable": True, "writable": True, "sandboxable": True, "checkpointable": True, } ) rt.created_at = _now_iso() rt.updated_at = _now_iso() session.add(rt) session.commit() context.rr_resource_type = rt @then("the resource type args_schema_json round-trips correctly") def step_resource_registry_type_args_json(context: Any) -> None: data = json.loads(context.rr_resource_type.args_schema_json) assert isinstance(data, list) assert data[0]["flag"] == "--path" @then("the resource type allowed_parent_types_json round-trips correctly") def step_resource_registry_type_parents_json(context: Any) -> None: data = json.loads(context.rr_resource_type.allowed_parent_types_json) assert isinstance(data, list) @then("the resource type allowed_child_types_json round-trips correctly") def step_resource_registry_type_children_json(context: Any) -> None: data = json.loads(context.rr_resource_type.allowed_child_types_json) assert isinstance(data, list) assert len(data) == 2 @then("the resource type auto_discover_json round-trips correctly") def step_resource_registry_type_autodiscover_json(context: Any) -> None: data = json.loads(context.rr_resource_type.auto_discover_json) assert data["enabled"] is True assert data["max_depth"] == 10 @then("the resource type capabilities_json round-trips correctly") def step_resource_registry_type_capabilities_json(context: Any) -> None: data = json.loads(context.rr_resource_type.capabilities_json) assert data["readable"] is True assert data["sandboxable"] is True @then('creating a duplicate resource type "{name}" raises an integrity error') def step_resource_registry_duplicate_type(context: Any, name: str) -> None: session: Session = context.rr_session namespace = name.split("/")[0] rt = ResourceTypeModel() rt.name = name rt.namespace = namespace rt.resource_kind = "physical" rt.user_addable = False rt.created_at = _now_iso() rt.updated_at = _now_iso() try: session.add(rt) session.commit() msg = "Expected IntegrityError for duplicate resource type" raise AssertionError(msg) except IntegrityError: session.rollback() @then('creating a resource type with kind "{kind}" raises an integrity error') def step_resource_registry_invalid_kind(context: Any, kind: str) -> None: session: Session = context.rr_session try: session.execute( text( "INSERT INTO resource_types (name, namespace, resource_kind, " "user_addable, created_at, updated_at) " "VALUES (:name, :ns, :kind, 0, :ts, :ts)" ), { "name": "test/invalid-kind", "ns": "test", "kind": kind, "ts": _now_iso(), }, ) session.commit() msg = "Expected IntegrityError for invalid resource_kind" raise AssertionError(msg) except IntegrityError: session.rollback() @then('listing resource types by namespace "{ns}" returns {count:d} results') def step_resource_registry_list_types_ns_plural( context: Any, ns: str, count: int ) -> None: session: Session = context.rr_session results = ( session.query(ResourceTypeModel).filter(ResourceTypeModel.namespace == ns).all() ) assert len(results) == count, f"Expected {count}, got {len(results)}" @then('listing resource types by namespace "{ns}" returns {count:d} result') def step_resource_registry_list_types_ns_singular( context: Any, ns: str, count: int ) -> None: session: Session = context.rr_session results = ( session.query(ResourceTypeModel).filter(ResourceTypeModel.namespace == ns).all() ) assert len(results) == count, f"Expected {count}, got {len(results)}" @then('listing resource types by kind "{kind}" returns {count:d} results') def step_resource_registry_list_types_kind_plural( context: Any, kind: str, count: int ) -> None: session: Session = context.rr_session results = ( session.query(ResourceTypeModel) .filter(ResourceTypeModel.resource_kind == kind) .all() ) assert len(results) == count, f"Expected {count}, got {len(results)}" @then('listing resource types by kind "{kind}" returns {count:d} result') def step_resource_registry_list_types_kind_singular( context: Any, kind: str, count: int ) -> None: session: Session = context.rr_session results = ( session.query(ResourceTypeModel) .filter(ResourceTypeModel.resource_kind == kind) .all() ) assert len(results) == count, f"Expected {count}, got {len(results)}" # --------------------------------------------------------------------------- # ResourceModel CRUD - prerequisites # --------------------------------------------------------------------------- @given('a resource type "{name}" exists') def step_resource_registry_ensure_type_physical(context: Any, name: str) -> None: session: Session = context.rr_session existing = session.get(ResourceTypeModel, name) if existing is None: namespace = name.split("/")[0] rt = ResourceTypeModel() rt.name = name rt.namespace = namespace rt.resource_kind = "physical" rt.user_addable = True rt.created_at = _now_iso() rt.updated_at = _now_iso() session.add(rt) session.commit() @given('a resource type "{name}" exists as virtual') def step_resource_registry_ensure_type_virtual(context: Any, name: str) -> None: session: Session = context.rr_session existing = session.get(ResourceTypeModel, name) if existing is None: namespace = name.split("/")[0] rt = ResourceTypeModel() rt.name = name rt.namespace = namespace rt.resource_kind = "virtual" rt.user_addable = False rt.created_at = _now_iso() rt.updated_at = _now_iso() session.add(rt) session.commit() @given('a resource type "{name}" exists as physical') def step_resource_registry_ensure_type_physical_explicit( context: Any, name: str ) -> None: session: Session = context.rr_session existing = session.get(ResourceTypeModel, name) if existing is None: namespace = name.split("/")[0] rt = ResourceTypeModel() rt.name = name rt.namespace = namespace rt.resource_kind = "physical" rt.user_addable = True rt.created_at = _now_iso() rt.updated_at = _now_iso() session.add(rt) session.commit() # --------------------------------------------------------------------------- # ResourceModel CRUD # --------------------------------------------------------------------------- @when('I create a resource "{rname}" of type "{tname}" with location "{loc}"') def step_resource_registry_create_resource( context: Any, rname: str, tname: str, loc: str ) -> None: session: Session = context.rr_session namespace = rname.split("/")[0] rid = _make_ulid(rname.replace("/", "_")) # Look up the type to get the kind rt = session.get(ResourceTypeModel, tname) kind = rt.resource_kind if rt else "physical" r = ResourceModel() r.resource_id = rid r.namespaced_name = rname r.namespace = namespace r.type_name = tname r.resource_kind = kind r.location = loc r.created_at = _now_iso() r.updated_at = _now_iso() session.add(r) session.commit() context.rr_resource = r @then("I can retrieve the resource by id") def step_resource_registry_get_resource(context: Any) -> None: session: Session = context.rr_session rid = context.rr_resource.resource_id r = session.get(ResourceModel, rid) assert r is not None, f"Resource {rid} not found" context.rr_resource = r @then('the retrieved resource namespaced_name is "{expected}"') def step_resource_registry_resource_name(context: Any, expected: str) -> None: assert context.rr_resource.namespaced_name == expected @then('the retrieved resource namespace is "{expected}"') def step_resource_registry_resource_namespace(context: Any, expected: str) -> None: assert context.rr_resource.namespace == expected @then('the retrieved resource type_name is "{expected}"') def step_resource_registry_resource_type(context: Any, expected: str) -> None: assert context.rr_resource.type_name == expected @then('the retrieved resource kind is "{expected}"') def step_resource_registry_resource_kind(context: Any, expected: str) -> None: assert context.rr_resource.resource_kind == expected @then('the retrieved resource location is "{expected}"') def step_resource_registry_resource_location(context: Any, expected: str) -> None: assert context.rr_resource.location == expected @when('I create an unnamed resource of type "{tname}"') def step_resource_registry_create_unnamed(context: Any, tname: str) -> None: session: Session = context.rr_session rid = _make_ulid("unnamed1") rt = session.get(ResourceTypeModel, tname) kind = rt.resource_kind if rt else "virtual" r = ResourceModel() r.resource_id = rid r.namespaced_name = None r.namespace = None r.type_name = tname r.resource_kind = kind r.created_at = _now_iso() r.updated_at = _now_iso() session.add(r) session.commit() context.rr_resource = r @then("I can retrieve the unnamed resource by id") def step_resource_registry_get_unnamed(context: Any) -> None: session: Session = context.rr_session rid = context.rr_resource.resource_id r = session.get(ResourceModel, rid) assert r is not None, f"Resource {rid} not found" context.rr_resource = r @then("the unnamed resource namespaced_name is null") def step_resource_registry_unnamed_null(context: Any) -> None: assert context.rr_resource.namespaced_name is None @when("I create a resource with JSON properties") def step_resource_registry_create_resource_json(context: Any) -> None: session: Session = context.rr_session rid = _make_ulid("jsonprops") r = ResourceModel() r.resource_id = rid r.namespaced_name = "local/json-test" r.namespace = "local" r.type_name = "builtin/git-checkout" r.resource_kind = "physical" r.location = "/path/to/repo" r.properties_json = json.dumps({"branch": "main", "remote": "origin"}) r.metadata_json = json.dumps({"tags": ["important"], "priority": 1}) r.created_at = _now_iso() r.updated_at = _now_iso() session.add(r) session.commit() context.rr_resource = r @then("the resource properties_json round-trips correctly") def step_resource_registry_resource_props(context: Any) -> None: data = json.loads(context.rr_resource.properties_json) assert data["branch"] == "main" assert data["remote"] == "origin" @then("the resource metadata_json round-trips correctly") def step_resource_registry_resource_meta(context: Any) -> None: data = json.loads(context.rr_resource.metadata_json) assert data["tags"] == ["important"] assert data["priority"] == 1 @then('creating a resource with kind "{kind}" raises an integrity error') def step_resource_registry_invalid_resource_kind(context: Any, kind: str) -> None: session: Session = context.rr_session try: session.execute( text( "INSERT INTO resources (resource_id, type_name, resource_kind, " "read_only, auto_discovered, created_at, updated_at) " "VALUES (:rid, :tname, :kind, 0, 0, :ts, :ts)" ), { "rid": _make_ulid("badkind"), "tname": "builtin/git-checkout", "kind": kind, "ts": _now_iso(), }, ) session.commit() msg = "Expected IntegrityError for invalid resource_kind" raise AssertionError(msg) except IntegrityError: session.rollback() @then("creating a resource with non-existent type raises an integrity error") def step_resource_registry_missing_type(context: Any) -> None: session: Session = context.rr_session r = ResourceModel() r.resource_id = _make_ulid("orphan") r.type_name = "nonexistent/type" r.resource_kind = "physical" r.created_at = _now_iso() r.updated_at = _now_iso() try: session.add(r) session.commit() msg = "Expected IntegrityError for missing FK" raise AssertionError(msg) except IntegrityError: session.rollback() @then('creating a duplicate named resource "{name}" raises an integrity error') def step_resource_registry_duplicate_resource(context: Any, name: str) -> None: session: Session = context.rr_session r = ResourceModel() r.resource_id = _make_ulid("dup") r.namespaced_name = name r.namespace = name.split("/")[0] r.type_name = "builtin/git-checkout" r.resource_kind = "physical" r.created_at = _now_iso() r.updated_at = _now_iso() try: session.add(r) session.commit() msg = "Expected IntegrityError for duplicate named resource" raise AssertionError(msg) except IntegrityError: session.rollback() @then('deleting resource type "{name}" raises an integrity error') def step_resource_registry_restrict_type_delete(context: Any, name: str) -> None: session: Session = context.rr_session rt = session.get(ResourceTypeModel, name) assert rt is not None try: session.delete(rt) session.flush() msg = "Expected IntegrityError for RESTRICT delete" raise AssertionError(msg) except IntegrityError: session.rollback() @then('querying resources by type "{tname}" returns {count:d} result') def step_resource_registry_query_by_type(context: Any, tname: str, count: int) -> None: session: Session = context.rr_session results = ( session.query(ResourceModel).filter(ResourceModel.type_name == tname).all() ) assert len(results) == count, f"Expected {count}, got {len(results)}" @then('querying resources by namespace "{ns}" returns {count:d} result') def step_resource_registry_query_by_ns(context: Any, ns: str, count: int) -> None: session: Session = context.rr_session results = session.query(ResourceModel).filter(ResourceModel.namespace == ns).all() assert len(results) == count, f"Expected {count}, got {len(results)}" # --------------------------------------------------------------------------- # ResourceEdgeModel - prerequisites # --------------------------------------------------------------------------- @given('a parent resource "{rname}" of type "{tname}"') def step_resource_registry_create_parent(context: Any, rname: str, tname: str) -> None: session: Session = context.rr_session rid = _make_ulid("parent_" + rname.replace("/", "_")) rt = session.get(ResourceTypeModel, tname) kind = rt.resource_kind if rt else "physical" r = ResourceModel() r.resource_id = rid r.namespaced_name = rname r.namespace = rname.split("/")[0] r.type_name = tname r.resource_kind = kind r.created_at = _now_iso() r.updated_at = _now_iso() session.add(r) session.commit() context.rr_parent_resource = r @given('a second parent resource "{rname}" of type "{tname}"') def step_resource_registry_create_second_parent( context: Any, rname: str, tname: str ) -> None: session: Session = context.rr_session rid = _make_ulid("parent2_" + rname.replace("/", "_")) rt = session.get(ResourceTypeModel, tname) kind = rt.resource_kind if rt else "physical" r = ResourceModel() r.resource_id = rid r.namespaced_name = rname r.namespace = rname.split("/")[0] r.type_name = tname r.resource_kind = kind r.created_at = _now_iso() r.updated_at = _now_iso() session.add(r) session.commit() context.rr_parent_resource_2 = r @given('a child resource of type "{tname}"') def step_resource_registry_create_child(context: Any, tname: str) -> None: session: Session = context.rr_session rid = _make_ulid("child1") rt = session.get(ResourceTypeModel, tname) kind = rt.resource_kind if rt else "physical" r = ResourceModel() r.resource_id = rid r.type_name = tname r.resource_kind = kind r.created_at = _now_iso() r.updated_at = _now_iso() session.add(r) session.commit() context.rr_child_resource = r @given('a first child resource of type "{tname}"') def step_resource_registry_create_first_child(context: Any, tname: str) -> None: session: Session = context.rr_session rid = _make_ulid("child_a") rt = session.get(ResourceTypeModel, tname) kind = rt.resource_kind if rt else "physical" r = ResourceModel() r.resource_id = rid r.type_name = tname r.resource_kind = kind r.created_at = _now_iso() r.updated_at = _now_iso() session.add(r) session.commit() context.rr_child_resource = r @given('a second child resource of type "{tname}"') def step_resource_registry_create_second_child(context: Any, tname: str) -> None: session: Session = context.rr_session rid = _make_ulid("child_b") rt = session.get(ResourceTypeModel, tname) kind = rt.resource_kind if rt else "physical" r = ResourceModel() r.resource_id = rid r.type_name = tname r.resource_kind = kind r.created_at = _now_iso() r.updated_at = _now_iso() session.add(r) session.commit() context.rr_child_resource_2 = r # --------------------------------------------------------------------------- # ResourceEdgeModel CRUD # --------------------------------------------------------------------------- @given('I create an edge from parent to child with link_type "{lt}"') @when('I create an edge from parent to child with link_type "{lt}"') def step_resource_registry_create_edge(context: Any, lt: str) -> None: session: Session = context.rr_session edge = ResourceEdgeModel() edge.parent_id = context.rr_parent_resource.resource_id edge.child_id = context.rr_child_resource.resource_id edge.link_type = lt edge.created_at = _now_iso() session.add(edge) session.commit() context.rr_edge = edge @when('I create an edge from second parent to child with link_type "{lt}"') def step_resource_registry_create_edge_second_parent(context: Any, lt: str) -> None: session: Session = context.rr_session edge = ResourceEdgeModel() edge.parent_id = context.rr_parent_resource_2.resource_id edge.child_id = context.rr_child_resource.resource_id edge.link_type = lt edge.created_at = _now_iso() session.add(edge) session.commit() @when("I create edges from parent to both children") def step_resource_registry_create_edges_multi(context: Any) -> None: session: Session = context.rr_session e1 = ResourceEdgeModel() e1.parent_id = context.rr_parent_resource.resource_id e1.child_id = context.rr_child_resource.resource_id e1.link_type = "contains" e1.created_at = _now_iso() session.add(e1) e2 = ResourceEdgeModel() e2.parent_id = context.rr_parent_resource.resource_id e2.child_id = context.rr_child_resource_2.resource_id e2.link_type = "contains" e2.created_at = _now_iso() session.add(e2) session.commit() @then("the edge exists in the database") def step_resource_registry_edge_exists(context: Any) -> None: session: Session = context.rr_session edge = ( session.query(ResourceEdgeModel) .filter( ResourceEdgeModel.parent_id == context.rr_parent_resource.resource_id, ResourceEdgeModel.child_id == context.rr_child_resource.resource_id, ) .first() ) assert edge is not None @then('the edge link_type is "{expected}"') def step_resource_registry_edge_link_type(context: Any, expected: str) -> None: assert context.rr_edge.link_type == expected @then("querying children of the parent returns {count:d} result") def step_resource_registry_children_count_single(context: Any, count: int) -> None: session: Session = context.rr_session edges = ( session.query(ResourceEdgeModel) .filter(ResourceEdgeModel.parent_id == context.rr_parent_resource.resource_id) .all() ) assert len(edges) == count, f"Expected {count}, got {len(edges)}" @then("querying children of the parent returns {count:d} results") def step_resource_registry_children_count(context: Any, count: int) -> None: session: Session = context.rr_session edges = ( session.query(ResourceEdgeModel) .filter(ResourceEdgeModel.parent_id == context.rr_parent_resource.resource_id) .all() ) assert len(edges) == count, f"Expected {count}, got {len(edges)}" @then("querying parents of the child returns {count:d} result") def step_resource_registry_parents_count_single(context: Any, count: int) -> None: session: Session = context.rr_session edges = ( session.query(ResourceEdgeModel) .filter(ResourceEdgeModel.child_id == context.rr_child_resource.resource_id) .all() ) assert len(edges) == count, f"Expected {count}, got {len(edges)}" @then("querying parents of the child returns {count:d} results") def step_resource_registry_parents_count(context: Any, count: int) -> None: session: Session = context.rr_session edges = ( session.query(ResourceEdgeModel) .filter(ResourceEdgeModel.child_id == context.rr_child_resource.resource_id) .all() ) assert len(edges) == count, f"Expected {count}, got {len(edges)}" @then("creating a duplicate edge raises an integrity error") def step_resource_registry_duplicate_edge(context: Any) -> None: session: Session = context.rr_session edge = ResourceEdgeModel() edge.parent_id = context.rr_parent_resource.resource_id edge.child_id = context.rr_child_resource.resource_id edge.link_type = "contains" edge.created_at = _now_iso() try: session.add(edge) session.commit() msg = "Expected IntegrityError for duplicate edge" raise AssertionError(msg) except IntegrityError: session.rollback() @then("creating a self-loop edge raises an integrity error") def step_resource_registry_self_loop(context: Any) -> None: session: Session = context.rr_session parent_id = context.rr_parent_resource.resource_id try: session.execute( text( "INSERT INTO resource_edges (parent_id, child_id, link_type, " "auto_discovered, created_at) " "VALUES (:pid, :cid, 'contains', 0, :ts)" ), {"pid": parent_id, "cid": parent_id, "ts": _now_iso()}, ) session.commit() msg = "Expected IntegrityError for self-loop" raise AssertionError(msg) except IntegrityError: session.rollback() @then("creating an edge with invalid link_type raises an integrity error") def step_resource_registry_invalid_link_type(context: Any) -> None: session: Session = context.rr_session try: session.execute( text( "INSERT INTO resource_edges (parent_id, child_id, link_type, " "auto_discovered, created_at) " "VALUES (:pid, :cid, 'invalid_type', 0, :ts)" ), { "pid": context.rr_parent_resource.resource_id, "cid": context.rr_child_resource.resource_id, "ts": _now_iso(), }, ) session.commit() msg = "Expected IntegrityError for invalid link_type" raise AssertionError(msg) except IntegrityError: session.rollback() @when("I delete the parent resource") def step_resource_registry_delete_parent(context: Any) -> None: session: Session = context.rr_session r = session.get(ResourceModel, context.rr_parent_resource.resource_id) assert r is not None session.delete(r) session.commit() @when("I delete the child resource") def step_resource_registry_delete_child(context: Any) -> None: session: Session = context.rr_session r = session.get(ResourceModel, context.rr_child_resource.resource_id) assert r is not None session.delete(r) session.commit() @then("the edge no longer exists in the database") def step_resource_registry_edge_gone(context: Any) -> None: session: Session = context.rr_session count = session.query(ResourceEdgeModel).count() assert count == 0, f"Expected 0 edges, got {count}" # --------------------------------------------------------------------------- # ORM relationship navigation # --------------------------------------------------------------------------- @then('navigating from resource to resource_type returns "{expected}"') def step_resource_registry_nav_to_type(context: Any, expected: str) -> None: session: Session = context.rr_session r = session.get(ResourceModel, context.rr_resource.resource_id) assert r is not None assert r.resource_type is not None assert r.resource_type.name == expected @then("navigating from resource_type to resources returns {count:d} items") def step_resource_registry_nav_to_resources(context: Any, count: int) -> None: session: Session = context.rr_session rt = session.get(ResourceTypeModel, "builtin/git-checkout") assert rt is not None assert len(rt.resources) == count, f"Expected {count}, got {len(rt.resources)}" @then("navigating from parent to child_edges returns {count:d} edge") def step_resource_registry_nav_child_edges(context: Any, count: int) -> None: session: Session = context.rr_session r = session.get(ResourceModel, context.rr_parent_resource.resource_id) assert r is not None assert len(r.child_edges) == count, f"Expected {count}, got {len(r.child_edges)}" @then("navigating from child to parent_edges returns {count:d} edge") def step_resource_registry_nav_parent_edges(context: Any, count: int) -> None: session: Session = context.rr_session r = session.get(ResourceModel, context.rr_child_resource.resource_id) assert r is not None assert len(r.parent_edges) == count, f"Expected {count}, got {len(r.parent_edges)}" # --------------------------------------------------------------------------- # Timestamps # --------------------------------------------------------------------------- @then("the resource type has valid created_at and updated_at timestamps") def step_resource_registry_type_timestamps(context: Any) -> None: rt = context.rr_resource_type assert rt.created_at is not None assert rt.updated_at is not None # Parse to verify ISO-8601 format datetime.fromisoformat(rt.created_at) datetime.fromisoformat(rt.updated_at) @then("the resource has valid created_at and updated_at timestamps") def step_resource_registry_resource_timestamps(context: Any) -> None: r = context.rr_resource assert r.created_at is not None assert r.updated_at is not None datetime.fromisoformat(r.created_at) datetime.fromisoformat(r.updated_at) @then("the edge has a valid created_at timestamp") def step_resource_registry_edge_timestamp(context: Any) -> None: edge = context.rr_edge assert edge.created_at is not None datetime.fromisoformat(edge.created_at) # --------------------------------------------------------------------------- # Migration smoke tests # --------------------------------------------------------------------------- @when("I run init_database to simulate migration") def step_run_init_database(context: Any) -> None: """Run init_database on a fresh temp file to simulate migration.""" import tempfile from cleveragents.infrastructure.database.models import init_database tmp = tempfile.mktemp(suffix=".db") db_url = f"sqlite:///{tmp}" engine = init_database(db_url) context.rr_migrated_engine = engine context.rr_migrated_tmp = tmp @when("I run init_database twice on the same database file") def step_run_init_database_twice(context: Any) -> None: """Run init_database twice on the same file to verify idempotency.""" import tempfile from cleveragents.infrastructure.database.models import init_database tmp = tempfile.mktemp(suffix=".db") db_url = f"sqlite:///{tmp}" engine1 = init_database(db_url) engine1.dispose() engine2 = init_database(db_url) context.rr_migrated_engine = engine2 context.rr_migrated_tmp = tmp @then('the migrated database should contain table "{table_name}"') def step_migrated_table_exists(context: Any, table_name: str) -> None: inspector = inspect(context.rr_migrated_engine) tables = inspector.get_table_names() assert table_name in tables, f"Table {table_name} not found. Tables: {tables}"