"""Step definitions for validation test fixtures.""" from __future__ import annotations from datetime import datetime from pathlib import Path from behave import given, then, when from behave.runner import Context from pydantic import ValidationError as PydanticValidationError from cleveragents.application.services.plan_service import PlanService from cleveragents.config.settings import Settings from cleveragents.core.exceptions import PlanError from cleveragents.domain.models.core import ( Change, OperationType, Project, ProjectSettings, ) from cleveragents.domain.models.core.action import ( ActionArgument, ArgumentRequirement, ArgumentType, ) # NOTE: Sections 1, 1b, and 2 (AST security validation, RestrictedPython, # Lambda AST validation) were removed as part of SEC1. eval()/exec() are no # longer used in stream_router.py; the SEC1 approach uses named operation and # transform registries instead. See features/security_eval.feature for the # replacement tests. # ────────────────────────────────────────────────── # Section 3: Python content sanitization # ────────────────────────────────────────────────── @given("I have a plan service for sanitization tests") def step_plan_service_for_sanitization(context: Context) -> None: settings = Settings() context.sanitize_service = PlanService( settings=settings, unit_of_work=None, # type: ignore[arg-type] ai_provider=None, ) @when('I sanitize the python content "{content}"') def step_sanitize_python_content(context: Context, content: str) -> None: result = context.sanitize_service._sanitize_python_content(content) context.sanitized_content = result[0] context.sanitized_error = result[1] context.sanitized_reason = result[2] @when("I sanitize python content containing a null byte") def step_sanitize_null_byte_content(context: Context) -> None: result = context.sanitize_service._sanitize_python_content("x = 1\x00") context.sanitized_content = result[0] context.sanitized_error = result[1] context.sanitized_reason = result[2] @when('I sanitize the python content with code fences wrapping "{inner}"') def step_sanitize_fenced_content(context: Context, inner: str) -> None: fenced = f"```python\n{inner}\n```" result = context.sanitize_service._sanitize_python_content(fenced) context.sanitized_content = result[0] context.sanitized_error = result[1] context.sanitized_reason = result[2] @then('the sanitized content should be "{expected}"') def step_check_sanitized_content(context: Context, expected: str) -> None: assert context.sanitized_content == expected, ( f"Expected '{expected}', got '{context.sanitized_content}'" ) @then("the sanitization error should be None") def step_check_sanitized_error_none(context: Context) -> None: assert context.sanitized_error is None, ( f"Expected no error, got: {context.sanitized_error}" ) @then("the sanitization error should not be None") def step_check_sanitized_error_not_none(context: Context) -> None: assert context.sanitized_error is not None, "Expected a SyntaxError" @then("the sanitization reason should be None") def step_check_sanitized_reason_none(context: Context) -> None: assert context.sanitized_reason is None, ( f"Expected no reason, got: {context.sanitized_reason}" ) @then('the sanitization reason should be "{expected}"') def step_check_sanitized_reason(context: Context, expected: str) -> None: assert context.sanitized_reason == expected, ( f"Expected reason '{expected}', got '{context.sanitized_reason}'" ) # ────────────────────────────────────────────────── # Section 4: Project model validation # ────────────────────────────────────────────────── @when('I try to create a project with name "{name}"') def step_try_create_project_invalid_name(context: Context, name: str) -> None: context.project_error = None context.created_project = None try: context.created_project = Project( id=None, name=name, path=Path("/tmp/test-project"), created_at=datetime.now(), updated_at=datetime.now(), current_plan_id=None, settings=ProjectSettings(), ) except (PydanticValidationError, ValueError) as exc: context.project_error = exc @then('a project validation error should be raised mentioning "{text}"') def step_check_project_validation_error(context: Context, text: str) -> None: assert context.project_error is not None, "Expected a validation error" assert text.lower() in str(context.project_error).lower(), ( f"Expected '{text}' in error: {context.project_error}" ) @when('I create a project fixture with name "{name}"') def step_create_project_valid_name(context: Context, name: str) -> None: context.created_project = Project( id=None, name=name, path=Path("/tmp/test-project"), created_at=datetime.now(), updated_at=datetime.now(), current_plan_id=None, settings=ProjectSettings(), ) @then("the project should be created with that name") def step_check_project_created_name(context: Context) -> None: assert context.created_project is not None assert context.created_project.name is not None @when('I create a project with relative path "{rel_path}"') def step_create_project_relative_path(context: Context, rel_path: str) -> None: context.created_project = Project( id=None, name="test-project", path=Path(rel_path), created_at=datetime.now(), updated_at=datetime.now(), current_plan_id=None, settings=ProjectSettings(), ) @then("the project fixture path should be absolute") def step_check_project_path_absolute(context: Context) -> None: assert context.created_project.path.is_absolute(), ( f"Expected absolute path, got: {context.created_project.path}" ) @when("I try to create a project with empty name") def step_try_create_project_empty_name(context: Context) -> None: context.project_error = None try: context.created_project = Project( id=None, name="", path=Path("/tmp/test-project"), created_at=datetime.now(), updated_at=datetime.now(), current_plan_id=None, settings=ProjectSettings(), ) except (PydanticValidationError, ValueError) as exc: context.project_error = exc @then("a project validation error should be raised") def step_check_project_validation_error_generic(context: Context) -> None: assert context.project_error is not None, "Expected a validation error" # ────────────────────────────────────────────────── # Section 5: Change list coercion # ────────────────────────────────────────────────── @given("I have a plan service for coercion tests") def step_plan_service_for_coercion(context: Context) -> None: settings = Settings() context.coerce_service = PlanService( settings=settings, unit_of_work=None, # type: ignore[arg-type] ai_provider=None, ) @when("I coerce an empty change list") def step_coerce_empty_list(context: Context) -> None: context.coerced_result = context.coerce_service._coerce_change_list( [], plan_name="test-plan", provider_name="test-provider", ) @then("the coerced result should be an empty list") def step_check_coerced_empty(context: Context) -> None: assert context.coerced_result == [], ( f"Expected empty list, got: {context.coerced_result}" ) @when("I coerce a list with one dict entry and one Change entry") def step_coerce_mixed_list(context: Context) -> None: dict_entry = { "id": None, "plan_id": 1, "file_path": "dict_file.py", "operation": OperationType.CREATE, "original_content": None, "new_content": "# from dict", "new_path": None, "applied": False, "applied_at": None, "created_at": datetime.now(), } change_entry = Change( id=None, plan_id=1, file_path="change_file.py", operation=OperationType.CREATE, original_content=None, new_content="# from Change", new_path=None, applied=False, applied_at=None, created_at=datetime.now(), ) context.coerced_result = context.coerce_service._coerce_change_list( [dict_entry, change_entry], plan_name="test-plan", provider_name="test-provider", ) @then("the coerced result should have {count:d} changes") def step_check_coerced_count(context: Context, count: int) -> None: assert len(context.coerced_result) == count, ( f"Expected {count} changes, got {len(context.coerced_result)}" ) @when("I try to coerce a non-list input") def step_coerce_non_list(context: Context) -> None: context.coerce_error = None try: context.coerce_service._coerce_change_list( {"not": "a list"}, # type: ignore[arg-type] plan_name="test-plan", provider_name="test-provider", ) except PlanError as exc: context.coerce_error = exc @then('a PlanError should be raised mentioning "{text}"') def step_check_coerce_plan_error(context: Context, text: str) -> None: assert context.coerce_error is not None, "Expected PlanError" assert isinstance(context.coerce_error, PlanError) assert text in str(context.coerce_error.message), ( f"Expected '{text}' in error: {context.coerce_error.message}" ) @when("I try to coerce a list containing an integer") def step_coerce_list_with_int(context: Context) -> None: context.coerce_error = None try: context.coerce_service._coerce_change_list( [42], plan_name="test-plan", provider_name="test-provider", ) except PlanError as exc: context.coerce_error = exc # ────────────────────────────────────────────────── # Section 6: ActionArgument parsing edge cases # ────────────────────────────────────────────────── @when('I try to parse action argument fixture "{arg_string}"') def step_try_parse_arg_fixture(context: Context, arg_string: str) -> None: context.arg_parse_error = None try: ActionArgument.parse(arg_string) except (ValueError, KeyError) as exc: context.arg_parse_error = exc @then("an argument fixture parse error should be raised") def step_check_arg_parse_error(context: Context) -> None: assert context.arg_parse_error is not None, "Expected a parse error" @when('I try to create argument with name "{name}"') def step_try_create_arg_with_name(context: Context, name: str) -> None: context.arg_create_error = None context.created_arg = None try: context.created_arg = ActionArgument( name=name, arg_type=ArgumentType.STRING, requirement=ArgumentRequirement.OPTIONAL, description="Test argument", ) except (PydanticValidationError, ValueError) as exc: context.arg_create_error = exc @then("the argument name should be accepted as valid identifier") def step_check_arg_name_accepted(context: Context) -> None: # "class" is a valid Python identifier (keyword but still passes str.isidentifier()) assert context.created_arg is not None, ( f"Expected argument creation to succeed, got error: {context.arg_create_error}" ) assert context.created_arg.name == "class"