cleveragents-core/robot/helper_acms_context_cli.py

"""Robot Framework helper for ACMS context CLI integration tests.

Provides a CLI-style interface for Robot to invoke ACMS context CLI
operations and verify the results. Exit code 0 = success, 1 = failure.

Usage:
    python robot/helper_acms_context_cli.py context-show
    python robot/helper_acms_context_cli.py context-show-empty
    python robot/helper_acms_context_cli.py context-clear-yes
    python robot/helper_acms_context_cli.py context-clear-path
    python robot/helper_acms_context_cli.py context-clear-tag
    python robot/helper_acms_context_cli.py context-clear-tier
    python robot/helper_acms_context_cli.py context-clear-no-match
    python robot/helper_acms_context_cli.py context-show-empty-view
    python robot/helper_acms_context_cli.py budget-format
"""

from __future__ import annotations

import sys
from collections.abc import Callable
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock, patch

_SRC = str(Path(__file__).resolve().parents[1] / "src")
# Always insert at position 0 to ensure the clone's source takes precedence
# over any other cleveragents installation (e.g., /app/src from the environment)
if _SRC in sys.path:
    sys.path.remove(_SRC)
sys.path.insert(0, _SRC)

_FEATURES = str(Path(__file__).resolve().parents[1])
if _FEATURES in sys.path:
    sys.path.remove(_FEATURES)
sys.path.insert(0, _FEATURES)

import typer  # noqa: E402
from features.mocks.acms_context_mocks import (  # noqa: E402
    make_mock_container,
    make_mock_tier_service,
    make_tiered_fragment,
)

from cleveragents.cli.commands.acms_context import (  # noqa: E402
    _format_budget_utilization,
    acms_context_clear,
    acms_context_show,
)
from cleveragents.domain.models.acms.tiers import ContextTier  # noqa: E402

_SAMPLE_FRAGMENTS = [
    make_tiered_fragment(
        fragment_id=f"frag-{i}",
        resource_id=f"src/module{i}.py",
        tier=ContextTier.HOT if i % 2 == 0 else ContextTier.WARM,
        token_count=1024 * (i + 1),
        project_name="test_project",
        tag="deprecated" if i < 2 else "default",
    )
    for i in range(5)
]


def _run_show(
    view: str,
    scoped_fragments: list[Any],
) -> tuple[str, int]:
    """Run acms_context_show and capture output."""
    mock_service = make_mock_tier_service(
        fragments=scoped_fragments,
        scoped_fragments=scoped_fragments,
    )
    mock_container = make_mock_container(mock_service)
    output_lines: list[str] = []
    mock_console = MagicMock()

    def capture_print(msg: str = "", **kwargs: Any) -> None:
        output_lines.append(str(msg))

    mock_console.print.side_effect = capture_print

    with (
        patch(
            "cleveragents.cli.commands.acms_context.get_container",
            return_value=mock_container,
        ),
        patch(
            "cleveragents.cli.commands.acms_context._get_console",
            return_value=mock_console,
        ),
    ):
        exit_code = 0
        try:
            acms_context_show(view)
        except typer.Exit as exc:
            exit_code = exc.exit_code if exc.exit_code is not None else 0
        except SystemExit as exc:
            exit_code = exc.code if exc.code is not None else 0

    return "\n".join(output_lines), exit_code


def _run_clear(
    all_fragments: list[Any],
    path: str | None = None,
    tag: str | None = None,
    tier: str | None = None,
    yes: bool = True,
) -> tuple[str, int]:
    """Run acms_context_clear and capture output."""
    mock_service = make_mock_tier_service(fragments=all_fragments)
    mock_container = make_mock_container(mock_service)
    output_lines: list[str] = []
    mock_console = MagicMock()

    def capture_print(msg: str = "", **kwargs: Any) -> None:
        output_lines.append(str(msg))

    mock_console.print.side_effect = capture_print

    with (
        patch(
            "cleveragents.cli.commands.acms_context.get_container",
            return_value=mock_container,
        ),
        patch(
            "cleveragents.cli.commands.acms_context._get_console",
            return_value=mock_console,
        ),
    ):
        exit_code = 0
        try:
            acms_context_clear(path=path, tag=tag, tier=tier, yes=yes)
        except typer.Exit as exc:
            exit_code = exc.exit_code if exc.exit_code is not None else 0
        except SystemExit as exc:
            exit_code = exc.code if exc.code is not None else 0

    return "\n".join(output_lines), exit_code


def _cmd_context_show() -> int:
    """Verify context show command returns assembled context data."""
    output, exit_code = _run_show("test_project", _SAMPLE_FRAGMENTS)
    if exit_code != 0:
        print(f"acms-fail: context show exited with code {exit_code}")
        return 1
    expected = "Assembled Context for View: test_project"
    if expected not in output:
        print(f"acms-fail: expected {expected!r} in output\n{output}")
        return 1
    # Verify per-tier budget utilization breakdown is present
    for tier_label in ("Hot tier", "Warm tier", "Cold tier"):
        if tier_label not in output:
            print(
                f"acms-fail: expected per-tier breakdown"
                f" '{tier_label}' in output\n{output}"
            )
            return 1
    print("acms-context-show-ok")
    return 0


def _cmd_context_show_empty() -> int:
    """Verify context show command handles view with no entries."""
    output, exit_code = _run_show("empty_view", [])
    if exit_code != 0:
        print(f"acms-fail: context show empty exited with code {exit_code}")
        return 1
    if "No context found for view: empty_view" not in output:
        print(f"acms-fail: expected 'No context found for view' in output\n{output}")
        return 1
    print("acms-context-show-empty-ok")
    return 0


def _cmd_context_clear_yes() -> int:
    """Verify context clear command removes entries when --yes flag is used."""
    output, exit_code = _run_clear(
        all_fragments=_SAMPLE_FRAGMENTS,
        yes=True,
    )
    if exit_code != 0:
        print(f"acms-fail: context clear --yes exited with code {exit_code}")
        return 1
    if "Removed" not in output or "context entries" not in output:
        print(f"acms-fail: expected 'Removed N context entries' in output\n{output}")
        return 1
    print("acms-context-clear-yes-ok")
    return 0


def _cmd_context_clear_path() -> int:
    """Verify context clear command filters entries by path pattern."""
    output, exit_code = _run_clear(
        all_fragments=_SAMPLE_FRAGMENTS,
        path="src/module0.py",
        yes=True,
    )
    if exit_code != 0:
        print(f"acms-fail: context clear --path exited with code {exit_code}")
        return 1
    if "context entries" not in output:
        print(f"acms-fail: expected 'context entries' in output\n{output}")
        return 1
    print("acms-context-clear-path-ok")
    return 0


def _cmd_context_clear_tag() -> int:
    """Verify context clear command filters entries by tag."""
    output, exit_code = _run_clear(
        all_fragments=_SAMPLE_FRAGMENTS,
        tag="deprecated",
        yes=True,
    )
    if exit_code != 0:
        print(f"acms-fail: context clear --tag exited with code {exit_code}")
        return 1
    if "context entries" not in output:
        print(f"acms-fail: expected 'context entries' in output\n{output}")
        return 1
    print("acms-context-clear-tag-ok")
    return 0


def _cmd_context_clear_tier() -> int:
    """Verify context clear command filters entries by tier."""
    output, exit_code = _run_clear(
        all_fragments=_SAMPLE_FRAGMENTS,
        tier="hot",
        yes=True,
    )
    if exit_code != 0:
        print(f"acms-fail: context clear --tier exited with code {exit_code}")
        return 1
    if "context entries" not in output:
        print(f"acms-fail: expected 'context entries' in output\n{output}")
        return 1
    print("acms-context-clear-tier-ok")
    return 0


def _cmd_context_clear_no_match() -> int:
    """Verify context clear command handles case with no matching entries."""
    output, exit_code = _run_clear(
        all_fragments=_SAMPLE_FRAGMENTS,
        path="nonexistent/*",
        yes=True,
    )
    if exit_code != 0:
        print(f"acms-fail: context clear no-match exited with code {exit_code}")
        return 1
    if "No context entries match the specified filters" not in output:
        print(f"acms-fail: expected 'No context entries match' in output\n{output}")
        return 1
    print("acms-context-clear-no-match-ok")
    return 0


def _cmd_context_show_empty_view() -> int:
    """Verify context show command rejects empty view name."""
    _output, exit_code = _run_show("", [])
    if exit_code == 0:
        print("acms-fail: expected non-zero exit code for empty view name")
        return 1
    print("acms-context-show-empty-view-ok")
    return 0


def _cmd_budget_format() -> int:
    """Verify budget utilization formatting helper."""
    cases = [
        (0, 0, "0%"),
        (500, 1000, "50.0%"),
        (1000, 1000, "100.0%"),
        (-1, 100, "N/A"),
        (100, -1, "N/A"),
    ]
    for used, total, expected in cases:
        result = _format_budget_utilization(used, total)
        if result != expected:
            print(
                f"acms-fail: _format_budget_utilization({used}, {total}) "
                f"= {result!r}, expected {expected!r}"
            )
            return 1
    print("acms-budget-format-ok")
    return 0


_COMMANDS: dict[str, Callable[[], int]] = {
    "context-show": _cmd_context_show,
    "context-show-empty": _cmd_context_show_empty,
    "context-clear-yes": _cmd_context_clear_yes,
    "context-clear-path": _cmd_context_clear_path,
    "context-clear-tag": _cmd_context_clear_tag,
    "context-clear-tier": _cmd_context_clear_tier,
    "context-clear-no-match": _cmd_context_clear_no_match,
    "context-show-empty-view": _cmd_context_show_empty_view,
    "budget-format": _cmd_budget_format,
}


def main() -> int:
    """Entry point called by Robot Framework ``Run Process``."""
    if len(sys.argv) < 2:
        print(
            "Usage: helper_acms_context_cli.py "
            "<context-show|context-show-empty|context-clear-yes"
            "|context-clear-path|context-clear-tag|context-clear-tier"
            "|context-clear-no-match|context-show-empty-view|budget-format>"
        )
        return 1

    command = sys.argv[1]
    handler = _COMMANDS.get(command)
    if handler is None:
        print(f"Unknown command: {command}")
        return 1

    return handler()


if __name__ == "__main__":
    sys.exit(main())