HAL9000
565263ab24
Add ContextTierService mock to step_m5_invoke_project_context_show so the CLI command can call _get_context_tier_service() without failing. Also patch _load_policy_json to prevent JSON decode errors from MagicMock session objects. Update CHANGELOG.md and CONTRIBUTORS.md. ISSUES CLOSED: #6323
419 lines
13 KiB
Python
419 lines
13 KiB
Python
"""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 _capturing_format_output(data: Any, format_type: str) -> str:
|
|
"""Wrapper around ``format_output`` that captures stdout writes.
|
|
|
|
``format_output`` for machine-readable formats (json, yaml, plain)
|
|
writes directly to ``sys.stdout`` and returns ``""``. This wrapper
|
|
redirects ``sys.stdout`` only during the ``format_output`` call so
|
|
that the rendered content is returned as a string, allowing the
|
|
caller's ``console.print(result)`` to write it to the test's Rich
|
|
console buffer. Structlog noise that would otherwise pollute the
|
|
captured stdout is excluded because the redirect is scoped tightly.
|
|
"""
|
|
import sys as _sys
|
|
|
|
from cleveragents.cli.formatting import format_output as _real_format_output
|
|
|
|
buf = StringIO()
|
|
old = _sys.stdout
|
|
_sys.stdout = buf
|
|
try:
|
|
ret = _real_format_output(data, format_type)
|
|
finally:
|
|
_sys.stdout = old
|
|
return ret or buf.getvalue().rstrip("\n")
|
|
|
|
|
|
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,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.project_context.format_output",
|
|
_capturing_format_output,
|
|
),
|
|
):
|
|
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}")
|
|
|
|
envelope = json.loads(output.strip())
|
|
if envelope.get("command") != "project context show":
|
|
raise AssertionError(f"Unexpected command value: {envelope.get('command')}")
|
|
|
|
payload = envelope.get("data")
|
|
if not isinstance(payload, dict):
|
|
raise AssertionError(f"Expected data payload dict, got: {type(payload)!r}")
|
|
|
|
context_policy = payload.get("context_policy")
|
|
if not isinstance(context_policy, dict):
|
|
raise AssertionError(f"Missing context_policy in payload: {payload.keys()}")
|
|
if context_policy.get("project") != "local/ctx-robot":
|
|
raise AssertionError(
|
|
f"context_policy project mismatch: {context_policy.get('project')}"
|
|
)
|
|
|
|
limits = payload.get("limits")
|
|
if not isinstance(limits, dict):
|
|
raise AssertionError(f"Missing limits in payload: {payload.keys()}")
|
|
if limits.get("hot_max_tokens") != 5000:
|
|
raise AssertionError(f"hot_max_tokens mismatch: {limits.get('hot_max_tokens')}")
|
|
|
|
if "summarization" not in payload:
|
|
raise AssertionError(
|
|
f"Missing summarization section in payload: {payload.keys()}"
|
|
)
|
|
if "current_usage" not in payload:
|
|
raise AssertionError(
|
|
f"Missing current_usage section in payload: {payload.keys()}"
|
|
)
|
|
|
|
messages = envelope.get("messages")
|
|
if messages != ["Context policy loaded"]:
|
|
raise AssertionError(f"Unexpected messages payload: {messages}")
|
|
|
|
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)
|