"""Helper script for project_context_set_exec_env_priority.robot smoke tests. Regression tests for bug #1079: the ``--execution-env-priority`` flag on ``project context set``. Each subcommand is a self-contained check that prints a sentinel on success. """ from __future__ import annotations import json as _json import sys from pathlib import Path from typing import Any from unittest.mock import MagicMock, patch # Ensure the local source tree takes priority over any installed copy. _SRC = str(Path(__file__).resolve().parents[1] / "src") if _SRC not in sys.path: sys.path.insert(0, _SRC) from sqlalchemy import create_engine, text # noqa: E402 from sqlalchemy.orm import sessionmaker # noqa: E402 from typer.testing import CliRunner # noqa: E402 from cleveragents.application.services.context_tiers import ( # noqa: E402 ContextTierService, ) from cleveragents.cli.commands.project_context import ( # noqa: E402 app as project_context_app, ) from cleveragents.infrastructure.database.models import Base # noqa: E402 from cleveragents.infrastructure.database.repositories import ( # noqa: E402 NamespacedProjectRepository, ) runner = CliRunner() # ------------------------------------------------------------------ # In-memory DB setup # ------------------------------------------------------------------ class _SafeSession: """Thin wrapper preventing close() from destroying shared in-memory session.""" def __init__(self, real: Any) -> None: object.__setattr__(self, "_real", real) def close(self) -> None: real: Any = object.__getattribute__(self, "_real") real.rollback() 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 _setup_db() -> tuple[Any, Any, NamespacedProjectRepository]: """Set up in-memory DB, create project, return (engine, session_factory, repo).""" from cleveragents.domain.models.core.project import ( NamespacedProject, parse_namespaced_name, ) engine = create_engine( "sqlite:///:memory:", echo=False, connect_args={"check_same_thread": False}, ) Base.metadata.create_all(engine) real = sessionmaker( bind=engine, expire_on_commit=False, autoflush=True, autocommit=False )() wrapper = _SafeSession(real) sf = lambda: wrapper # noqa: E731 repo = NamespacedProjectRepository(session_factory=sf) parsed = parse_namespaced_name("local/robot-bug1079") proj = NamespacedProject(name=parsed.name, namespace=parsed.namespace) repo.create(proj) return engine, sf, repo def _run_context_set( sf: Any, repo: NamespacedProjectRepository, cli_args: list[str] ) -> tuple[int, str]: """Run ``project context set`` via CliRunner, return (exit_code, output). Uses the CLI interface so that unknown flags (like the missing ``--execution-env-priority``) produce a non-zero exit code and error output rather than a TypeError. """ mc = MagicMock() mc.namespaced_project_repo.return_value = repo mc.session_factory.return_value = sf mc.context_tier_service.return_value = ContextTierService() with patch( "cleveragents.application.container.get_container", return_value=mc, ): result = runner.invoke(project_context_app, cli_args) return result.exit_code, result.output def _load_raw_blob(sf: Any, ns: str) -> dict[str, Any] | None: session = sf() row = session.execute( text("SELECT context_policy_json FROM ns_projects WHERE namespaced_name = :ns"), {"ns": ns}, ).fetchone() if row is None or row[0] is None: return None return _json.loads(row[0]) # ------------------------------------------------------------------ # Subcommands # ------------------------------------------------------------------ def set_env_priority_override() -> None: """Verify project context set accepts --execution-env-priority override.""" engine, sf, repo = _setup_db() try: exit_code, output = _run_context_set( sf, repo, [ "set", "local/robot-bug1079", "--execution-environment", "host", "--execution-env-priority", "override", ], ) if exit_code == 0: blob = _load_raw_blob(sf, "local/robot-bug1079") if blob and blob.get("execution_env_priority") == "override": print("project-context-set-env-priority-override-ok") else: print( f"FAIL: blob={blob!r}", file=sys.stderr, ) sys.exit(1) else: print( f"FAIL: exit code {exit_code}, output: {output}", file=sys.stderr, ) sys.exit(1) finally: engine.dispose() def set_env_priority_fallback() -> None: """Verify project context set accepts --execution-env-priority fallback.""" engine, sf, repo = _setup_db() try: exit_code, output = _run_context_set( sf, repo, [ "set", "local/robot-bug1079", "--execution-environment", "host", "--execution-env-priority", "fallback", ], ) if exit_code == 0: blob = _load_raw_blob(sf, "local/robot-bug1079") if blob and blob.get("execution_env_priority") == "fallback": print("project-context-set-env-priority-fallback-ok") else: print( f"FAIL: blob={blob!r}", file=sys.stderr, ) sys.exit(1) else: print( f"FAIL: exit code {exit_code}, output: {output}", file=sys.stderr, ) sys.exit(1) finally: engine.dispose() def set_env_priority_persists() -> None: """Verify execution_env_priority is persisted and retrievable.""" engine, sf, repo = _setup_db() try: exit_code, output = _run_context_set( sf, repo, [ "set", "local/robot-bug1079", "--execution-environment", "host", "--execution-env-priority", "override", ], ) if exit_code != 0: print( f"FAIL: set exit code {exit_code}, output: {output}", file=sys.stderr, ) sys.exit(1) blob = _load_raw_blob(sf, "local/robot-bug1079") if blob is None: print("FAIL: no blob stored", file=sys.stderr) sys.exit(1) got = blob.get("execution_env_priority") if got != "override": print( f"FAIL: expected 'override', got {got!r}", file=sys.stderr, ) sys.exit(1) if blob.get("execution_environment") != "host": print( f"FAIL: expected 'host', got {blob.get('execution_environment')!r}", file=sys.stderr, ) sys.exit(1) print("project-context-set-env-priority-persists-ok") finally: engine.dispose() # ------------------------------------------------------------------ # Main dispatcher # ------------------------------------------------------------------ _COMMANDS = { "set-env-priority-override": set_env_priority_override, "set-env-priority-fallback": set_env_priority_fallback, "set-env-priority-persists": set_env_priority_persists, } def main() -> None: if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS: print( f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr, ) sys.exit(2) _COMMANDS[sys.argv[1]]() if __name__ == "__main__": main()