HAL9000
6f2b3ec8b4
CI / push-validation (pull_request) Successful in 20s
CI / helm (pull_request) Successful in 23s
CI / build (pull_request) Successful in 26s
CI / lint (pull_request) Failing after 28s
CI / quality (pull_request) Successful in 42s
CI / e2e_tests (pull_request) Successful in 2m59s
CI / typecheck (pull_request) Successful in 4m0s
CI / integration_tests (pull_request) Successful in 4m0s
CI / security (pull_request) Successful in 4m4s
CI / coverage (pull_request) Has been skipped
CI / unit_tests (pull_request) Failing after 5m41s
CI / docker (pull_request) Has been skipped
CI / status-check (pull_request) Failing after 1s
CI / benchmark-publish (pull_request) Has been skipped
CI / benchmark-regression (pull_request) Has been skipped
ISSUES CLOSED: #6323
800 lines
24 KiB
Python
800 lines
24 KiB
Python
"""Step definitions for project_context_cli.feature.
|
|
|
|
Tests the ``agents project context set/show/inspect/simulate``
|
|
CLI commands using an in-memory SQLite database.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from contextlib import suppress
|
|
from io import StringIO
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import yaml
|
|
from behave import given, then, when # type: ignore[import-untyped]
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
# ------------------------------------------------------------------
|
|
# Session wrapper (no-op close)
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
class _UnclosableSession:
|
|
"""Wraps a SQLAlchemy Session; ``close()`` is a no-op."""
|
|
|
|
def __init__(self, real_session: Session) -> None:
|
|
object.__setattr__(self, "_real", real_session)
|
|
|
|
def close(self) -> None:
|
|
"""No-op."""
|
|
|
|
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 _make_ctx_session_factory(
|
|
context: Any,
|
|
) -> tuple[Any, Any]:
|
|
"""In-memory SQLite with all tables."""
|
|
from cleveragents.infrastructure.database.models import Base
|
|
|
|
engine = create_engine(
|
|
"sqlite:///:memory:",
|
|
echo=False,
|
|
connect_args={"check_same_thread": False},
|
|
)
|
|
Base.metadata.create_all(engine)
|
|
real_session = sessionmaker(
|
|
bind=engine,
|
|
expire_on_commit=False,
|
|
autoflush=True,
|
|
autocommit=False,
|
|
)()
|
|
wrapper = _UnclosableSession(real_session)
|
|
|
|
def _factory() -> Any:
|
|
return wrapper
|
|
|
|
return engine, _factory
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Background
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@given("a project context CLI in-memory database is initialized")
|
|
def step_init_ctx_cli_db(context: Any) -> None:
|
|
from cleveragents.infrastructure.database.repositories import (
|
|
NamespacedProjectRepository,
|
|
)
|
|
|
|
engine, session_factory = _make_ctx_session_factory(context)
|
|
context.ctx_engine = engine
|
|
context.ctx_session_factory = session_factory
|
|
context.ctx_project_repo = NamespacedProjectRepository(
|
|
session_factory=session_factory
|
|
)
|
|
context.ctx_output = ""
|
|
context.ctx_exit_code = 0
|
|
context.ctx_error = None
|
|
|
|
|
|
@given('a project "{name}" exists for context CLI')
|
|
def step_create_project_for_ctx(context: Any, name: str) -> None:
|
|
from cleveragents.domain.models.core.project import (
|
|
NamespacedProject,
|
|
parse_namespaced_name,
|
|
)
|
|
|
|
parsed = parse_namespaced_name(name)
|
|
proj = NamespacedProject(
|
|
name=parsed.name,
|
|
namespace=parsed.namespace,
|
|
)
|
|
context.ctx_project_repo.create(proj)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
def _mock_container(context: Any) -> MagicMock:
|
|
"""Build a mock container wired to the test DB."""
|
|
from cleveragents.application.services.context_tiers import (
|
|
ContextTierService,
|
|
)
|
|
|
|
mock_container = MagicMock()
|
|
mock_container.namespaced_project_repo.return_value = context.ctx_project_repo
|
|
mock_container.session_factory.return_value = context.ctx_session_factory
|
|
# Provide a real (in-memory) ContextTierService for inspect/simulate
|
|
if not hasattr(context, "ctx_tier_service"):
|
|
context.ctx_tier_service = ContextTierService()
|
|
mock_container.context_tier_service.return_value = context.ctx_tier_service
|
|
return mock_container
|
|
|
|
|
|
def _run_with_container(context: Any, func: Any, *args: Any, **kwargs: Any) -> None:
|
|
"""Execute a CLI function with a mocked DI container."""
|
|
import typer
|
|
from rich.console import Console as RichConsole
|
|
|
|
mock_cont = _mock_container(context)
|
|
buf = StringIO()
|
|
# force_terminal=False and highlight=False prevent Rich from emitting
|
|
# ANSI escape codes when FORCE_COLOR is set or a real colour terminal
|
|
# is detected.
|
|
test_console = RichConsole(
|
|
file=buf,
|
|
no_color=True,
|
|
highlight=False,
|
|
force_terminal=False,
|
|
width=500,
|
|
soft_wrap=True,
|
|
)
|
|
|
|
def _test_format_output(data, format_type, **kwargs):
|
|
import sys as _sys
|
|
from io import StringIO as _SIO
|
|
|
|
from cleveragents.cli.formatting import format_output as _fo
|
|
|
|
_b = _SIO()
|
|
_old = _sys.stdout
|
|
_sys.stdout = _b
|
|
try:
|
|
_r = _fo(data, format_type, **kwargs)
|
|
finally:
|
|
_sys.stdout = _old
|
|
return _r or _b.getvalue().rstrip("\n")
|
|
|
|
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",
|
|
_test_format_output,
|
|
),
|
|
):
|
|
try:
|
|
func(*args, **kwargs)
|
|
context.ctx_exit_code = 0
|
|
except typer.Exit as exc:
|
|
context.ctx_exit_code = exc.exit_code
|
|
except typer.Abort:
|
|
context.ctx_exit_code = 1
|
|
except NotImplementedError as exc:
|
|
context.ctx_error = exc
|
|
context.ctx_exit_code = 99
|
|
|
|
context.ctx_output = buf.getvalue()
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# context set
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@when(
|
|
'I run context set on "{project}" with view "{view}" and include-resource "{res}"'
|
|
)
|
|
def step_ctx_set_include_res(context: Any, project: str, view: str, res: str) -> None:
|
|
from cleveragents.cli.commands.project_context import (
|
|
context_set,
|
|
)
|
|
|
|
_run_with_container(
|
|
context,
|
|
context_set,
|
|
project=project,
|
|
view=view,
|
|
include_resource=[res],
|
|
)
|
|
|
|
|
|
@when(
|
|
'I run context set on "{project}" with view "{view}" and exclude-resource "{res}"'
|
|
)
|
|
def step_ctx_set_exclude_res(context: Any, project: str, view: str, res: str) -> None:
|
|
from cleveragents.cli.commands.project_context import (
|
|
context_set,
|
|
)
|
|
|
|
_run_with_container(
|
|
context,
|
|
context_set,
|
|
project=project,
|
|
view=view,
|
|
exclude_resource=[res],
|
|
)
|
|
|
|
|
|
@when('I run context set on "{project}" with view "{view}" and include-path "{path}"')
|
|
def step_ctx_set_include_path(context: Any, project: str, view: str, path: str) -> None:
|
|
from cleveragents.cli.commands.project_context import (
|
|
context_set,
|
|
)
|
|
|
|
_run_with_container(
|
|
context,
|
|
context_set,
|
|
project=project,
|
|
view=view,
|
|
include_path=[path],
|
|
)
|
|
|
|
|
|
@when('I run context set on "{project}" with view "{view}" and exclude-path "{path}"')
|
|
def step_ctx_set_exclude_path(context: Any, project: str, view: str, path: str) -> None:
|
|
from cleveragents.cli.commands.project_context import (
|
|
context_set,
|
|
)
|
|
|
|
_run_with_container(
|
|
context,
|
|
context_set,
|
|
project=project,
|
|
view=view,
|
|
exclude_path=[path],
|
|
)
|
|
|
|
|
|
@when('I run context set on "{project}" with view "{view}" and max-file-size {size:d}')
|
|
def step_ctx_set_max_file_size(
|
|
context: Any, project: str, view: str, size: int
|
|
) -> None:
|
|
from cleveragents.cli.commands.project_context import (
|
|
context_set,
|
|
)
|
|
|
|
_run_with_container(
|
|
context,
|
|
context_set,
|
|
project=project,
|
|
view=view,
|
|
max_file_size=size,
|
|
)
|
|
|
|
|
|
@when('I run context set on "{project}" with view "{view}" and max-total-size {size:d}')
|
|
def step_ctx_set_max_total_size(
|
|
context: Any, project: str, view: str, size: int
|
|
) -> None:
|
|
from cleveragents.cli.commands.project_context import (
|
|
context_set,
|
|
)
|
|
|
|
_run_with_container(
|
|
context,
|
|
context_set,
|
|
project=project,
|
|
view=view,
|
|
max_total_size=size,
|
|
)
|
|
|
|
|
|
@when('I run context set on "{project}" with invalid view "{view}"')
|
|
def step_ctx_set_invalid_view(context: Any, project: str, view: str) -> None:
|
|
from cleveragents.cli.commands.project_context import (
|
|
context_set,
|
|
)
|
|
|
|
_run_with_container(
|
|
context,
|
|
context_set,
|
|
project=project,
|
|
view=view,
|
|
)
|
|
|
|
|
|
@when('I run context set on "{project}" with view "{view}" and clear flag')
|
|
def step_ctx_set_clear(context: Any, project: str, view: str) -> None:
|
|
from cleveragents.cli.commands.project_context import (
|
|
context_set,
|
|
)
|
|
|
|
_run_with_container(
|
|
context,
|
|
context_set,
|
|
project=project,
|
|
view=view,
|
|
clear=True,
|
|
)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# context show
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@when('I run context show on "{project}" without options')
|
|
def step_ctx_show(context: Any, project: str) -> None:
|
|
from cleveragents.cli.commands.project_context import (
|
|
context_show,
|
|
)
|
|
|
|
kwargs: dict[str, Any] = {}
|
|
fmt_override = getattr(context, "ctx_show_format", None)
|
|
if fmt_override is not None:
|
|
kwargs["output_format"] = fmt_override
|
|
del context.ctx_show_format
|
|
|
|
_run_with_container(context, context_show, project=project, **kwargs)
|
|
|
|
|
|
@when('I run context show on "{project}" with view "{view}"')
|
|
def step_ctx_show_view(context: Any, project: str, view: str) -> None:
|
|
from cleveragents.cli.commands.project_context import (
|
|
context_show,
|
|
)
|
|
|
|
kwargs: dict[str, Any] = {}
|
|
fmt_override = getattr(context, "ctx_show_format", None)
|
|
if fmt_override is not None:
|
|
kwargs["output_format"] = fmt_override
|
|
del context.ctx_show_format
|
|
|
|
_run_with_container(context, context_show, project=project, view=view, **kwargs)
|
|
|
|
|
|
@when('I run context show on "{project}" with format "{fmt}"')
|
|
def step_ctx_show_format(context: Any, project: str, fmt: str) -> None:
|
|
from cleveragents.cli.commands.project_context import (
|
|
context_show,
|
|
)
|
|
|
|
_run_with_container(
|
|
context,
|
|
context_show,
|
|
project=project,
|
|
output_format=fmt,
|
|
)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# context inspect / simulate stubs
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@when('I run context inspect on "{project}"')
|
|
def step_ctx_inspect(context: Any, project: str) -> None:
|
|
from cleveragents.cli.commands.project_context import (
|
|
context_inspect,
|
|
)
|
|
|
|
_run_with_container(context, context_inspect, project=project)
|
|
|
|
|
|
@when('I run context simulate on "{project}"')
|
|
def step_ctx_simulate(context: Any, project: str) -> None:
|
|
from cleveragents.cli.commands.project_context import (
|
|
context_simulate,
|
|
)
|
|
|
|
_run_with_container(context, context_simulate, project=project)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Given steps (pre-conditions)
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@given('I have set a strategize view on "{project}" with defaults')
|
|
def step_given_strategize_view(context: Any, project: str) -> None:
|
|
from cleveragents.cli.commands.project_context import (
|
|
context_set,
|
|
)
|
|
|
|
_run_with_container(
|
|
context,
|
|
context_set,
|
|
project=project,
|
|
view="strategize",
|
|
include_resource=["strat-default"],
|
|
)
|
|
|
|
|
|
@given('I have set a default view on "{project}" with include-resource "{res}"')
|
|
def step_given_default_view(context: Any, project: str, res: str) -> None:
|
|
from cleveragents.cli.commands.project_context import (
|
|
context_set,
|
|
)
|
|
|
|
_run_with_container(
|
|
context,
|
|
context_set,
|
|
project=project,
|
|
view="default",
|
|
include_resource=[res],
|
|
)
|
|
|
|
|
|
@given('I have set a strategize view on "{project}" with include-resource "{res}"')
|
|
def step_given_strategize_view_res(context: Any, project: str, res: str) -> None:
|
|
from cleveragents.cli.commands.project_context import (
|
|
context_set,
|
|
)
|
|
|
|
_run_with_container(
|
|
context,
|
|
context_set,
|
|
project=project,
|
|
view="strategize",
|
|
include_resource=[res],
|
|
)
|
|
|
|
|
|
@given('I have set an execute view on "{project}" with include-resource "{res}"')
|
|
def step_given_execute_view_res(context: Any, project: str, res: str) -> None:
|
|
from cleveragents.cli.commands.project_context import (
|
|
context_set,
|
|
)
|
|
|
|
_run_with_container(
|
|
context,
|
|
context_set,
|
|
project=project,
|
|
view="execute",
|
|
include_resource=[res],
|
|
)
|
|
|
|
|
|
@given('the context show output format is "{fmt}"')
|
|
def step_set_ctx_show_format(context: Any, fmt: str) -> None:
|
|
context.ctx_show_format = fmt
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Then assertions
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@then("the context set command should succeed")
|
|
def step_ctx_set_ok(context: Any) -> None:
|
|
assert context.ctx_exit_code == 0, (
|
|
f"Expected exit 0, got {context.ctx_exit_code}. Output: {context.ctx_output}"
|
|
)
|
|
|
|
|
|
@then("the context set command should fail")
|
|
def step_ctx_set_fail(context: Any) -> None:
|
|
assert context.ctx_exit_code != 0, (
|
|
f"Expected non-zero exit, got {context.ctx_exit_code}"
|
|
)
|
|
|
|
|
|
@then("the context show command should succeed")
|
|
def step_ctx_show_ok(context: Any) -> None:
|
|
assert context.ctx_exit_code == 0, (
|
|
f"Expected exit 0, got {context.ctx_exit_code}. Output: {context.ctx_output}"
|
|
)
|
|
|
|
|
|
@then("the context show command should fail")
|
|
def step_ctx_show_fail(context: Any) -> None:
|
|
assert context.ctx_exit_code != 0, (
|
|
f"Expected non-zero exit, got {context.ctx_exit_code}"
|
|
)
|
|
|
|
|
|
def _read_stored_policy(context: Any) -> Any:
|
|
"""Read the stored policy from the test DB."""
|
|
from cleveragents.cli.commands.project_context import (
|
|
_read_policy,
|
|
)
|
|
|
|
return _read_policy(context.ctx_session_factory, "local/ctx-app")
|
|
|
|
|
|
@then('the stored policy default view should include resource "{res}"')
|
|
def step_policy_include_res(context: Any, res: str) -> None:
|
|
policy = _read_stored_policy(context)
|
|
assert res in policy.default_view.include_resources, (
|
|
f"{res} not in {policy.default_view.include_resources}"
|
|
)
|
|
|
|
|
|
@then('the stored policy default view should exclude resource "{res}"')
|
|
def step_policy_exclude_res(context: Any, res: str) -> None:
|
|
policy = _read_stored_policy(context)
|
|
assert res in policy.default_view.exclude_resources, (
|
|
f"{res} not in {policy.default_view.exclude_resources}"
|
|
)
|
|
|
|
|
|
@then('the stored policy default view should include path "{path}"')
|
|
def step_policy_include_path(context: Any, path: str) -> None:
|
|
policy = _read_stored_policy(context)
|
|
assert path in policy.default_view.include_paths, (
|
|
f"{path} not in {policy.default_view.include_paths}"
|
|
)
|
|
|
|
|
|
@then('the stored policy default view should exclude path "{path}"')
|
|
def step_policy_exclude_path(context: Any, path: str) -> None:
|
|
policy = _read_stored_policy(context)
|
|
assert path in policy.default_view.exclude_paths, (
|
|
f"{path} not in {policy.default_view.exclude_paths}"
|
|
)
|
|
|
|
|
|
@then("the stored policy default view max file size should be {size:d}")
|
|
def step_policy_max_file_size(context: Any, size: int) -> None:
|
|
policy = _read_stored_policy(context)
|
|
assert policy.default_view.max_file_size == size, (
|
|
f"Expected {size}, got {policy.default_view.max_file_size}"
|
|
)
|
|
|
|
|
|
@then("the stored policy default view max total size should be {size:d}")
|
|
def step_policy_max_total_size(context: Any, size: int) -> None:
|
|
policy = _read_stored_policy(context)
|
|
assert policy.default_view.max_total_size == size, (
|
|
f"Expected {size}, got {policy.default_view.max_total_size}"
|
|
)
|
|
|
|
|
|
@then('the stored policy strategize view should include resource "{res}"')
|
|
def step_policy_strat_include_res(context: Any, res: str) -> None:
|
|
policy = _read_stored_policy(context)
|
|
assert policy.strategize_view is not None, "strategize_view is None"
|
|
assert res in policy.strategize_view.include_resources
|
|
|
|
|
|
@then("the stored policy strategize view should be None")
|
|
def step_policy_strat_none(context: Any) -> None:
|
|
policy = _read_stored_policy(context)
|
|
assert policy.strategize_view is None, (
|
|
f"Expected None, got {policy.strategize_view}"
|
|
)
|
|
|
|
|
|
@then("the context inspect command should succeed")
|
|
def step_ctx_inspect_ok(context: Any) -> None:
|
|
assert context.ctx_exit_code == 0, (
|
|
f"Expected exit 0, got {context.ctx_exit_code}. Output: {context.ctx_output}"
|
|
)
|
|
|
|
|
|
@then("the context simulate command should succeed")
|
|
def step_ctx_simulate_ok(context: Any) -> None:
|
|
assert context.ctx_exit_code == 0, (
|
|
f"Expected exit 0, got {context.ctx_exit_code}. Output: {context.ctx_output}"
|
|
)
|
|
|
|
|
|
@then('a NotImplementedError should be raised with message containing "{msg}"')
|
|
def step_not_implemented(context: Any, msg: str) -> None:
|
|
assert context.ctx_error is not None, "No error was raised"
|
|
assert isinstance(context.ctx_error, NotImplementedError), (
|
|
f"Expected NotImplementedError, got {type(context.ctx_error)}"
|
|
)
|
|
assert msg in str(context.ctx_error), f"'{msg}' not in '{context.ctx_error}'"
|
|
|
|
|
|
@then('the resolved view should include resource "{res}"')
|
|
def step_resolved_view_include_res(context: Any, res: str) -> None:
|
|
# The show command output contains the resolved view
|
|
# We read the policy directly for verification
|
|
policy = _read_stored_policy(context)
|
|
# Determine which phase was queried from the output
|
|
# For simplicity, check all possible resolved views
|
|
found = False
|
|
for phase in ["default", "strategize", "execute", "apply"]:
|
|
resolved = policy.resolve_view(phase)
|
|
if res in resolved.include_resources:
|
|
found = True
|
|
break
|
|
assert found, f"Resource '{res}' not found in any resolved view"
|
|
|
|
|
|
@then("the context output should be valid JSON")
|
|
def step_ctx_output_valid_json(context: Any) -> None:
|
|
try:
|
|
json.loads(context.ctx_output.strip())
|
|
except json.JSONDecodeError as exc:
|
|
raise AssertionError(
|
|
f"Output is not valid JSON: {exc}\nOutput: {context.ctx_output!r}"
|
|
) from exc
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Issue #6323 spec compliance helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@given('issue 6323 project context sample is configured for "{project}"')
|
|
def step_issue6323_configure_policy(context: Any, project: str) -> None:
|
|
from cleveragents.cli.commands.project_context import (
|
|
_format_size as _format_size_helper,
|
|
)
|
|
from cleveragents.cli.commands.project_context import context_set
|
|
|
|
max_file = 1_048_576
|
|
max_total = 50 * 1_048_576
|
|
_run_with_container(
|
|
context,
|
|
context_set,
|
|
project=project,
|
|
view="strategize",
|
|
include_resource=["repo"],
|
|
exclude_path=["**/node_modules/**"],
|
|
hot_max_tokens=12_000,
|
|
warm_max_decisions=50,
|
|
cold_max_decisions=200,
|
|
query_limit=20,
|
|
summarize=True,
|
|
summary_max_tokens=800,
|
|
max_file_size=max_file,
|
|
max_total_size=max_total,
|
|
)
|
|
|
|
context.issue6323_expected = {
|
|
"context_policy": {
|
|
"project": project,
|
|
"view": "strategize",
|
|
"include_resources": ["repo"],
|
|
"exclude_resources": [],
|
|
"include_paths": [],
|
|
"exclude_paths": ["**/node_modules/**"],
|
|
},
|
|
"limits": {
|
|
"hot_max_tokens": 12_000,
|
|
"warm_max_decisions": 50,
|
|
"cold_max_decisions": 200,
|
|
"query_limit": 20,
|
|
"max_file_size": _format_size_helper(max_file),
|
|
"max_total_size": _format_size_helper(max_total),
|
|
},
|
|
"summarization": {
|
|
"enabled": True,
|
|
"max_tokens": 800,
|
|
},
|
|
"current_usage": {
|
|
"hot_context_tokens": 0,
|
|
"hot_context_limit": 12_000,
|
|
"warm_entries": 0,
|
|
"warm_limit": 50,
|
|
"cold_entries": 0,
|
|
"cold_limit": 200,
|
|
"indexed_resources": 0,
|
|
},
|
|
}
|
|
|
|
|
|
@given('the context tier service has sample usage data for "{project}"')
|
|
def step_issue6323_seed_usage(context: Any, project: str) -> None:
|
|
from cleveragents.application.services.context_tiers import ContextTierService
|
|
from cleveragents.domain.models.acms.tiers import ContextTier, TieredFragment
|
|
|
|
svc = getattr(context, "ctx_tier_service", ContextTierService())
|
|
context.ctx_tier_service = svc
|
|
|
|
svc._hot.clear()
|
|
svc._warm.clear()
|
|
svc._cold.clear()
|
|
with suppress(Exception):
|
|
svc._budget.max_tokens_hot = 12_000 # pragma: no cover - fallback guard
|
|
|
|
svc.store(
|
|
TieredFragment(
|
|
fragment_id="hot-frag",
|
|
content="hot context fragment",
|
|
tier=ContextTier.HOT,
|
|
project_name=project,
|
|
resource_id="res:repo",
|
|
token_count=8_420,
|
|
)
|
|
)
|
|
|
|
for idx in range(12):
|
|
svc.store(
|
|
TieredFragment(
|
|
fragment_id=f"warm-{idx}",
|
|
content="warm fragment",
|
|
tier=ContextTier.WARM,
|
|
project_name=project,
|
|
resource_id="res:repo",
|
|
token_count=10,
|
|
)
|
|
)
|
|
|
|
for idx in range(47):
|
|
svc.store(
|
|
TieredFragment(
|
|
fragment_id=f"cold-{idx}",
|
|
content="cold fragment",
|
|
tier=ContextTier.COLD,
|
|
project_name=project,
|
|
resource_id="res:repo",
|
|
token_count=5,
|
|
)
|
|
)
|
|
|
|
context.issue6323_expected["current_usage"].update(
|
|
{
|
|
"hot_context_tokens": "***REDACTED***",
|
|
"warm_entries": 12,
|
|
"cold_entries": 47,
|
|
"indexed_resources": 1,
|
|
}
|
|
)
|
|
|
|
|
|
def _issue6323_parse_json(context: Any) -> dict[str, Any]:
|
|
raw = context.ctx_output.strip()
|
|
return json.loads(raw)
|
|
|
|
|
|
def _issue6323_parse_yaml(context: Any) -> dict[str, Any]:
|
|
raw = context.ctx_output.strip()
|
|
return yaml.safe_load(raw)
|
|
|
|
|
|
def _issue6323_assert_payload(
|
|
parsed: dict[str, Any], expected: dict[str, Any], project: str, view: str
|
|
) -> None:
|
|
assert parsed["command"] == "project context show"
|
|
assert parsed["status"] == "ok"
|
|
assert parsed["exit_code"] == 0
|
|
assert "timing" in parsed and "duration_ms" in parsed["timing"]
|
|
assert parsed["messages"] == ["Context policy loaded"]
|
|
|
|
payload = parsed.get("data")
|
|
assert payload is not None, "Expected 'data' field in output payload"
|
|
|
|
context_policy = payload.get("context_policy")
|
|
assert context_policy == expected["context_policy"], (
|
|
f"Unexpected context_policy payload. Expected {expected['context_policy']}, got {context_policy}"
|
|
)
|
|
|
|
limits = payload.get("limits")
|
|
assert limits == expected["limits"], (
|
|
f"Unexpected limits payload. Expected {expected['limits']}, got {limits}"
|
|
)
|
|
|
|
summarization = payload.get("summarization")
|
|
assert summarization == expected["summarization"], (
|
|
"Unexpected summarization payload. "
|
|
f"Expected {expected['summarization']}, got {summarization}"
|
|
)
|
|
|
|
usage = payload.get("current_usage")
|
|
assert usage == expected["current_usage"], (
|
|
f"Unexpected current_usage payload. Expected {expected['current_usage']}, got {usage}"
|
|
)
|
|
|
|
|
|
@then(
|
|
'the context show json output should match issue 6323 spec for "{project}" and view "{view}"'
|
|
)
|
|
def step_issue6323_assert_json(context: Any, project: str, view: str) -> None:
|
|
parsed = _issue6323_parse_json(context)
|
|
_issue6323_assert_payload(parsed, context.issue6323_expected, project, view)
|
|
|
|
|
|
@then(
|
|
'the context show yaml output should match issue 6323 spec for "{project}" and view "{view}"'
|
|
)
|
|
def step_issue6323_assert_yaml(context: Any, project: str, view: str) -> None:
|
|
parsed = _issue6323_parse_yaml(context)
|
|
_issue6323_assert_payload(parsed, context.issue6323_expected, project, view)
|