HAL9000
0aacec7845
- Fix ruff import ordering in project_context_cli_steps.py (yaml moved to third-party section) - Remove unused noqa: F401 suppression in rendering/__init__.py - Add *args, **kwargs to _test_format_output stub in coverage boost steps (fixes TypeError) - Extract context-set step definitions into dedicated module (project_context_set_steps.py) to bring project_context_cli_steps.py under 500-line limit (469 -> 316 lines) - Revert spec changes for context show/inspect/simulate message format (bare strings) to keep PR focused on issue #6319 (context set only),
441 lines
14 KiB
Python
441 lines
14 KiB
Python
"""Step definitions for the ``agents project context set`` CLI command.
|
|
|
|
Extracted from ``project_context_cli_steps.py`` to keep that file under
|
|
the 500-line limit mandated by CONTRIBUTING.md. All shared fixtures and
|
|
runners are imported back into this module.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
# third-party
|
|
import yaml
|
|
from behave import given, then, when # type: ignore[import-untyped]
|
|
|
|
# Shared helpers from the parent CLI steps module
|
|
from features.steps.project_context_cli_steps import (
|
|
_run_with_container,
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# When steps — context set operations
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@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 spec example on "{project}" with format "{fmt}"')
|
|
def step_ctx_set_spec_example(context: Any, project: str, fmt: str) -> None:
|
|
from cleveragents.cli.commands.project_context import (
|
|
context_set,
|
|
)
|
|
|
|
_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,
|
|
max_file_size=1_048_576,
|
|
max_total_size=50 * 1_048_576,
|
|
summarize=True,
|
|
summary_max_tokens=800,
|
|
output_format=fmt,
|
|
)
|
|
|
|
|
|
@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,
|
|
)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Given steps — context set 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],
|
|
)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Then assertions — context set output and policy verification
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@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 set JSON output should match the spec envelope")
|
|
def step_ctx_set_json_spec(context: Any) -> None:
|
|
output = context.ctx_output.strip()
|
|
assert output, "No CLI output captured"
|
|
payload = json.loads(output)
|
|
|
|
assert payload.get("command") == "project context set"
|
|
assert payload.get("status") == "ok"
|
|
assert payload.get("exit_code") == 0
|
|
|
|
timing = payload.get("timing", {})
|
|
assert isinstance(timing.get("duration_ms"), int)
|
|
|
|
messages = payload.get("messages", [])
|
|
assert messages and messages[0].get("text") == "Context policy updated"
|
|
|
|
data = payload.get("data", {})
|
|
|
|
context_policy = data.get("context_policy", {})
|
|
assert context_policy.get("project") == "local/ctx-app"
|
|
assert context_policy.get("view") == "strategize"
|
|
assert context_policy.get("include_resources") == ["repo"]
|
|
assert context_policy.get("exclude_paths") == ["**/node_modules/**"]
|
|
|
|
limits = data.get("limits", {})
|
|
assert limits.get("hot_max_tokens") == 12_000
|
|
assert limits.get("warm_max_decisions") == 50
|
|
assert limits.get("cold_max_decisions") == 200
|
|
assert limits.get("query_limit") == 20
|
|
assert limits.get("max_file_size") == "1 MB"
|
|
assert limits.get("max_total_size") == "50 MB"
|
|
|
|
summarization = data.get("summarization", {})
|
|
assert summarization.get("enabled") is True
|
|
assert summarization.get("max_tokens") == 800
|
|
|
|
other_views = data.get("other_views", {})
|
|
assert other_views.get("default") == "(unset)"
|
|
assert other_views.get("execute") == "(default)"
|
|
assert other_views.get("apply") == "(default)"
|
|
|
|
|
|
@then("the context set YAML output should match the spec envelope")
|
|
def step_ctx_set_yaml_spec(context: Any) -> None:
|
|
output = context.ctx_output.strip()
|
|
assert output, "No CLI output captured"
|
|
payload = yaml.safe_load(output)
|
|
|
|
assert payload.get("command") == "project context set"
|
|
assert payload.get("status") == "ok"
|
|
assert payload.get("exit_code") == 0
|
|
|
|
timing = payload.get("timing", {})
|
|
assert isinstance(timing.get("duration_ms"), int)
|
|
|
|
messages = payload.get("messages", [])
|
|
assert messages and messages[0].get("text") == "Context policy updated"
|
|
|
|
data = payload.get("data", {})
|
|
context_policy = data.get("context_policy", {})
|
|
assert context_policy.get("project") == "local/ctx-app"
|
|
assert context_policy.get("view") == "strategize"
|
|
assert context_policy.get("include_resources") == ["repo"]
|
|
assert context_policy.get("exclude_paths") == ["**/node_modules/**"]
|
|
|
|
limits = data.get("limits", {})
|
|
assert limits.get("hot_max_tokens") == 12_000
|
|
assert limits.get("warm_max_decisions") == 50
|
|
assert limits.get("cold_max_decisions") == 200
|
|
assert limits.get("query_limit") == 20
|
|
assert limits.get("max_file_size") == "1 MB"
|
|
assert limits.get("max_total_size") == "50 MB"
|
|
|
|
summarization = data.get("summarization", {})
|
|
assert summarization.get("enabled") is True
|
|
assert summarization.get("max_tokens") == 800
|
|
|
|
other_views = data.get("other_views", {})
|
|
assert other_views.get("default") == "(unset)"
|
|
assert other_views.get("execute") == "(default)"
|
|
assert other_views.get("apply") == "(default)"
|
|
|
|
|
|
@then("the context set rich output should include spec panels")
|
|
def step_ctx_set_rich_spec(context: Any) -> None:
|
|
output = context.ctx_output
|
|
assert "Context Policy" in output
|
|
assert "Limits" in output
|
|
assert "Summarization" in output
|
|
assert "Other Views" in output
|
|
assert "Project: local/ctx-app" in output
|
|
assert "View: strategize" in output
|
|
assert "Include: repo" in output
|
|
assert "Exclude: **/node_modules/**" in output
|
|
assert "Hot Tokens: 12000 (soft cap)" in output
|
|
assert "Warm Decisions: 50" in output
|
|
assert "Cold Decisions: 200" in output
|
|
assert "Query Limit: 20" in output
|
|
assert "Max File Size: 1 MB" in output
|
|
assert "Max Total Size: 50 MB" in output
|
|
assert "Enabled: yes" in output
|
|
assert "Max Tokens: 800" in output
|
|
assert "default: (unset)" in output
|
|
assert "execute: (default)" in output
|
|
assert "apply: (default)" in output
|
|
assert "\u2713 Context policy updated" in output
|
|
|
|
|
|
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}"
|
|
)
|