brent.edwards
ece5e61725
CI / lint (pull_request) Successful in 16s
CI / quality (pull_request) Successful in 21s
CI / security (pull_request) Successful in 40s
CI / typecheck (pull_request) Successful in 41s
CI / benchmark-publish (pull_request) Has been skipped
CI / build (pull_request) Successful in 16s
CI / integration_tests (pull_request) Successful in 2m47s
CI / benchmark-regression (pull_request) Successful in 28m30s
CI / unit_tests (pull_request) Failing after 33m23s
CI / docker (pull_request) Has been skipped
CI / coverage (pull_request) Failing after 53m42s
555 lines
19 KiB
Python
555 lines
19 KiB
Python
"""Step definitions for M5 ACMS pipeline and large-project context smoke tests.
|
|
|
|
All step names are prefixed with ``m5 smoke`` to avoid ``AmbiguousStep``
|
|
conflicts with existing steps.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import fnmatch
|
|
import json
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from pydantic import ValidationError
|
|
from typer.testing import CliRunner
|
|
|
|
from cleveragents.cli.commands.context import app as context_app
|
|
from cleveragents.cli.commands.project_context import (
|
|
context_inspect,
|
|
context_simulate,
|
|
)
|
|
from cleveragents.domain.models.core.context import Context as ContextModel
|
|
from cleveragents.domain.models.core.context_policy import (
|
|
ContextView,
|
|
ProjectContextPolicy,
|
|
)
|
|
|
|
_FIXTURES_DIR = Path(__file__).resolve().parents[1] / "fixtures" / "m5"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a m5 smoke test runner")
|
|
def step_m5_smoke_runner(context: Context) -> None:
|
|
"""Set up the CLI runner for M5 smoke tests."""
|
|
context.runner = CliRunner()
|
|
|
|
|
|
@given("a m5 smoke mocked context service")
|
|
def step_m5_smoke_mocked_service(context: Context) -> None:
|
|
"""Set up mocked context service for M5 smoke tests."""
|
|
context.mock_context_service = MagicMock()
|
|
context.m5_error = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixture loading
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I m5 smoke load the ACMS context policy fixture")
|
|
def step_m5_load_policy_fixture(context: Context) -> None:
|
|
fixture_path = _FIXTURES_DIR / "acms_context_policy.json"
|
|
context.m5_policy_fixture = json.loads(fixture_path.read_text(encoding="utf-8"))
|
|
|
|
|
|
@then("the m5 smoke policy fixture should have a default view")
|
|
def step_m5_policy_has_default(context: Context) -> None:
|
|
assert "default_view" in context.m5_policy_fixture
|
|
|
|
|
|
@then("the m5 smoke policy fixture should have a strategize view")
|
|
def step_m5_policy_has_strategize(context: Context) -> None:
|
|
assert "strategize_view" in context.m5_policy_fixture
|
|
|
|
|
|
@then("the m5 smoke policy fixture should have an execute view")
|
|
def step_m5_policy_has_execute(context: Context) -> None:
|
|
assert "execute_view" in context.m5_policy_fixture
|
|
|
|
|
|
@when("I m5 smoke load the large project context fixture")
|
|
def step_m5_load_large_project(context: Context) -> None:
|
|
fixture_path = _FIXTURES_DIR / "large_project_context.json"
|
|
context.m5_large_project = json.loads(fixture_path.read_text(encoding="utf-8"))
|
|
|
|
|
|
@then("the m5 smoke large project fixture should have file entries")
|
|
def step_m5_large_project_has_files(context: Context) -> None:
|
|
assert "file_entries" in context.m5_large_project
|
|
assert len(context.m5_large_project["file_entries"]) > 0
|
|
|
|
|
|
@then("the m5 smoke large project fixture should have context tiers")
|
|
def step_m5_large_project_has_tiers(context: Context) -> None:
|
|
tiers = context.m5_large_project.get("context_tiers", {})
|
|
assert "hot" in tiers
|
|
assert "warm" in tiers
|
|
assert "cold" in tiers
|
|
|
|
|
|
@when("I m5 smoke load the context analysis results fixture")
|
|
def step_m5_load_analysis(context: Context) -> None:
|
|
fixture_path = _FIXTURES_DIR / "context_analysis_results.json"
|
|
context.m5_analysis_fixture = json.loads(fixture_path.read_text(encoding="utf-8"))
|
|
|
|
|
|
@then("the m5 smoke analysis fixture should have a summary")
|
|
def step_m5_analysis_has_summary(context: Context) -> None:
|
|
assert context.m5_analysis_fixture.get("summary")
|
|
|
|
|
|
@then("the m5 smoke analysis fixture should have dependencies")
|
|
def step_m5_analysis_has_deps(context: Context) -> None:
|
|
assert "dependencies" in context.m5_analysis_fixture
|
|
assert len(context.m5_analysis_fixture["dependencies"]) > 0
|
|
|
|
|
|
@then("the m5 smoke analysis fixture should have relevance scores")
|
|
def step_m5_analysis_has_scores(context: Context) -> None:
|
|
assert "relevance_scores" in context.m5_analysis_fixture
|
|
assert len(context.m5_analysis_fixture["relevance_scores"]) > 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Context policy resolution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a m5 smoke empty context policy")
|
|
def step_m5_empty_policy(context: Context) -> None:
|
|
context.m5_policy = ProjectContextPolicy()
|
|
|
|
|
|
@given("a m5 smoke policy with only default view")
|
|
def step_m5_default_only_policy(context: Context) -> None:
|
|
context.m5_policy = ProjectContextPolicy(
|
|
default_view=ContextView(
|
|
include_paths=["src/**/*.py"],
|
|
exclude_paths=["**/__pycache__/**"],
|
|
max_file_size=262144,
|
|
),
|
|
)
|
|
|
|
|
|
@given("a m5 smoke policy with strategize override")
|
|
def step_m5_strategize_override(context: Context) -> None:
|
|
context.m5_policy = ProjectContextPolicy(
|
|
default_view=ContextView(
|
|
include_paths=["src/**/*.py"],
|
|
),
|
|
strategize_view=ContextView(
|
|
include_paths=["src/**/*.py", "docs/**/*.md"],
|
|
max_file_size=131072,
|
|
),
|
|
)
|
|
|
|
|
|
@when('I m5 smoke resolve the view for phase "{phase}"')
|
|
def step_m5_resolve_view(context: Context, phase: str) -> None:
|
|
context.m5_resolved_view = context.m5_policy.resolve_view(phase)
|
|
|
|
|
|
@when('I m5 smoke resolve the view for invalid phase "{phase}"')
|
|
def step_m5_resolve_invalid_phase(context: Context, phase: str) -> None:
|
|
try:
|
|
context.m5_policy.resolve_view(phase)
|
|
context.m5_error = None
|
|
except ValueError as exc:
|
|
context.m5_error = exc
|
|
|
|
|
|
@then("the m5 smoke resolved view should include all paths")
|
|
def step_m5_view_includes_all(context: Context) -> None:
|
|
assert context.m5_resolved_view.include_paths == []
|
|
|
|
|
|
@then("the m5 smoke resolved view should match the default view")
|
|
def step_m5_view_matches_default(context: Context) -> None:
|
|
default_view = context.m5_policy.default_view
|
|
assert context.m5_resolved_view == default_view
|
|
|
|
|
|
@then("the m5 smoke resolved view should use the strategize override")
|
|
def step_m5_view_uses_strategize(context: Context) -> None:
|
|
assert context.m5_resolved_view == context.m5_policy.strategize_view
|
|
|
|
|
|
@then("a m5 smoke ValueError should be raised")
|
|
def step_m5_value_error_raised(context: Context) -> None:
|
|
assert context.m5_error is not None
|
|
assert isinstance(context.m5_error, ValueError)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Budget enforcement
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a m5 smoke context view with max_file_size {size:d}")
|
|
def step_m5_view_max_file_size(context: Context, size: int) -> None:
|
|
context.m5_view = ContextView(max_file_size=size)
|
|
|
|
|
|
@given("a m5 smoke context view with max_total_size {size:d}")
|
|
def step_m5_view_max_total_size(context: Context, size: int) -> None:
|
|
context.m5_view = ContextView(max_total_size=size)
|
|
|
|
|
|
@given("a m5 smoke context view with no size limits")
|
|
def step_m5_view_no_limits(context: Context) -> None:
|
|
context.m5_view = ContextView()
|
|
|
|
|
|
@then("a m5 smoke file of size {size:d} should exceed the budget")
|
|
def step_m5_file_exceeds(context: Context, size: int) -> None:
|
|
assert context.m5_view.max_file_size is not None
|
|
assert size > context.m5_view.max_file_size
|
|
|
|
|
|
@then("a m5 smoke file of size {size:d} should be within the budget")
|
|
def step_m5_file_within(context: Context, size: int) -> None:
|
|
if context.m5_view.max_file_size is None:
|
|
return # No limit means always within budget
|
|
assert size <= context.m5_view.max_file_size
|
|
|
|
|
|
@then("a m5 smoke aggregate of size {size:d} should exceed the budget")
|
|
def step_m5_aggregate_exceeds(context: Context, size: int) -> None:
|
|
assert context.m5_view.max_total_size is not None
|
|
assert size > context.m5_view.max_total_size
|
|
|
|
|
|
@then("a m5 smoke aggregate of size {size:d} should be within the budget")
|
|
def step_m5_aggregate_within(context: Context, size: int) -> None:
|
|
if context.m5_view.max_total_size is None:
|
|
return # No limit means always within budget
|
|
assert size <= context.m5_view.max_total_size
|
|
|
|
|
|
@when("I m5 smoke create a context view with max_file_size {size:d}")
|
|
def step_m5_create_view_file_size(context: Context, size: int) -> None:
|
|
try:
|
|
ContextView(max_file_size=size)
|
|
context.m5_error = None
|
|
except ValidationError as exc:
|
|
context.m5_error = exc
|
|
|
|
|
|
@when("I m5 smoke create a context view with max_total_size {size:d}")
|
|
def step_m5_create_view_total_size(context: Context, size: int) -> None:
|
|
try:
|
|
ContextView(max_total_size=size)
|
|
context.m5_error = None
|
|
except ValidationError as exc:
|
|
context.m5_error = exc
|
|
|
|
|
|
@then("a m5 smoke validation error should be raised")
|
|
def step_m5_validation_error(context: Context) -> None:
|
|
assert context.m5_error is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Context assembly via CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a m5 smoke project with no context")
|
|
def step_m5_project_no_context(context: Context) -> None:
|
|
context.mock_context_service.list_context.return_value = []
|
|
|
|
|
|
@given("a m5 smoke project with mocked context service")
|
|
def step_m5_project_mocked_service(context: Context) -> None:
|
|
context.mock_context_service.add_to_context.return_value = (
|
|
[Path("src/main.py")],
|
|
[],
|
|
)
|
|
|
|
|
|
@given("a m5 smoke project with context entries")
|
|
def step_m5_project_with_entries(context: Context) -> None:
|
|
mock_entry = MagicMock(spec=ContextModel)
|
|
mock_entry.path = "src/main.py"
|
|
mock_entry.size = 1024
|
|
mock_entry.type = "FILE"
|
|
context.mock_context_service.list_context.return_value = [mock_entry]
|
|
context.mock_context_service.show_context_content.return_value = {
|
|
"src/main.py": "print('hello')"
|
|
}
|
|
context.mock_context_service.get_context_content.return_value = "print('hello')"
|
|
context.mock_context_service.clear_context.return_value = 1
|
|
|
|
|
|
@when("I m5 smoke invoke context list")
|
|
def step_m5_invoke_context_list(context: Context) -> None:
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.context.ContextService",
|
|
return_value=context.mock_context_service,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.context._resolve_project",
|
|
return_value=MagicMock(),
|
|
),
|
|
):
|
|
result = context.runner.invoke(context_app, ["list", "--format", "json"])
|
|
context.m5_result = result
|
|
|
|
|
|
@then("the m5 smoke context list should succeed")
|
|
def step_m5_context_list_success(context: Context) -> None:
|
|
assert context.m5_result.exit_code == 0, (
|
|
f"Exit code: {context.m5_result.exit_code}\n{context.m5_result.output}"
|
|
)
|
|
|
|
|
|
@then("the m5 smoke context list output should be empty")
|
|
def step_m5_context_list_empty(context: Context) -> None:
|
|
# Empty context list may show "[]" or "No context" message
|
|
output = context.m5_result.output.strip()
|
|
assert output == "[]" or "no context" in output.lower() or output == ""
|
|
|
|
|
|
@when('I m5 smoke invoke context add with path "{path}"')
|
|
def step_m5_invoke_context_add(context: Context, path: str) -> None:
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.context.ContextService",
|
|
return_value=context.mock_context_service,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.context._resolve_project",
|
|
return_value=MagicMock(),
|
|
),
|
|
):
|
|
result = context.runner.invoke(context_app, ["add", path])
|
|
context.m5_result = result
|
|
|
|
|
|
@then("the m5 smoke context add should succeed")
|
|
def step_m5_context_add_success(context: Context) -> None:
|
|
assert context.m5_result.exit_code == 0, (
|
|
f"Exit code: {context.m5_result.exit_code}\n{context.m5_result.output}"
|
|
)
|
|
|
|
|
|
@when('I m5 smoke invoke context show for path "{path}"')
|
|
def step_m5_invoke_context_show(context: Context, path: str) -> None:
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.context.ContextService",
|
|
return_value=context.mock_context_service,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.context._resolve_project",
|
|
return_value=MagicMock(),
|
|
),
|
|
):
|
|
result = context.runner.invoke(context_app, ["show", path])
|
|
context.m5_result = result
|
|
|
|
|
|
@then("the m5 smoke context show should succeed")
|
|
def step_m5_context_show_success(context: Context) -> None:
|
|
assert context.m5_result.exit_code == 0, (
|
|
f"Exit code: {context.m5_result.exit_code}\n{context.m5_result.output}"
|
|
)
|
|
|
|
|
|
@then("the m5 smoke context show output should contain content")
|
|
def step_m5_context_show_has_content(context: Context) -> None:
|
|
assert len(context.m5_result.output.strip()) > 0
|
|
|
|
|
|
@when("I m5 smoke invoke context clear")
|
|
def step_m5_invoke_context_clear(context: Context) -> None:
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.context.ContextService",
|
|
return_value=context.mock_context_service,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.context._resolve_project",
|
|
return_value=MagicMock(),
|
|
),
|
|
):
|
|
result = context.runner.invoke(context_app, ["clear", "--yes"])
|
|
context.m5_result = result
|
|
|
|
|
|
@then("the m5 smoke context clear should succeed")
|
|
def step_m5_context_clear_success(context: Context) -> None:
|
|
assert context.m5_result.exit_code == 0, (
|
|
f"Exit code: {context.m5_result.exit_code}\n{context.m5_result.output}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Project context policy CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a m5 smoke project with a saved context policy")
|
|
def step_m5_project_with_policy(context: Context) -> None:
|
|
context.m5_saved_policy = ProjectContextPolicy(
|
|
default_view=ContextView(
|
|
include_paths=["src/**/*.py"],
|
|
max_file_size=262144,
|
|
),
|
|
)
|
|
|
|
|
|
@when("I m5 smoke invoke project context show")
|
|
def step_m5_invoke_project_context_show(context: Context) -> None:
|
|
mock_project = MagicMock()
|
|
mock_project.context_policy = context.m5_saved_policy
|
|
|
|
with patch(
|
|
"cleveragents.cli.commands.project_context._get_namespaced_project_repo",
|
|
return_value=(mock_project, MagicMock()),
|
|
):
|
|
from cleveragents.cli.commands.project_context import app as pc_app
|
|
|
|
result = context.runner.invoke(pc_app, ["show", "local/m5-proj"])
|
|
context.m5_result = result
|
|
|
|
|
|
@then("the m5 smoke project context show should succeed")
|
|
def step_m5_project_context_show_success(context: Context) -> None:
|
|
assert context.m5_result.exit_code == 0, (
|
|
f"Exit code: {context.m5_result.exit_code}\n{context.m5_result.output}"
|
|
)
|
|
|
|
|
|
@then("the m5 smoke project context output should contain phase views")
|
|
def step_m5_project_context_has_phases(context: Context) -> None:
|
|
output = context.m5_result.output
|
|
assert "default" in output.lower() or "view" in output.lower()
|
|
|
|
|
|
@when("I m5 smoke invoke project context inspect")
|
|
def step_m5_invoke_context_inspect(context: Context) -> None:
|
|
try:
|
|
context_inspect(project="local/m5-proj")
|
|
context.m5_error = None
|
|
except NotImplementedError as exc:
|
|
context.m5_error = exc
|
|
|
|
|
|
@when("I m5 smoke invoke project context simulate")
|
|
def step_m5_invoke_context_simulate(context: Context) -> None:
|
|
try:
|
|
context_simulate(project="local/m5-proj")
|
|
context.m5_error = None
|
|
except NotImplementedError as exc:
|
|
context.m5_error = exc
|
|
|
|
|
|
@then('a m5 smoke NotImplementedError should be raised mentioning "{text}"')
|
|
def step_m5_not_implemented_mentioning(context: Context, text: str) -> None:
|
|
assert context.m5_error is not None, "Expected NotImplementedError was not raised"
|
|
assert isinstance(context.m5_error, NotImplementedError)
|
|
assert text.lower() in str(context.m5_error).lower(), (
|
|
f"Expected '{text}' in: {context.m5_error}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Context analysis agent
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a m5 smoke mocked context analysis agent")
|
|
def step_m5_mocked_analysis_agent(context: Context) -> None:
|
|
fixture_path = _FIXTURES_DIR / "context_analysis_results.json"
|
|
fixture_data = json.loads(fixture_path.read_text(encoding="utf-8"))
|
|
context.m5_mock_analysis = {
|
|
"file_paths": list(fixture_data["dependencies"].keys()),
|
|
"documents": [],
|
|
"dependencies": fixture_data["dependencies"],
|
|
"summary": fixture_data["summary"],
|
|
"relevance_scores": fixture_data["relevance_scores"],
|
|
"chunks": [],
|
|
"error": None,
|
|
}
|
|
|
|
|
|
@when("I m5 smoke invoke context analysis")
|
|
def step_m5_invoke_analysis(context: Context) -> None:
|
|
context.m5_analysis_result = context.m5_mock_analysis
|
|
|
|
|
|
@then("the m5 smoke analysis result should have a non-empty summary")
|
|
def step_m5_analysis_has_nonempty_summary(context: Context) -> None:
|
|
assert context.m5_analysis_result["summary"]
|
|
|
|
|
|
@then("the m5 smoke analysis result should have dependency entries")
|
|
def step_m5_analysis_has_dep_entries(context: Context) -> None:
|
|
assert len(context.m5_analysis_result["dependencies"]) > 0
|
|
|
|
|
|
@then("the m5 smoke analysis result should have relevance scores")
|
|
def step_m5_analysis_has_rel_scores(context: Context) -> None:
|
|
assert len(context.m5_analysis_result["relevance_scores"]) > 0
|
|
|
|
|
|
@then("all m5 smoke relevance scores should be between 0 and 1")
|
|
def step_m5_scores_bounded(context: Context) -> None:
|
|
for path, score in context.m5_analysis_result["relevance_scores"].items():
|
|
assert 0.0 <= score <= 1.0, f"Score for {path} out of range: {score}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Multi-project context
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('m5 smoke project "{name}" with {count:d} context files')
|
|
def step_m5_project_with_files(context: Context, name: str, count: int) -> None:
|
|
if not hasattr(context, "m5_projects"):
|
|
context.m5_projects = {}
|
|
context.m5_projects[name] = [
|
|
{"path": f"src/file_{i}.py", "size": 1024} for i in range(count)
|
|
]
|
|
|
|
|
|
@then('m5 smoke project "{name}" should have {count:d} context entries')
|
|
def step_m5_project_entry_count(context: Context, name: str, count: int) -> None:
|
|
assert len(context.m5_projects[name]) == count, (
|
|
f"Expected {count} entries for {name}, got {len(context.m5_projects[name])}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Context exclusion patterns
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('a m5 smoke context view excluding "{pattern}"')
|
|
def step_m5_view_excluding(context: Context, pattern: str) -> None:
|
|
context.m5_view = ContextView(exclude_paths=[pattern])
|
|
context.m5_exclude_pattern = pattern
|
|
|
|
|
|
@then('the m5 smoke path "{path}" should be excluded')
|
|
def step_m5_path_excluded(context: Context, path: str) -> None:
|
|
assert fnmatch.fnmatch(path, context.m5_exclude_pattern), (
|
|
f"Expected '{path}' to match exclude pattern '{context.m5_exclude_pattern}'"
|
|
)
|
|
|
|
|
|
@then('the m5 smoke path "{path}" should not be excluded')
|
|
def step_m5_path_not_excluded(context: Context, path: str) -> None:
|
|
assert not fnmatch.fnmatch(path, context.m5_exclude_pattern), (
|
|
f"Expected '{path}' NOT to match exclude pattern"
|
|
)
|