"""Step definitions for project repository tests. Tests the NamespacedProjectRepository and ProjectResourceLinkRepository classes with an in-memory SQLite database. """ from __future__ import annotations 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, text from sqlalchemy.orm import Session, sessionmaker from cleveragents.domain.models.core.project import NamespacedProject from cleveragents.infrastructure.database.models import ( Base, NamespacedProjectModel, ResourceModel, ResourceTypeModel, ) from cleveragents.infrastructure.database.repositories import ( DuplicateLinkError, # noqa: F401 NamespacedProjectRepository, ProjectNotFoundError, ProjectResourceLinkRepository, ) def _now_iso() -> str: """Return current UTC time as ISO-8601 string.""" return datetime.now(tz=UTC).isoformat() def _set_sqlite_pragma_pr(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() def _setup_resource_type_pr(session: Session) -> None: """Create a resource type so FK constraints pass for links.""" existing = ( session.query(ResourceTypeModel).filter_by(name="builtin/git-checkout").first() ) if existing is not None: return rt = ResourceTypeModel() rt.name = "builtin/git-checkout" # type: ignore[assignment] rt.namespace = "builtin" # type: ignore[assignment] rt.resource_kind = "physical" # type: ignore[assignment] rt.user_addable = True # type: ignore[assignment] rt.created_at = _now_iso() # type: ignore[assignment] rt.updated_at = _now_iso() # type: ignore[assignment] session.add(rt) session.commit() # --------------------------------------------------------------------------- # Background # --------------------------------------------------------------------------- @given("a project repository in-memory database is initialized") def step_pr_init_db(context: Any) -> None: """Create an in-memory SQLite database with all tables.""" engine = create_engine("sqlite:///:memory:") event.listen(engine, "connect", _set_sqlite_pragma_pr) with engine.connect() as conn: conn.execute(text("PRAGMA foreign_keys = ON")) conn.commit() Base.metadata.create_all(engine) sf = sessionmaker(bind=engine) context.pr_engine = engine context.pr_session_factory = sf session = sf() context.pr_session = session # Set up the resource type for FK constraints _setup_resource_type_pr(context.pr_session) # Use a lambda that always returns the SAME session so that # context.pr_session.commit() commits data flushed by the repos. context.pr_project_repo = NamespacedProjectRepository( session_factory=lambda: session, ) context.pr_link_repo = ProjectResourceLinkRepository( session_factory=lambda: session, ) context.pr_error = None context.pr_result = None context.pr_project = None context.pr_projects = [] context.pr_link = None context.pr_links = [] context.pr_delete_result = None context.pr_remove_result = None context.pr_stored_link_id = None # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_project(namespaced_name: str) -> NamespacedProject: """Create a minimal NamespacedProject domain object.""" parts = namespaced_name.split("/", 1) namespace = parts[0] if len(parts) > 1 else "local" name = parts[1] if len(parts) > 1 else parts[0] return NamespacedProject(name=name, namespace=namespace) def _create_resource(session: Session, resource_id: str) -> None: """Create a minimal resource so FK constraints pass.""" existing = session.query(ResourceModel).filter_by(resource_id=resource_id).first() if existing is not None: return r = ResourceModel() r.resource_id = resource_id # type: ignore[assignment] r.type_name = "builtin/git-checkout" # type: ignore[assignment] r.resource_kind = "physical" # type: ignore[assignment] r.created_at = _now_iso() # type: ignore[assignment] r.updated_at = _now_iso() # type: ignore[assignment] session.add(r) session.commit() # --------------------------------------------------------------------------- # NamespacedProjectRepository steps # --------------------------------------------------------------------------- @when('I create a namespaced project "{ns_name}" via the repository') def step_pr_create_project(context: Any, ns_name: str) -> None: project = _make_project(ns_name) context.pr_project = context.pr_project_repo.create(project) context.pr_session.commit() @then('the project "{ns_name}" should exist in the repository') def step_pr_project_exists(context: Any, ns_name: str) -> None: project = context.pr_project_repo.get(ns_name) assert project is not None, f"Project {ns_name} not found" @given('a namespaced project "{ns_name}" exists in the repository') def step_pr_given_project(context: Any, ns_name: str) -> None: project = _make_project(ns_name) context.pr_project_repo.create(project) context.pr_session.commit() @when('I get the project "{ns_name}" from the repository') def step_pr_get_project(context: Any, ns_name: str) -> None: context.pr_project = context.pr_project_repo.get(ns_name) @then('the returned project name should be "{name}"') def step_pr_check_name(context: Any, name: str) -> None: assert context.pr_project.name == name @then('the returned project namespace should be "{ns}"') def step_pr_check_namespace(context: Any, ns: str) -> None: assert context.pr_project.namespace == ns @when('I get the project "{ns_name}" from the repository expecting an error') def step_pr_get_project_error(context: Any, ns_name: str) -> None: try: context.pr_project_repo.get(ns_name) context.pr_error = None except Exception as exc: context.pr_error = exc @then('the repository error should be "{error_type}"') def step_pr_check_error_type(context: Any, error_type: str) -> None: assert context.pr_error is not None, "Expected an error but none was raised" actual_type = type(context.pr_error).__name__ assert actual_type == error_type, f"Expected {error_type}, got {actual_type}" @when("I list all projects from the repository") def step_pr_list_all(context: Any) -> None: context.pr_projects = context.pr_project_repo.list_projects() @then("the project list should contain {count:d} projects") def step_pr_list_count(context: Any, count: int) -> None: assert len(context.pr_projects) == count, ( f"Expected {count}, got {len(context.pr_projects)}" ) @when('I list projects with namespace "{ns}" from the repository') def step_pr_list_namespace(context: Any, ns: str) -> None: context.pr_projects = context.pr_project_repo.list_projects(namespace=ns) @then('the first project in the list should be "{ns_name}"') def step_pr_first_project(context: Any, ns_name: str) -> None: assert len(context.pr_projects) > 0, "Project list is empty" assert context.pr_projects[0].namespaced_name == ns_name @when("I list projects with limit {limit:d} from the repository") def step_pr_list_limit(context: Any, limit: int) -> None: context.pr_projects = context.pr_project_repo.list_projects(limit=limit) @when('I update the project "{ns_name}" description to "{desc}"') def step_pr_update_desc(context: Any, ns_name: str, desc: str) -> None: project = context.pr_project_repo.get(ns_name) updated = project.model_copy( update={ "description": desc, "updated_at": datetime.now(tz=UTC), } ) context.pr_project_repo.update(updated) context.pr_session.commit() @then('the project "{ns_name}" description should be "{desc}"') def step_pr_check_desc(context: Any, ns_name: str, desc: str) -> None: project = context.pr_project_repo.get(ns_name) assert project.description == desc, ( f"Expected '{desc}', got '{project.description}'" ) @when('I update a non-existent project "{ns_name}" expecting an error') def step_pr_update_not_found(context: Any, ns_name: str) -> None: try: fake = _make_project(ns_name) context.pr_project_repo.update(fake) context.pr_error = None except Exception as exc: context.pr_error = exc @when('I delete the project "{ns_name}" from the repository') def step_pr_delete_project(context: Any, ns_name: str) -> None: context.pr_delete_result = context.pr_project_repo.delete(ns_name) context.pr_session.commit() @then("the delete result should be True") def step_pr_delete_true(context: Any) -> None: assert context.pr_delete_result is True @then("the delete result should be False") def step_pr_delete_false(context: Any) -> None: assert context.pr_delete_result is False @then('getting "{ns_name}" should raise ProjectNotFoundError') def step_pr_get_raises(context: Any, ns_name: str) -> None: raised = False try: context.pr_project_repo.get(ns_name) except ProjectNotFoundError: raised = True assert raised, f"Expected ProjectNotFoundError for {ns_name}" @given('a resource link exists for project "{ns_name}"') def step_pr_given_link_for_cascade(context: Any, ns_name: str) -> None: """Create a resource and link it to the project for cascade testing.""" resource_id = "00000000000000000000000099" _create_resource(context.pr_session, resource_id) context.pr_link_repo.create_link( project_name=ns_name, resource_id=resource_id, ) context.pr_session.commit() @then('the links for project "{ns_name}" should be empty') def step_pr_links_empty(context: Any, ns_name: str) -> None: links = context.pr_link_repo.list_links(ns_name) assert len(links) == 0, f"Expected 0 links, got {len(links)}" @when('I create a duplicate project "{ns_name}" expecting an error') def step_pr_create_dup(context: Any, ns_name: str) -> None: """Attempt to create a duplicate project. Uses a direct session insert to bypass retry logic and reliably trigger the IntegrityError / DatabaseError. """ from sqlalchemy.exc import IntegrityError as _IntegrityError session = context.pr_session_factory() try: dup_model = NamespacedProjectModel( namespaced_name=ns_name, namespace=ns_name.split("/")[0], tags_json="[]", created_at=_now_iso(), updated_at=_now_iso(), ) session.add(dup_model) session.flush() session.commit() context.pr_error = None except _IntegrityError: session.rollback() from cleveragents.core.exceptions import DatabaseError as _DBErr context.pr_error = _DBErr(f"Project '{ns_name}' already exists") except Exception as exc: session.rollback() context.pr_error = exc @then('the repository error message should contain "{text}"') def step_pr_error_contains(context: Any, text: str) -> None: assert context.pr_error is not None, "Expected an error" assert text in str(context.pr_error), f"Expected '{text}' in '{context.pr_error}'" # --------------------------------------------------------------------------- # ProjectResourceLinkRepository steps # --------------------------------------------------------------------------- @given('a resource exists with id "{resource_id}"') def step_pr_given_resource(context: Any, resource_id: str) -> None: _create_resource(context.pr_session, resource_id) @when('I create a link for project "{ns_name}" to resource "{resource_id}"') def step_pr_create_link(context: Any, ns_name: str, resource_id: str) -> None: context.pr_link = context.pr_link_repo.create_link( project_name=ns_name, resource_id=resource_id, ) context.pr_session.commit() context.pr_stored_link_id = context.pr_link.link_id @then("the link should have a valid link_id") def step_pr_link_id_valid(context: Any) -> None: assert context.pr_link is not None assert context.pr_link.link_id is not None assert len(str(context.pr_link.link_id)) == 26 @then('the link project_name should be "{ns_name}"') def step_pr_link_project_name(context: Any, ns_name: str) -> None: assert str(context.pr_link.project_name) == ns_name @when( 'I create a link for project "{ns_name}" to resource "{resource_id}" without alias' ) def step_pr_create_link_no_alias(context: Any, ns_name: str, resource_id: str) -> None: context.pr_link = context.pr_link_repo.create_link( project_name=ns_name, resource_id=resource_id, ) context.pr_session.commit() context.pr_stored_link_id = context.pr_link.link_id @when( 'I create an aliased link for project "{ns_name}" resource "{resource_id}" alias "{alias}"' ) def step_pr_create_link_alias( context: Any, ns_name: str, resource_id: str, alias: str ) -> None: context.pr_link = context.pr_link_repo.create_link( project_name=ns_name, resource_id=resource_id, alias=alias, ) context.pr_session.commit() context.pr_stored_link_id = context.pr_link.link_id @then('the link alias should be "{alias}"') def step_pr_link_alias(context: Any, alias: str) -> None: assert str(context.pr_link.alias) == alias @given( 'an aliased link exists for project "{ns_name}" resource "{resource_id}" alias "{alias}"' ) def step_pr_given_link_alias( context: Any, ns_name: str, resource_id: str, alias: str ) -> None: context.pr_link = context.pr_link_repo.create_link( project_name=ns_name, resource_id=resource_id, alias=alias, ) context.pr_session.commit() context.pr_stored_link_id = context.pr_link.link_id @when( 'I create a duplicate alias link for project "{ns_name}" resource "{resource_id}" alias "{alias}"' ) def step_pr_create_link_alias_error( context: Any, ns_name: str, resource_id: str, alias: str ) -> None: try: context.pr_link_repo.create_link( project_name=ns_name, resource_id=resource_id, alias=alias, ) context.pr_session.commit() context.pr_error = None except Exception as exc: context.pr_error = exc @given( 'a link exists for project "{ns_name}" to resource "{resource_id}" without alias' ) def step_pr_given_link(context: Any, ns_name: str, resource_id: str) -> None: context.pr_link = context.pr_link_repo.create_link( project_name=ns_name, resource_id=resource_id, ) context.pr_session.commit() context.pr_stored_link_id = context.pr_link.link_id @when('I list links for project "{ns_name}"') def step_pr_list_links(context: Any, ns_name: str) -> None: context.pr_links = context.pr_link_repo.list_links(ns_name) @then("the link list should contain {count:d} links") def step_pr_link_count(context: Any, count: int) -> None: assert len(context.pr_links) == count, ( f"Expected {count}, got {len(context.pr_links)}" ) @when("I get the link by its stored id") def step_pr_get_link_by_stored(context: Any) -> None: assert context.pr_stored_link_id is not None context.pr_link = context.pr_link_repo.get_link(context.pr_stored_link_id) @then("the returned link should not be None") def step_pr_link_not_none(context: Any) -> None: assert context.pr_link is not None @when('I get a link with id "{link_id}"') def step_pr_get_link_by_id(context: Any, link_id: str) -> None: context.pr_link = context.pr_link_repo.get_link(link_id) @then("the returned link should be None") def step_pr_link_is_none(context: Any) -> None: assert context.pr_link is None @when("I remove the link by its stored id") def step_pr_remove_link_by_stored(context: Any) -> None: assert context.pr_stored_link_id is not None context.pr_remove_result = context.pr_link_repo.remove_link( context.pr_stored_link_id ) context.pr_session.commit() @then("the remove result should be True") def step_pr_remove_true(context: Any) -> None: assert context.pr_remove_result is True @then('the link list for project "{ns_name}" should be empty') def step_pr_link_list_empty(context: Any, ns_name: str) -> None: links = context.pr_link_repo.list_links(ns_name) assert len(links) == 0, f"Expected 0 links, got {len(links)}" @when('I remove a link with id "{link_id}"') def step_pr_remove_link_by_id(context: Any, link_id: str) -> None: context.pr_remove_result = context.pr_link_repo.remove_link(link_id) @then("the remove result should be False") def step_pr_remove_false(context: Any) -> None: assert context.pr_remove_result is False @when('I create a read-only link for project "{ns_name}" to resource "{resource_id}"') def step_pr_create_ro_link(context: Any, ns_name: str, resource_id: str) -> None: context.pr_link = context.pr_link_repo.create_link( project_name=ns_name, resource_id=resource_id, read_only=True, ) context.pr_session.commit() @then("the link read_only flag should be True") def step_pr_link_readonly(context: Any) -> None: assert bool(context.pr_link.read_only) is True # --------------------------------------------------------------------------- # Cross-session persistence verification steps # --------------------------------------------------------------------------- @then("the link should be visible in a new session from the same engine") def step_pr_link_visible_new_session(context: Any) -> None: """Open a fresh session from the same engine and verify the link exists. This guards against the bug where ``create_link()`` only called ``flush()`` without ``commit()``, making data visible within the same session but lost when a different session queries the database. """ from cleveragents.infrastructure.database.models import ProjectResourceLinkModel new_session = context.pr_session_factory() try: row = ( new_session.query(ProjectResourceLinkModel) .filter_by(link_id=context.pr_stored_link_id) .first() ) assert row is not None, ( f"Link {context.pr_stored_link_id} not found in a new session — " "commit() may be missing" ) finally: new_session.close() @then("the removed link should be absent in a new session from the same engine") def step_pr_removed_link_absent_new_session(context: Any) -> None: """Open a fresh session and confirm the removed link no longer exists. This guards against the bug where ``remove_link()`` only called ``flush()`` without ``commit()``, making the deletion visible within the same session but not persisted across sessions. """ from cleveragents.infrastructure.database.models import ProjectResourceLinkModel new_session = context.pr_session_factory() try: row = ( new_session.query(ProjectResourceLinkModel) .filter_by(link_id=context.pr_stored_link_id) .first() ) assert row is None, ( f"Link {context.pr_stored_link_id} still present in a new session — " "commit() may be missing from remove_link()" ) finally: new_session.close()