"""Step definitions for project_cli.feature. All domain/infrastructure imports are deferred to function bodies so that behave can load this step module even when the ``/app/src`` shadow on ``sys.path`` hides the dependency-branch additions. """ from __future__ import annotations import json import re from typing import Any from behave import given, then, when # type: ignore[import-untyped] from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker # --------------------------------------------------------------------------- # Shared session wrapper (reuse pattern from resource_project_services_steps) # --------------------------------------------------------------------------- class _UnclosableSession: """Wraps a SQLAlchemy Session but makes ``close()`` a no-op.""" def __init__(self, real_session: Session) -> None: object.__setattr__(self, "_real", real_session) def close(self) -> None: """No-op so the shared session stays usable across calls.""" def __getattr__(self, name: str) -> Any: return getattr(object.__getattribute__(self, "_real"), name) def __setattr__(self, name: str, value: Any) -> None: setattr(object.__getattribute__(self, "_real"), name, value) def _make_pcli_session_factory(context: Any) -> tuple[Any, Any]: """Create an in-memory SQLite database and return (engine, session_factory).""" from cleveragents.infrastructure.database.models import Base 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, )() wrapper = _UnclosableSession(real_session) def _factory() -> Any: return wrapper return engine, _factory # --------------------------------------------------------------------------- # Background # --------------------------------------------------------------------------- @given("a project CLI in-memory database is initialized") def step_init_project_cli_db(context: Any) -> None: from cleveragents.application.services.resource_registry_service import ( ResourceRegistryService, ) from cleveragents.infrastructure.database.repositories import ( NamespacedProjectRepository, ProjectResourceLinkRepository, ) engine, session_factory = _make_pcli_session_factory(context) context.pcli_engine = engine context.pcli_session_factory = session_factory context.pcli_project_repo = NamespacedProjectRepository( session_factory=session_factory ) context.pcli_link_repo = ProjectResourceLinkRepository( session_factory=session_factory ) context.pcli_resource_svc = ResourceRegistryService(session_factory=session_factory) # Bootstrap built-in types so we can register resources context.pcli_resource_svc.bootstrap_builtin_types() context.pcli_output = "" context.pcli_exit_code = 0 context.pcli_spec_dict = {} context.pcli_resource_map = {} context.pcli_last_resource_id = None # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _pcli_create_project( context: Any, name: str, description: str | None = None ) -> Any: """Create a project in the test DB.""" from cleveragents.domain.models.core.project import ( NamespacedProject, parse_namespaced_name, ) parsed = parse_namespaced_name(name) proj = NamespacedProject( name=parsed.name, namespace=parsed.namespace, server=parsed.server, description=description, ) context.pcli_project_repo.create(proj) return proj def _pcli_register_resource(context: Any, resource_alias: str) -> str: """Register a dummy resource and return its resource_id.""" res = context.pcli_resource_svc.register_resource( type_name="git-checkout", name=f"local/{resource_alias}", location="/tmp/test-resource", description=f"Test resource {resource_alias}", ) return res.resource_id # --------------------------------------------------------------------------- # Create steps # --------------------------------------------------------------------------- @when('I run project create with name "{name}"') def step_pcli_create_project(context: Any, name: str) -> None: from cleveragents.domain.models.core.project import ( NamespacedProject, parse_namespaced_name, ) from cleveragents.infrastructure.database.repositories import ( ProjectNotFoundError, ) try: parsed = parse_namespaced_name(name) proj = NamespacedProject( name=parsed.name, namespace=parsed.namespace, server=parsed.server, ) # Check for duplicate before creating (mirrors real CLI behaviour) try: context.pcli_project_repo.get(proj.namespaced_name) # If get succeeds the project already exists raise ValueError(f"Project '{proj.namespaced_name}' already exists") except ProjectNotFoundError: pass # Good - does not exist yet context.pcli_project_repo.create(proj) context.pcli_exit_code = 0 context.pcli_output = f"Project '{proj.namespaced_name}' created." context.pcli_last_project = proj except Exception as exc: context.pcli_exit_code = 1 context.pcli_output = str(exc) @when('I run project create named "{name}" described as "{desc}"') def step_pcli_create_project_with_desc(context: Any, name: str, desc: str) -> None: from cleveragents.domain.models.core.project import ( NamespacedProject, parse_namespaced_name, ) try: parsed = parse_namespaced_name(name) proj = NamespacedProject( name=parsed.name, namespace=parsed.namespace, server=parsed.server, description=desc, ) context.pcli_project_repo.create(proj) context.pcli_exit_code = 0 context.pcli_output = f"Project '{proj.namespaced_name}' created." context.pcli_last_project = proj except Exception as exc: context.pcli_exit_code = 1 context.pcli_output = str(exc) @when('I run project create with invalid name "{name}"') def step_pcli_create_project_invalid(context: Any, name: str) -> None: from cleveragents.domain.models.core.project import ( NamespacedProject, parse_namespaced_name, ) try: parsed = parse_namespaced_name(name) proj = NamespacedProject( name=parsed.name, namespace=parsed.namespace, server=parsed.server, ) context.pcli_project_repo.create(proj) context.pcli_exit_code = 0 context.pcli_output = f"Project '{proj.namespaced_name}' created." except (ValueError, Exception) as exc: context.pcli_exit_code = 1 context.pcli_output = str(exc) @when('I run project create named "{name}" with invariants "{inv1}" plus "{inv2}"') def step_pcli_create_project_with_invariants( context: Any, name: str, inv1: str, inv2: str ) -> None: from cleveragents.domain.models.core.project import ( NamespacedProject, parse_namespaced_name, ) try: parsed = parse_namespaced_name(name) proj = NamespacedProject( name=parsed.name, namespace=parsed.namespace, server=parsed.server, ) context.pcli_project_repo.create(proj) context.pcli_exit_code = 0 context.pcli_output = f"Project '{proj.namespaced_name}' created." except Exception as exc: context.pcli_exit_code = 1 context.pcli_output = str(exc) def _capture_format_output(data, fmt): """Call format_output capturing stdout (machine-readable formats write there).""" from contextlib import redirect_stdout from io import StringIO from cleveragents.cli.formatting import format_output buf = StringIO() with redirect_stdout(buf): result = format_output(data, fmt) return result or buf.getvalue().rstrip("\n") @when('I run project create named "{name}" formatted as "{fmt}"') def step_pcli_create_project_with_format(context: Any, name: str, fmt: str) -> None: from cleveragents.cli.commands.project import _project_spec_dict from cleveragents.domain.models.core.project import ( NamespacedProject, parse_namespaced_name, ) try: parsed = parse_namespaced_name(name) proj = NamespacedProject( name=parsed.name, namespace=parsed.namespace, server=parsed.server, ) context.pcli_project_repo.create(proj) data = _project_spec_dict(proj) context.pcli_output = _capture_format_output(data, fmt) context.pcli_exit_code = 0 except Exception as exc: context.pcli_exit_code = 1 context.pcli_output = str(exc) # --------------------------------------------------------------------------- # Given helpers # --------------------------------------------------------------------------- @given('a project "{name}" already exists') def step_pcli_project_exists(context: Any, name: str) -> None: _pcli_create_project(context, name) @given('a project "{name}" already exists with description "{desc}"') def step_pcli_project_exists_with_desc(context: Any, name: str, desc: str) -> None: _pcli_create_project(context, name, description=desc) @given('a registered resource "{res_name}" exists') def step_pcli_registered_resource_exists(context: Any, res_name: str) -> None: resource_id = _pcli_register_resource(context, res_name) context.pcli_last_resource_id = resource_id context.pcli_resource_map[res_name] = resource_id @given('a resource is linked to project "{project_name}"') def step_pcli_resource_linked_to_project(context: Any, project_name: str) -> None: safe_alias = project_name.replace("/", "-") resource_id = _pcli_register_resource(context, f"linked-res-{safe_alias}") context.pcli_link_repo.create_link( project_name=project_name, resource_id=resource_id, ) context.pcli_last_resource_id = resource_id # --------------------------------------------------------------------------- # List steps # --------------------------------------------------------------------------- @when("I list projects") def step_pcli_list_projects(context: Any) -> None: try: projects = context.pcli_project_repo.list_projects() if not projects: context.pcli_output = "No projects found." else: names = [p.namespaced_name for p in projects] context.pcli_output = "\n".join(names) context.pcli_exit_code = 0 context.pcli_project_list = projects except Exception as exc: context.pcli_exit_code = 1 context.pcli_output = str(exc) @when('I list projects with namespace "{ns}"') def step_pcli_list_projects_ns(context: Any, ns: str) -> None: try: projects = context.pcli_project_repo.list_projects(namespace=ns) if not projects: context.pcli_output = "No projects found." else: names = [p.namespaced_name for p in projects] context.pcli_output = "\n".join(names) context.pcli_exit_code = 0 context.pcli_project_list = projects except Exception as exc: context.pcli_exit_code = 1 context.pcli_output = str(exc) @when('I list projects with regex "{regex_str}"') def step_pcli_list_projects_regex(context: Any, regex_str: str) -> None: try: projects = context.pcli_project_repo.list_projects() pattern = re.compile(regex_str) filtered = [p for p in projects if pattern.search(p.namespaced_name)] if not filtered: context.pcli_output = "No projects found." else: names = [p.namespaced_name for p in filtered] context.pcli_output = "\n".join(names) context.pcli_exit_code = 0 context.pcli_project_list = filtered except Exception as exc: context.pcli_exit_code = 1 context.pcli_output = str(exc) @when('I list projects with invalid regex "{regex_str}"') def step_pcli_list_projects_invalid_regex(context: Any, regex_str: str) -> None: try: re.compile(regex_str) context.pcli_exit_code = 0 except re.error as exc: context.pcli_exit_code = 1 context.pcli_output = str(exc) @when('I list projects with format "{fmt}"') def step_pcli_list_projects_fmt(context: Any, fmt: str) -> None: from cleveragents.cli.commands.project import _project_spec_dict try: projects = context.pcli_project_repo.list_projects() data = [_project_spec_dict(p) for p in projects] context.pcli_output = _capture_format_output(data, fmt) context.pcli_exit_code = 0 except Exception as exc: context.pcli_exit_code = 1 context.pcli_output = str(exc) # --------------------------------------------------------------------------- # Show steps # --------------------------------------------------------------------------- @when('I show project "{name}"') def step_pcli_show_project(context: Any, name: str) -> None: from cleveragents.cli.commands.project import _project_spec_dict from cleveragents.infrastructure.database.repositories import ( ProjectNotFoundError, ) try: proj = context.pcli_project_repo.get(name) data = _project_spec_dict(proj) context.pcli_output = json.dumps(data, default=str) context.pcli_exit_code = 0 except (ProjectNotFoundError, Exception) as exc: context.pcli_exit_code = 1 context.pcli_output = str(exc) @when('I show project details for "{name}" as "{fmt}"') def step_pcli_show_project_fmt(context: Any, name: str, fmt: str) -> None: from cleveragents.cli.commands.project import _project_spec_dict from cleveragents.infrastructure.database.repositories import ( ProjectNotFoundError, ) try: proj = context.pcli_project_repo.get(name) data = _project_spec_dict(proj) context.pcli_output = _capture_format_output(data, fmt) context.pcli_exit_code = 0 except (ProjectNotFoundError, Exception) as exc: context.pcli_exit_code = 1 context.pcli_output = str(exc) # --------------------------------------------------------------------------- # Link / Unlink steps # --------------------------------------------------------------------------- @when('I link resource "{res_name}" to project "{project_name}"') def step_pcli_link_resource(context: Any, res_name: str, project_name: str) -> None: try: resource_id = context.pcli_resource_map.get(res_name) if resource_id is None: context.pcli_exit_code = 1 context.pcli_output = f"Resource '{res_name}' not found" return context.pcli_link_repo.create_link( project_name=project_name, resource_id=resource_id, ) context.pcli_exit_code = 0 context.pcli_output = f"Linked '{res_name}' to '{project_name}'." except Exception as exc: context.pcli_exit_code = 1 context.pcli_output = str(exc) @when('I link resource "{res_name}" to project "{project_name}" with read-only') def step_pcli_link_resource_readonly( context: Any, res_name: str, project_name: str ) -> None: try: resource_id = context.pcli_resource_map.get(res_name) if resource_id is None: context.pcli_exit_code = 1 context.pcli_output = f"Resource '{res_name}' not found" return context.pcli_link_repo.create_link( project_name=project_name, resource_id=resource_id, read_only=True, ) context.pcli_exit_code = 0 context.pcli_output = f"Linked '{res_name}' (read-only) to '{project_name}'." except Exception as exc: context.pcli_exit_code = 1 context.pcli_output = str(exc) @when('I attempt to link to nonexistent project "{project_name}"') def step_pcli_link_to_nonexistent_project(context: Any, project_name: str) -> None: from cleveragents.infrastructure.database.repositories import ( ProjectNotFoundError, ) try: context.pcli_project_repo.get(project_name) context.pcli_exit_code = 0 except ProjectNotFoundError: context.pcli_exit_code = 1 context.pcli_output = f"Project '{project_name}' not found." @when('I link nonexistent resource "{res_name}" to project "{project_name}"') def step_pcli_link_nonexistent_resource( context: Any, res_name: str, project_name: str ) -> None: from cleveragents.core.exceptions import NotFoundError try: context.pcli_resource_svc.show_resource(res_name) context.pcli_exit_code = 0 except NotFoundError: context.pcli_exit_code = 1 context.pcli_output = f"Resource '{res_name}' not found." @when('I unlink the resource from project "{project_name}" with yes') def step_pcli_unlink_resource(context: Any, project_name: str) -> None: try: links = context.pcli_link_repo.list_links(project_name) if not links: context.pcli_exit_code = 1 context.pcli_output = "No resources linked." return link = links[0] context.pcli_link_repo.remove_link(str(link.link_id)) context.pcli_exit_code = 0 context.pcli_output = f"Unlinked resource from '{project_name}'." except Exception as exc: context.pcli_exit_code = 1 context.pcli_output = str(exc) @when('I unlink nonexistent resource from project "{project_name}" with yes') def step_pcli_unlink_nonexistent(context: Any, project_name: str) -> None: try: links = context.pcli_link_repo.list_links(project_name) if not links: context.pcli_exit_code = 1 context.pcli_output = "Resource not linked." return context.pcli_exit_code = 0 except Exception as exc: context.pcli_exit_code = 1 context.pcli_output = str(exc) # --------------------------------------------------------------------------- # Delete steps # --------------------------------------------------------------------------- @when('I delete project "{name}" with yes') def step_pcli_delete_project(context: Any, name: str) -> None: from cleveragents.infrastructure.database.repositories import ( ProjectNotFoundError, ) try: context.pcli_project_repo.get(name) deleted = context.pcli_project_repo.delete(name) context.pcli_exit_code = 0 if deleted else 1 context.pcli_output = f"Project '{name}' deleted." if deleted else "Not found." except ProjectNotFoundError: context.pcli_exit_code = 1 context.pcli_output = f"Project '{name}' not found." @when('I delete project "{name}" with yes but no force') def step_pcli_delete_project_no_force(context: Any, name: str) -> None: from cleveragents.infrastructure.database.repositories import ( ProjectNotFoundError, ) try: proj = context.pcli_project_repo.get(name) if proj.linked_resources: context.pcli_exit_code = 1 context.pcli_output = ( f"Project '{name}' has {len(proj.linked_resources)} " f"linked resource(s). Use --force to delete anyway." ) return deleted = context.pcli_project_repo.delete(name) context.pcli_exit_code = 0 if deleted else 1 except ProjectNotFoundError: context.pcli_exit_code = 1 context.pcli_output = f"Project '{name}' not found." @when('I delete project "{name}" with yes and force') def step_pcli_delete_project_force(context: Any, name: str) -> None: from cleveragents.infrastructure.database.repositories import ( ProjectNotFoundError, ) try: context.pcli_project_repo.get(name) deleted = context.pcli_project_repo.delete(name) context.pcli_exit_code = 0 if deleted else 1 context.pcli_output = f"Project '{name}' deleted." if deleted else "Not found." except ProjectNotFoundError: context.pcli_exit_code = 1 context.pcli_output = f"Project '{name}' not found." # --------------------------------------------------------------------------- # Spec dict steps # --------------------------------------------------------------------------- @when('I generate the project spec dict for "{name}"') def step_pcli_gen_spec_dict(context: Any, name: str) -> None: from cleveragents.cli.commands.project import _project_spec_dict proj = context.pcli_project_repo.get(name) context.pcli_spec_dict = _project_spec_dict(proj) # --------------------------------------------------------------------------- # Then assertions # --------------------------------------------------------------------------- @then('the project "{name}" should exist') def step_pcli_project_should_exist(context: Any, name: str) -> None: from cleveragents.infrastructure.database.repositories import ( ProjectNotFoundError, ) try: context.pcli_project_repo.get(name) except ProjectNotFoundError: raise AssertionError( f"Project '{name}' should exist but was not found" ) from None @then('the project "{name}" should not exist') def step_pcli_project_should_not_exist(context: Any, name: str) -> None: from cleveragents.infrastructure.database.repositories import ( ProjectNotFoundError, ) try: context.pcli_project_repo.get(name) raise AssertionError(f"Project '{name}' should not exist but was found") except ProjectNotFoundError: pass @then('the project namespace should be "{ns}"') def step_pcli_project_ns(context: Any, ns: str) -> None: proj = context.pcli_last_project if proj.namespace != ns: raise AssertionError(f"Expected namespace '{ns}', got '{proj.namespace}'") @then('the project description should be "{desc}"') def step_pcli_project_desc(context: Any, desc: str) -> None: proj = context.pcli_last_project if proj.description != desc: raise AssertionError(f"Expected description '{desc}', got '{proj.description}'") @then("the create command should fail with exit code {code:d}") def step_pcli_create_fail(context: Any, code: int) -> None: if context.pcli_exit_code != code: raise AssertionError( f"Expected exit code {code}, got {context.pcli_exit_code}: " f"{context.pcli_output}" ) @then('the project list should contain "{name}"') def step_pcli_list_contains(context: Any, name: str) -> None: if name not in context.pcli_output: raise AssertionError( f"Expected project list to contain '{name}', output: {context.pcli_output}" ) @then('the project list should not contain "{name}"') def step_pcli_list_not_contains(context: Any, name: str) -> None: if name in context.pcli_output: raise AssertionError( f"Expected project list NOT to contain '{name}', " f"output: {context.pcli_output}" ) @then('the project CLI output should contain "{text}"') def step_pcli_output_contains(context: Any, text: str) -> None: if text not in context.pcli_output: raise AssertionError( f"Expected output to contain '{text}', output: {context.pcli_output}" ) @then('the show output should contain "{text}"') def step_pcli_show_output_contains(context: Any, text: str) -> None: if text not in context.pcli_output: raise AssertionError( f"Expected show output to contain '{text}', output: {context.pcli_output}" ) @then("the show command should fail with exit code {code:d}") def step_pcli_show_fail(context: Any, code: int) -> None: if context.pcli_exit_code != code: raise AssertionError(f"Expected exit code {code}, got {context.pcli_exit_code}") @then("the list command should fail with exit code {code:d}") def step_pcli_list_fail(context: Any, code: int) -> None: if context.pcli_exit_code != code: raise AssertionError(f"Expected exit code {code}, got {context.pcli_exit_code}") @then("the link should succeed") def step_pcli_link_succeed(context: Any) -> None: if context.pcli_exit_code != 0: raise AssertionError( f"Link failed with exit code {context.pcli_exit_code}: " f"{context.pcli_output}" ) @then("the link command should fail with exit code {code:d}") def step_pcli_link_fail(context: Any, code: int) -> None: if context.pcli_exit_code != code: raise AssertionError( f"Expected exit code {code}, got {context.pcli_exit_code}: " f"{context.pcli_output}" ) @then("the unlink should succeed") def step_pcli_unlink_succeed(context: Any) -> None: if context.pcli_exit_code != 0: raise AssertionError( f"Unlink failed with exit code {context.pcli_exit_code}: " f"{context.pcli_output}" ) @then("the unlink command should fail with exit code {code:d}") def step_pcli_unlink_fail(context: Any, code: int) -> None: if context.pcli_exit_code != code: raise AssertionError( f"Expected exit code {code}, got {context.pcli_exit_code}: " f"{context.pcli_output}" ) @then("the delete command should fail with exit code {code:d}") def step_pcli_delete_fail(context: Any, code: int) -> None: if context.pcli_exit_code != code: raise AssertionError( f"Expected exit code {code}, got {context.pcli_exit_code}: " f"{context.pcli_output}" ) @then('the spec dict should have key "{key}"') def step_pcli_spec_dict_has_key(context: Any, key: str) -> None: if key not in context.pcli_spec_dict: raise AssertionError( f"Expected spec dict to have key '{key}', " f"keys: {list(context.pcli_spec_dict.keys())}" )