cleveragents-core/robot/helper_project_context_cli.py

"""Helper script for project context CLI Robot Framework tests.

Usage: python robot/helper_project_context_cli.py <test-name>
"""

from __future__ import annotations

import json
import sys
from io import StringIO
from typing import Any
from unittest.mock import MagicMock, patch

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from cleveragents.application.services.context_tiers import ContextTierService
from cleveragents.domain.models.acms.tiers import (
    ContextTier,
    TieredFragment,
)
from cleveragents.domain.models.core.context_policy import (
    ContextView,
    ProjectContextPolicy,
)
from cleveragents.domain.models.core.project import (
    NamespacedProject,
    parse_namespaced_name,
)
from cleveragents.infrastructure.database.models import Base
from cleveragents.infrastructure.database.repositories import (
    NamespacedProjectRepository,
)


class _NoClose:
    """Session wrapper that no-ops close()."""

    def __init__(self, s: object) -> None:
        object.__setattr__(self, "_s", s)

    def close(self) -> None:
        pass

    def __getattr__(self, n: str) -> object:
        return getattr(object.__getattribute__(self, "_s"), n)


def _setup_db() -> tuple[NamespacedProjectRepository, Any]:
    """Create in-memory SQLite DB, return (repo, session_factory)."""
    engine = create_engine("sqlite:///:memory:", echo=False)
    Base.metadata.create_all(engine)
    session = sessionmaker(bind=engine, expire_on_commit=False)()
    wrapper = _NoClose(session)

    def factory() -> object:
        return wrapper

    repo = NamespacedProjectRepository(session_factory=factory)
    return repo, factory


def _create_project(
    repo: NamespacedProjectRepository,
    name: str,
) -> None:
    parsed = parse_namespaced_name(name)
    proj = NamespacedProject(
        name=parsed.name,
        namespace=parsed.namespace,
    )
    repo.create(proj)


def _make_mock_container(
    repo: NamespacedProjectRepository,
    session_factory: Any,
    tier_service: ContextTierService | None = None,
) -> MagicMock:
    """Build a mock DI container wired to the test fixtures."""
    mock_container = MagicMock()
    mock_container.namespaced_project_repo.return_value = repo
    mock_container.session_factory.return_value = session_factory
    if tier_service is None:
        tier_service = ContextTierService()
    mock_container.context_tier_service.return_value = tier_service
    return mock_container


def _run_cli_func(
    func: Any,
    repo: NamespacedProjectRepository,
    session_factory: Any,
    tier_service: ContextTierService | None = None,
    **kwargs: Any,
) -> tuple[int, str]:
    """Execute a CLI function with mocked DI and capture output."""
    import typer
    from rich.console import Console as RichConsole

    mock_cont = _make_mock_container(repo, session_factory, tier_service)
    buf = StringIO()
    test_console = RichConsole(
        file=buf,
        no_color=True,
        highlight=False,
        force_terminal=False,
        width=500,
        soft_wrap=True,
    )

    exit_code = 0
    with (
        patch(
            "cleveragents.application.container.get_container",
            return_value=mock_cont,
        ),
        patch(
            "cleveragents.cli.commands.project_context.console",
            test_console,
        ),
        patch(
            "cleveragents.cli.commands.project_context.err_console",
            test_console,
        ),
    ):
        try:
            func(**kwargs)
        except typer.Exit as exc:
            exit_code = exc.exit_code
        except typer.Abort:
            exit_code = 1

    return exit_code, buf.getvalue()


def test_context_set_show() -> None:
    """Set a context policy and then read it back."""
    from cleveragents.cli.commands.project_context import (
        _read_policy,
        _write_policy,
    )

    repo, sf = _setup_db()
    _create_project(repo, "ctx-robot")

    # Write a policy
    policy = ProjectContextPolicy(
        default_view=ContextView(
            include_resources=["db-*"],
            exclude_paths=["*.pyc"],
            max_file_size=1048576,
        ),
        strategize_view=ContextView(
            include_resources=["cache-*"],
        ),
    )
    _write_policy(sf, "local/ctx-robot", policy)

    # Read it back
    loaded = _read_policy(sf, "local/ctx-robot")
    if loaded.default_view.include_resources != ["db-*"]:
        raise AssertionError(
            f"include_resources mismatch: {loaded.default_view.include_resources}"
        )
    if loaded.strategize_view is None:
        raise AssertionError("strategize_view is None")
    if loaded.strategize_view.include_resources != ["cache-*"]:
        raise AssertionError("strategize include_resources mismatch")

    # Test inheritance
    resolved = loaded.resolve_view("execute")
    if resolved.include_resources != ["cache-*"]:
        raise AssertionError(
            f"execute should inherit from strategize, got {resolved.include_resources}"
        )

    print("context-set-show-ok")


def test_context_inspect_wired() -> None:
    """Verify inspect is wired to ACMS and succeeds."""
    from cleveragents.cli.commands.project_context import context_inspect

    repo, sf = _setup_db()
    _create_project(repo, "ctx-robot")

    tier_svc = ContextTierService()
    # Pre-populate a fragment
    tier_svc.store(
        TieredFragment(
            fragment_id="robot-f1",
            content="test fragment",
            tier=ContextTier.HOT,
            project_name="local/ctx-robot",
            token_count=50,
        )
    )

    exit_code, output = _run_cli_func(
        context_inspect,
        repo,
        sf,
        tier_service=tier_svc,
        project="local/ctx-robot",
        output_format="json",
    )
    if exit_code != 0:
        raise AssertionError(f"inspect failed with exit {exit_code}: {output}")

    data = json.loads(output.strip())
    if "tier_metrics" not in data:
        raise AssertionError(f"Missing tier_metrics in output: {data.keys()}")
    if "project_fragments" not in data:
        raise AssertionError(f"Missing project_fragments in output: {data.keys()}")
    if data["project_fragments"]["total"] != 1:
        raise AssertionError(
            f"Expected 1 fragment, got {data['project_fragments']['total']}"
        )

    print("context-inspect-wired-ok")


def test_context_simulate_wired() -> None:
    """Verify simulate is wired to ACMS and succeeds."""
    from cleveragents.cli.commands.project_context import context_simulate

    repo, sf = _setup_db()
    _create_project(repo, "ctx-robot")

    tier_svc = ContextTierService()
    # Pre-populate a fragment for simulation
    tier_svc.store(
        TieredFragment(
            fragment_id="robot-sim1",
            content="simulation test fragment",
            tier=ContextTier.HOT,
            project_name="local/ctx-robot",
            token_count=100,
        )
    )

    exit_code, output = _run_cli_func(
        context_simulate,
        repo,
        sf,
        tier_service=tier_svc,
        project="local/ctx-robot",
        output_format="json",
    )
    if exit_code != 0:
        raise AssertionError(f"simulate failed with exit {exit_code}: {output}")

    data = json.loads(output.strip())
    if "total_tokens" not in data:
        raise AssertionError(f"Missing total_tokens in output: {data.keys()}")
    if data["total_tokens"] <= 0:
        raise AssertionError(
            f"Expected positive total_tokens, got {data['total_tokens']}"
        )
    if "strategies_used" not in data:
        raise AssertionError(f"Missing strategies_used in output: {data.keys()}")

    print("context-simulate-wired-ok")


def test_context_set_acms_options() -> None:
    """Verify context set persists ACMS pipeline config."""
    from cleveragents.cli.commands.project_context import (
        _read_acms_config,
        context_set,
    )

    repo, sf = _setup_db()
    _create_project(repo, "ctx-robot")

    exit_code, output = _run_cli_func(
        context_set,
        repo,
        sf,
        project="local/ctx-robot",
        hot_max_tokens=4000,
        warm_max_decisions=200,
        temporal_scope="recent",
        auto_refresh=False,
    )
    if exit_code != 0:
        raise AssertionError(f"set failed with exit {exit_code}: {output}")

    acms = _read_acms_config(sf, "local/ctx-robot")
    if acms["hot_max_tokens"] != 4000:
        raise AssertionError(f"hot_max_tokens mismatch: {acms['hot_max_tokens']}")
    if acms["warm_max_decisions"] != 200:
        raise AssertionError(
            f"warm_max_decisions mismatch: {acms['warm_max_decisions']}"
        )
    if acms["temporal_scope"] != "recent":
        raise AssertionError(f"temporal_scope mismatch: {acms['temporal_scope']}")
    if acms["auto_refresh"] is not False:
        raise AssertionError(f"auto_refresh mismatch: {acms['auto_refresh']}")

    print("context-set-acms-ok")


def test_context_show_acms_config() -> None:
    """Verify context show includes ACMS config in JSON output."""
    from cleveragents.cli.commands.project_context import context_set, context_show

    repo, sf = _setup_db()
    _create_project(repo, "ctx-robot")

    # First set some ACMS config
    _run_cli_func(
        context_set,
        repo,
        sf,
        project="local/ctx-robot",
        hot_max_tokens=5000,
    )

    exit_code, output = _run_cli_func(
        context_show,
        repo,
        sf,
        project="local/ctx-robot",
        output_format="json",
    )
    if exit_code != 0:
        raise AssertionError(f"show failed with exit {exit_code}: {output}")

    data = json.loads(output.strip())
    if "acms_config" not in data:
        raise AssertionError(f"Missing acms_config in output: {data.keys()}")
    if data["acms_config"]["hot_max_tokens"] != 5000:
        raise AssertionError(
            f"hot_max_tokens mismatch: {data['acms_config']['hot_max_tokens']}"
        )

    print("context-show-acms-ok")


TESTS: dict[str, Any] = {
    "set-show": test_context_set_show,
    "inspect-wired": test_context_inspect_wired,
    "simulate-wired": test_context_simulate_wired,
    "set-acms": test_context_set_acms_options,
    "show-acms": test_context_show_acms_config,
}


if __name__ == "__main__":
    if len(sys.argv) < 2 or sys.argv[1] not in TESTS:
        print(
            f"Usage: {sys.argv[0]} <{'|'.join(TESTS)}>",
            file=sys.stderr,
        )
        sys.exit(2)
    try:
        TESTS[sys.argv[1]]()
    except Exception as exc:
        print(f"FAIL: {exc}", file=sys.stderr)
        sys.exit(1)