forked from HAL9000/cleveragents-core
freemo
2cb82f8acc
Implements M7 (v3.6.0) acceptance criterion: an end-to-end porting task completes autonomously using the full CleverAgents pipeline. Changes: - Add reproducible E2E porting task fixture (examples/actions/e2e-porting.yaml) with invariants, typed arguments, and supervised automation profile - Add E2EPortingService that ties together ACMS context assembly, plan lifecycle (Strategize → Execute → Apply), subplan decomposition, correction engine, and apply phase - Add validate_ported_output() for Python 3 syntax, type annotation, and Python 2 construct checks - Add _apply_stub_porting() deterministic Python 2→3 transformation for test path (no LLM required) - Add Behave BDD tests (features/e2e_porting_task.feature) covering fixture validation, action registration, plan lifecycle, output validation, and lifecycle observability - Add Robot Framework E2E acceptance test (robot/e2e/m7_e2e_porting.robot) covering all M7 acceptance criteria Closes #857
471 lines
17 KiB
Python
471 lines
17 KiB
Python
"""Step definitions for E2E porting task feature.
|
|
|
|
Tests the full autonomous porting pipeline:
|
|
- YAML fixture loading and validation
|
|
- Action registration with PlanLifecycleService
|
|
- Plan lifecycle (Strategize → Execute → Apply)
|
|
- Output validation checks
|
|
- Lifecycle event observability
|
|
|
|
All mocks are confined to this file (no external mock modules needed).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.action.schema import ActionConfigSchema
|
|
from cleveragents.application.services.e2e_porting_service import (
|
|
E2EPortingService,
|
|
PortingTaskConfig,
|
|
_apply_stub_porting,
|
|
validate_ported_output,
|
|
)
|
|
from cleveragents.application.services.plan_lifecycle_service import (
|
|
PlanLifecycleService,
|
|
)
|
|
from cleveragents.config.settings import Settings
|
|
|
|
__all__: list[str] = []
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Repository root
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sample Python 2 source for testing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_SAMPLE_PY2_SOURCE = '''\
|
|
"""Calculator module (Python 2)."""
|
|
|
|
|
|
def add(a, b):
|
|
"""Add two numbers."""
|
|
return a + b
|
|
|
|
|
|
def greet(name):
|
|
"""Print a greeting."""
|
|
print "Hello, " + name
|
|
|
|
|
|
def count_items(items):
|
|
"""Count items using xrange."""
|
|
total = 0
|
|
for i in xrange(len(items)):
|
|
total += 1
|
|
return total
|
|
|
|
|
|
def get_keys(d):
|
|
"""Get dictionary keys."""
|
|
return list(d.iterkeys())
|
|
'''
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("an E2E porting test environment")
|
|
def step_setup_porting_env(context: Context) -> None:
|
|
"""Set up a clean test environment for E2E porting tests."""
|
|
Settings._instance = None
|
|
context.settings = Settings()
|
|
context.service = PlanLifecycleService(settings=context.settings)
|
|
context.porting_service = E2EPortingService(
|
|
settings=context.settings,
|
|
lifecycle_service=context.service,
|
|
)
|
|
context.error = None
|
|
context.fixture_schema: ActionConfigSchema | None = None
|
|
context.plan = None
|
|
context.porting_result = None
|
|
context.validation_passed: bool | None = None
|
|
context.validation_errors: list[str] = []
|
|
context.stub_output: str = ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixture loading
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I load the e2e-porting action fixture from {fixture_path}")
|
|
def step_load_fixture(context: Context, fixture_path: str) -> None:
|
|
"""Load the E2E porting action YAML fixture."""
|
|
full_path = _REPO_ROOT / fixture_path
|
|
try:
|
|
context.fixture_schema = ActionConfigSchema.from_yaml_file(full_path)
|
|
context.error = None
|
|
except Exception as exc:
|
|
context.error = exc
|
|
context.fixture_schema = None
|
|
|
|
|
|
@then("the fixture should parse without errors")
|
|
def step_fixture_no_errors(context: Context) -> None:
|
|
assert context.error is None, f"Fixture parse error: {context.error}"
|
|
assert context.fixture_schema is not None, "Fixture schema is None"
|
|
|
|
|
|
@then('the fixture name should be "{expected_name}"')
|
|
def step_fixture_name(context: Context, expected_name: str) -> None:
|
|
assert context.fixture_schema is not None
|
|
assert context.fixture_schema.name == expected_name, (
|
|
f"Expected name '{expected_name}', got '{context.fixture_schema.name}'"
|
|
)
|
|
|
|
|
|
@then("the fixture should have a strategy_actor")
|
|
def step_fixture_has_strategy_actor(context: Context) -> None:
|
|
assert context.fixture_schema is not None
|
|
assert context.fixture_schema.strategy_actor, "strategy_actor is empty"
|
|
|
|
|
|
@then("the fixture should have an execution_actor")
|
|
def step_fixture_has_execution_actor(context: Context) -> None:
|
|
assert context.fixture_schema is not None
|
|
assert context.fixture_schema.execution_actor, "execution_actor is empty"
|
|
|
|
|
|
@then("the fixture should have a definition_of_done")
|
|
def step_fixture_has_dod(context: Context) -> None:
|
|
assert context.fixture_schema is not None
|
|
assert context.fixture_schema.definition_of_done, "definition_of_done is empty"
|
|
|
|
|
|
@then('the fixture invariants should include "{keyword}"')
|
|
def step_fixture_invariants_include(context: Context, keyword: str) -> None:
|
|
assert context.fixture_schema is not None
|
|
invariants_text = " ".join(context.fixture_schema.invariants).lower()
|
|
assert keyword.lower() in invariants_text, (
|
|
f"Invariants do not mention '{keyword}'. "
|
|
f"Invariants: {context.fixture_schema.invariants}"
|
|
)
|
|
|
|
|
|
@then('the fixture should have an argument named "{arg_name}"')
|
|
def step_fixture_has_argument(context: Context, arg_name: str) -> None:
|
|
assert context.fixture_schema is not None
|
|
arg_names = [a.name for a in context.fixture_schema.arguments]
|
|
assert arg_name in arg_names, (
|
|
f"Argument '{arg_name}' not found. Available: {arg_names}"
|
|
)
|
|
|
|
|
|
@then('the "{arg_name}" argument should be required')
|
|
def step_argument_is_required(context: Context, arg_name: str) -> None:
|
|
assert context.fixture_schema is not None
|
|
for arg in context.fixture_schema.arguments:
|
|
if arg.name == arg_name:
|
|
assert arg.required, f"Argument '{arg_name}' is not required"
|
|
return
|
|
raise AssertionError(f"Argument '{arg_name}' not found in fixture")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Action registration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I register the porting action with the lifecycle service")
|
|
def step_register_porting_action(context: Context) -> None:
|
|
"""Register the porting action (idempotent)."""
|
|
try:
|
|
context.porting_service.register_porting_action()
|
|
context.error = None
|
|
except Exception as exc:
|
|
context.error = exc
|
|
|
|
|
|
@when("I register the porting action with the lifecycle service again")
|
|
def step_register_porting_action_again(context: Context) -> None:
|
|
"""Register the porting action a second time (should be idempotent)."""
|
|
try:
|
|
context.porting_service.register_porting_action()
|
|
context.error = None
|
|
except Exception as exc:
|
|
context.error = exc
|
|
|
|
|
|
@then('the action "{action_name}" should be available')
|
|
def step_action_available(context: Context, action_name: str) -> None:
|
|
action = context.service.get_action(action_name)
|
|
assert action is not None, f"Action '{action_name}' not found"
|
|
|
|
|
|
@then('the porting action state should be "{expected_state}"')
|
|
def step_porting_action_state(context: Context, expected_state: str) -> None:
|
|
from cleveragents.application.services.e2e_porting_service import (
|
|
PORTING_ACTION_NAME,
|
|
)
|
|
|
|
action = context.service.get_action(PORTING_ACTION_NAME)
|
|
assert action.state.value == expected_state, (
|
|
f"Expected state '{expected_state}', got '{action.state.value}'"
|
|
)
|
|
|
|
|
|
@then("no error should have occurred")
|
|
def step_no_error(context: Context) -> None:
|
|
assert context.error is None, f"Unexpected error: {context.error}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Plan creation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the porting action is registered")
|
|
def step_porting_action_registered(context: Context) -> None:
|
|
"""Ensure the porting action is registered."""
|
|
context.porting_service.register_porting_action()
|
|
|
|
|
|
@when('I use the porting action with source_module "{source_module}"')
|
|
def step_use_porting_action(context: Context, source_module: str) -> None:
|
|
"""Use the porting action to create a plan."""
|
|
from cleveragents.application.services.e2e_porting_service import (
|
|
PORTING_ACTION_NAME,
|
|
)
|
|
from cleveragents.domain.models.core.plan import ProjectLink
|
|
|
|
context.plan = context.service.use_action(
|
|
action_name=PORTING_ACTION_NAME,
|
|
project_links=[ProjectLink(project_name="test-project")],
|
|
arguments={
|
|
"source_module": source_module,
|
|
"target_module": source_module,
|
|
"add_tests": True,
|
|
},
|
|
)
|
|
|
|
|
|
@then("a porting plan should be created")
|
|
def step_porting_plan_created(context: Context) -> None:
|
|
assert context.plan is not None, "No plan was created"
|
|
assert context.plan.identity.plan_id, "Plan has no ID"
|
|
|
|
|
|
@then('the porting plan phase should be "{expected_phase}"')
|
|
def step_porting_plan_phase(context: Context, expected_phase: str) -> None:
|
|
assert context.plan is not None
|
|
assert context.plan.phase.value == expected_phase, (
|
|
f"Expected phase '{expected_phase}', got '{context.plan.phase.value}'"
|
|
)
|
|
|
|
|
|
@then('the porting plan processing_state should be "{expected_state}"')
|
|
def step_porting_plan_processing_state(context: Context, expected_state: str) -> None:
|
|
assert context.plan is not None
|
|
actual = (
|
|
context.plan.processing_state.value if context.plan.processing_state else "none"
|
|
)
|
|
assert actual == expected_state, (
|
|
f"Expected processing_state '{expected_state}', got '{actual}'"
|
|
)
|
|
|
|
|
|
@then('the porting plan invariants should include "{keyword}"')
|
|
def step_porting_plan_invariants_include(context: Context, keyword: str) -> None:
|
|
assert context.plan is not None
|
|
invariants_text = " ".join(inv.text for inv in context.plan.invariants).lower()
|
|
assert keyword.lower() in invariants_text, (
|
|
f"Plan invariants do not mention '{keyword}'. "
|
|
f"Invariants: {[inv.text for inv in context.plan.invariants]}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Full lifecycle execution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a Python 2 source module is available")
|
|
def step_py2_source_available(context: Context) -> None:
|
|
"""Set up a sample Python 2 source module."""
|
|
context.py2_source = _SAMPLE_PY2_SOURCE
|
|
|
|
|
|
@when("I execute the porting task autonomously")
|
|
def step_execute_porting_task(context: Context) -> None:
|
|
"""Execute the full porting task through the autonomous pipeline."""
|
|
config = PortingTaskConfig(
|
|
source_module="legacy/calculator.py",
|
|
target_module="src/calculator.py",
|
|
add_tests=True,
|
|
project_name="test-porting-project",
|
|
)
|
|
context.porting_result = context.porting_service.execute_porting_task(
|
|
config=config,
|
|
source_content=context.py2_source,
|
|
)
|
|
|
|
|
|
@then("the porting result should indicate success")
|
|
def step_porting_result_success(context: Context) -> None:
|
|
assert context.porting_result is not None
|
|
assert context.porting_result.success, (
|
|
f"Porting task failed: {context.porting_result.error_message}"
|
|
)
|
|
|
|
|
|
@then('the lifecycle events should include a "{phase_name}" phase event')
|
|
def step_lifecycle_events_include_phase(context: Context, phase_name: str) -> None:
|
|
assert context.porting_result is not None
|
|
phases = [ev.phase for ev in context.porting_result.lifecycle_events]
|
|
assert any(phase_name in p for p in phases), (
|
|
f"No '{phase_name}' phase event found. Events: {phases}"
|
|
)
|
|
|
|
|
|
@then('the lifecycle events should include an "{phase_name}" phase event')
|
|
def step_lifecycle_events_include_phase_an(context: Context, phase_name: str) -> None:
|
|
step_lifecycle_events_include_phase(context, phase_name)
|
|
|
|
|
|
@then("the lifecycle events should not be empty")
|
|
def step_lifecycle_events_not_empty(context: Context) -> None:
|
|
assert context.porting_result is not None
|
|
assert len(context.porting_result.lifecycle_events) > 0, (
|
|
"No lifecycle events were recorded"
|
|
)
|
|
|
|
|
|
@then("each lifecycle event should have a phase")
|
|
def step_lifecycle_events_have_phase(context: Context) -> None:
|
|
assert context.porting_result is not None
|
|
for ev in context.porting_result.lifecycle_events:
|
|
assert ev.phase, f"Lifecycle event missing phase: {ev}"
|
|
|
|
|
|
@then("each lifecycle event should have a state")
|
|
def step_lifecycle_events_have_state(context: Context) -> None:
|
|
assert context.porting_result is not None
|
|
for ev in context.porting_result.lifecycle_events:
|
|
assert ev.state, f"Lifecycle event missing state: {ev}"
|
|
|
|
|
|
@then("each lifecycle event should have a timestamp")
|
|
def step_lifecycle_events_have_timestamp(context: Context) -> None:
|
|
assert context.porting_result is not None
|
|
for ev in context.porting_result.lifecycle_events:
|
|
assert ev.timestamp is not None, f"Lifecycle event missing timestamp: {ev}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Validation checks
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the porting result validation should pass")
|
|
def step_porting_validation_passed(context: Context) -> None:
|
|
assert context.porting_result is not None
|
|
assert context.porting_result.validation_passed, (
|
|
f"Validation failed: {context.porting_result.error_message}"
|
|
)
|
|
|
|
|
|
@then("the ported output should be valid Python 3")
|
|
def step_ported_output_valid_py3(context: Context) -> None:
|
|
assert context.porting_result is not None
|
|
content = context.porting_result.ported_output
|
|
assert content is not None, "No ported output available"
|
|
try:
|
|
compile(content, "<test>", "exec")
|
|
except SyntaxError as exc:
|
|
raise AssertionError(f"Ported output is not valid Python 3: {exc}") from exc
|
|
|
|
|
|
@then('the ported output should not contain "{text}"')
|
|
def step_ported_output_not_contain(context: Context, text: str) -> None:
|
|
assert context.porting_result is not None
|
|
content = context.porting_result.ported_output or ""
|
|
assert text not in content, f"Ported output still contains '{text}'"
|
|
|
|
|
|
@then("the ported output should contain type annotations")
|
|
def step_ported_output_has_annotations(context: Context) -> None:
|
|
assert context.porting_result is not None
|
|
content = context.porting_result.ported_output or ""
|
|
has_annotation = "->" in content or (": " in content and "def " in content)
|
|
assert has_annotation, "Ported output has no type annotations"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# validate_ported_output helper steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I validate ported output:")
|
|
def step_validate_ported_output(context: Context) -> None:
|
|
"""Validate the ported output from the docstring."""
|
|
content = context.text or ""
|
|
context.validation_passed, context.validation_errors = validate_ported_output(
|
|
content
|
|
)
|
|
|
|
|
|
@then("the porting validation should pass")
|
|
def step_porting_validation_passes(context: Context) -> None:
|
|
assert context.validation_passed is True, (
|
|
f"Validation failed with errors: {context.validation_errors}"
|
|
)
|
|
|
|
|
|
@then("there should be no porting validation errors")
|
|
def step_no_porting_validation_errors(context: Context) -> None:
|
|
assert len(context.validation_errors) == 0, (
|
|
f"Expected no errors, got: {context.validation_errors}"
|
|
)
|
|
|
|
|
|
@then("the porting validation should fail")
|
|
def step_porting_validation_fails(context: Context) -> None:
|
|
assert context.validation_passed is False, (
|
|
"Validation passed but was expected to fail"
|
|
)
|
|
|
|
|
|
@then('the validation errors should mention "{keyword}"')
|
|
def step_validation_errors_mention(context: Context, keyword: str) -> None:
|
|
errors_text = " ".join(context.validation_errors).lower()
|
|
assert keyword.lower() in errors_text, (
|
|
f"Validation errors do not mention '{keyword}'. "
|
|
f"Errors: {context.validation_errors}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stub porting transformation steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I apply stub porting to:")
|
|
def step_apply_stub_porting(context: Context) -> None:
|
|
"""Apply the stub porting transformation to the docstring."""
|
|
source = context.text or ""
|
|
context.stub_output = _apply_stub_porting(source)
|
|
|
|
|
|
@then('the stub output should contain "{text}"')
|
|
def step_stub_output_contains(context: Context, text: str) -> None:
|
|
assert text in context.stub_output, (
|
|
f"Stub output does not contain '{text}'.\nOutput:\n{context.stub_output}"
|
|
)
|
|
|
|
|
|
@then('the stub output should not contain "{text}"')
|
|
def step_stub_output_not_contains(context: Context, text: str) -> None:
|
|
assert text not in context.stub_output, (
|
|
f"Stub output still contains '{text}'.\nOutput:\n{context.stub_output}"
|
|
)
|