placeholder/robot/helper_project_context_set_exec_env_priority.py

"""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()