"""Step definitions for validation edge case scenarios. All step names are prefixed with 'validation edge' to avoid AmbiguousStep collisions with existing step definitions. """ from __future__ import annotations import json from pathlib import Path from typing import Any from behave import given, then, when # --------------------------------------------------------------------------- # Fixture directory # --------------------------------------------------------------------------- _FIXTURE_DIR = Path(__file__).resolve().parents[1] / "fixtures" / "validation" def _load_fixture(name: str) -> list[dict[str, Any]]: """Load a named fixture JSON file and return its fixture list.""" path = _FIXTURE_DIR / f"{name}.json" with open(path) as fh: data = json.load(fh) return data["fixtures"] def _find_case(fixtures: list[dict[str, Any]], case_name: str) -> dict[str, Any]: """Find a single test case by name within a fixture list.""" for case in fixtures: if case["name"] == case_name: return case msg = f"fixture case '{case_name}' not found" raise KeyError(msg) # --------------------------------------------------------------------------- # Validation helpers that operate on fixture data (no live services) # --------------------------------------------------------------------------- def _validate_tool_output(case: dict[str, Any]) -> str | None: """Simulate tool output validation and return error string or None.""" inp = case["input"] output = inp.get("output") if "tool_name" not in inp: return "missing required field: tool_name" if isinstance(output, str): return "tool output must be an object, got string" if not isinstance(output, dict): return f"tool output must be an object, got {type(output).__name__}" if not output: return "tool output must contain at least 'status' and 'result'" if "status" not in output: return "missing required field: status" if not isinstance(output.get("status"), str): actual_type = type(output["status"]).__name__ return f"invalid type for field 'status': expected string, got {actual_type}" if "result" in output and output["result"] is None: return "result payload must not be null" null_fields = [ k for k, v in output.items() if v is None and k not in ("status", "result") ] if null_fields: return f"unexpected null fields: {', '.join(null_fields)}" # Check nested types if result is a dict result = output.get("result") if isinstance(result, dict): type_errors = [] for key, val in result.items(): if key == "files_processed" and not isinstance(val, int): type_errors.append(f"{key} expected int") if key == "errors" and not isinstance(val, list): type_errors.append(f"{key} expected list") if type_errors: return f"nested type mismatch in result: {', '.join(type_errors)}" return None def _validate_resource_reference(case: dict[str, Any]) -> str | None: """Simulate resource reference validation and return error string or None.""" inp = case["input"] # Circular dependency check if "resources" in inp: adjacency: dict[str, list[str]] = {} for res in inp["resources"]: adjacency[res["name"]] = res.get("depends_on", []) def _has_cycle(node: str, visited: set[str], path: list[str]) -> str | None: if node in visited: cycle_start = path.index(node) cycle = [*path[cycle_start:], node] return f"circular dependency detected: {' -> '.join(cycle)}" visited.add(node) path.append(node) for dep in adjacency.get(node, []): result = _has_cycle(dep, visited, path) if result: return result path.pop() return None for node in adjacency: cycle_err = _has_cycle(node, set(), []) if cycle_err: return cycle_err # Dangling action reference if "action_ref" in inp: action = inp["action_ref"] if "nonexistent" in action: return f"action not found: {action}" # Unresolvable project path if "project_refs" in inp: for ref in inp["project_refs"]: if ".." in ref: return f"unresolvable project path: {ref}" # Missing actor for phase if "phase" in inp and "actor_config" in inp: phase = inp["phase"] actor_key = f"{phase}_actor" if actor_key not in inp["actor_config"]: return f"missing actor for phase '{phase}': {actor_key} not configured" # Dangling parent plan if "parent_plan_id" in inp: parent = inp["parent_plan_id"] if "missing" in parent: return f"parent plan not found: {parent}" # Resource type not registered if "resource_type" in inp: rtype = inp["resource_type"] if "unregistered" in rtype: return f"resource type not registered: {rtype}" return None def _simulate_timeout(case: dict[str, Any]) -> str | None: """Simulate timeout validation and return error string or None.""" inp = case["input"] validator = inp["validator_name"] timeout_ms = inp["timeout_ms"] if timeout_ms < 0: return f"invalid timeout: {timeout_ms}ms (must be non-negative)" simulated_duration = inp.get("simulated_duration_ms", 0) if timeout_ms == 0 or simulated_duration > timeout_ms: items_done = inp.get("items_before_timeout") total = inp.get("total_items") if items_done is not None and total is not None: return f"validation timeout: {validator} completed {items_done}/{total} items before deadline" return f"validation timeout: {validator} exceeded {timeout_ms}ms deadline" batch_size = inp.get("batch_size", 0) per_item = inp.get("simulated_per_item_ms", 0) if batch_size > 0 and (batch_size * per_item) > timeout_ms: return f"deadline exceeded: {validator} processing {batch_size} items" return None def _validate_transform(case: dict[str, Any]) -> str | None: """Simulate schema transform validation and return error string or None.""" inp = case["input"] transform_name = inp["transform_name"] result = inp["transform_result"] if result is None: return f"transform '{transform_name}' returned null schema" if not isinstance(result, dict): actual_type = type(result).__name__ return f"transform '{transform_name}' returned invalid schema: expected dict, got {actual_type}" if "type" not in result: return f"transform '{transform_name}' returned schema without 'type' field" valid_types = {"object", "array", "string", "integer", "number", "boolean", "null"} if result["type"] not in valid_types: return f"transform '{transform_name}' returned unknown schema type: {result['type']}" # Check for circular $ref for _key, val in result.get("properties", {}).items(): if isinstance(val, dict) and "$ref" in val: ref_target = val["$ref"] if ref_target.startswith("#/properties/"): return f"transform '{transform_name}' introduced circular reference in schema" # Check required array integrity if "required" in result: for elem in result["required"]: if not isinstance(elem, str): return ( f"transform '{transform_name}' corrupted 'required' array: " "elements must be strings" ) return None def _sort_validations(case: dict[str, Any]) -> tuple[list[str] | None, str | None]: """Sort validations by level priority and detect duplicates. Returns (sorted_ids, error_string). """ inp = case["input"] validations = inp["validations"] valid_levels = {"required", "informational"} # Check for unknown levels for v in validations: if v["level"] not in valid_levels: return ( None, f"unknown validation level: {v['level']} (expected: required, informational)", ) # Check for duplicate attachment IDs attachment_map: dict[str, str] = {} for v in validations: att_id = v.get("attachment_id") if att_id is not None: if att_id in attachment_map: first = attachment_map[att_id] return ( None, f"duplicate attachment ID: {att_id} (used by {first} and {v['id']})", ) attachment_map[att_id] = v["id"] # Sort: required first, then informational, preserving relative order required = [v for v in validations if v["level"] == "required"] informational = [v for v in validations if v["level"] == "informational"] sorted_ids = [v["id"] for v in required + informational] return sorted_ids, None # --------------------------------------------------------------------------- # Background # --------------------------------------------------------------------------- @given("a validation edge case test environment") def step_validation_edge_env(context: Any) -> None: """Set up an isolated validation edge case test environment.""" context.validation_edge_fixtures = None context.validation_edge_error = None context.validation_edge_sorted_ids = None context.validation_edge_expected_order = None context.validation_edge_rollback_count = 0 context.validation_edge_rollback_result = None context.validation_edge_conflict_detected = False # --------------------------------------------------------------------------- # Fixture loading # --------------------------------------------------------------------------- @given('validation edge fixture "{fixture_name}" is loaded') def step_validation_edge_load_fixture(context: Any, fixture_name: str) -> None: """Load a named fixture JSON from the validation fixtures directory.""" context.validation_edge_fixtures = _load_fixture(fixture_name) # --------------------------------------------------------------------------- # Malformed tool output steps # --------------------------------------------------------------------------- @when('validation edge processes fixture "{case_name}"') def step_validation_edge_process_fixture(context: Any, case_name: str) -> None: """Validate a malformed tool output fixture case.""" case = _find_case(context.validation_edge_fixtures, case_name) context.validation_edge_error = _validate_tool_output(case) # --------------------------------------------------------------------------- # Missing resource steps # --------------------------------------------------------------------------- @when('validation edge checks resource fixture "{case_name}"') def step_validation_edge_check_resource(context: Any, case_name: str) -> None: """Validate a missing resource fixture case.""" case = _find_case(context.validation_edge_fixtures, case_name) context.validation_edge_error = _validate_resource_reference(case) # --------------------------------------------------------------------------- # Timeout steps # --------------------------------------------------------------------------- @when('validation edge simulates timeout for fixture "{case_name}"') def step_validation_edge_simulate_timeout(context: Any, case_name: str) -> None: """Simulate a validation timeout for a fixture case.""" case = _find_case(context.validation_edge_fixtures, case_name) context.validation_edge_error = _simulate_timeout(case) # --------------------------------------------------------------------------- # Schema transform steps # --------------------------------------------------------------------------- @when('validation edge applies transform fixture "{case_name}"') def step_validation_edge_apply_transform(context: Any, case_name: str) -> None: """Validate a schema transform fixture case.""" case = _find_case(context.validation_edge_fixtures, case_name) context.validation_edge_error = _validate_transform(case) # --------------------------------------------------------------------------- # Ordering steps # --------------------------------------------------------------------------- @when('validation edge sorts ordering fixture "{case_name}"') def step_validation_edge_sort_ordering(context: Any, case_name: str) -> None: """Sort and validate ordering for a fixture case.""" case = _find_case(context.validation_edge_fixtures, case_name) sorted_ids, error = _sort_validations(case) context.validation_edge_sorted_ids = sorted_ids context.validation_edge_error = error if "expected_order" in case["input"]: context.validation_edge_expected_order = case["input"]["expected_order"] @then("validation edge ordering should match expected order") def step_validation_edge_ordering_matches(context: Any) -> None: """Assert the sorted validation IDs match the expected order.""" assert context.validation_edge_sorted_ids is not None, ( "sorted IDs should not be None" ) assert context.validation_edge_expected_order is not None, ( "expected order should not be None" ) assert ( context.validation_edge_sorted_ids == context.validation_edge_expected_order ), ( f"expected {context.validation_edge_expected_order}, " f"got {context.validation_edge_sorted_ids}" ) # --------------------------------------------------------------------------- # Concurrent validation steps # --------------------------------------------------------------------------- @given('validation edge has two concurrent validators for resource "{resource}"') def step_validation_edge_concurrent_validators(context: Any, resource: str) -> None: """Set up two concurrent validators targeting the same resource.""" context.validation_edge_resource = resource context.validation_edge_validator_a = { "name": "validator_a", "resource": resource, "version": 1, } context.validation_edge_validator_b = { "name": "validator_b", "resource": resource, "version": 1, } @when("validation edge runs both validators concurrently") def step_validation_edge_run_concurrent(context: Any) -> None: """Simulate running both validators and detecting a conflict.""" # Simulate: validator A bumps the version, then validator B tries to write context.validation_edge_validator_a["version"] = 2 if ( context.validation_edge_validator_b["version"] < context.validation_edge_validator_a["version"] ): context.validation_edge_conflict_detected = True @then("validation edge should detect a concurrent modification conflict") def step_validation_edge_conflict_detected(context: Any) -> None: """Assert that a concurrent modification conflict was detected.""" assert context.validation_edge_conflict_detected is True, ( "expected concurrent conflict" ) # --------------------------------------------------------------------------- # Rollback steps # --------------------------------------------------------------------------- @given("validation edge has a staged change set with {count:d} pending changes") def step_validation_edge_staged_changes(context: Any, count: int) -> None: """Stage a number of pending changes for rollback testing.""" context.validation_edge_pending_changes = [ {"id": f"change-{i}", "status": "pending"} for i in range(count) ] context.validation_edge_rollback_count = count @when("validation edge encounters a required validation failure") def step_validation_edge_required_failure(context: Any) -> None: """Simulate a required validation failure that triggers rollback.""" rolled = [] for change in context.validation_edge_pending_changes: change["status"] = "rolled back" rolled.append(change["id"]) context.validation_edge_rollback_result = ( f"rolled back {len(rolled)} changes: {', '.join(rolled)}" ) @then("validation edge should rollback all {count:d} pending changes") def step_validation_edge_rollback_count(context: Any, count: int) -> None: """Assert all pending changes were rolled back.""" rolled = [ c for c in context.validation_edge_pending_changes if c["status"] == "rolled back" ] assert len(rolled) == count, f"expected {count} rollbacks, got {len(rolled)}" @then('validation edge rollback result should contain "{text}"') def step_validation_edge_rollback_contains(context: Any, text: str) -> None: """Assert the rollback result message contains expected text.""" assert context.validation_edge_rollback_result is not None, ( "rollback result should not be None" ) assert text in context.validation_edge_rollback_result, ( f"expected '{text}' in '{context.validation_edge_rollback_result}'" ) # --------------------------------------------------------------------------- # Common error assertions # --------------------------------------------------------------------------- @then('validation edge should report error containing "{expected}"') def step_validation_edge_error_contains(context: Any, expected: str) -> None: """Assert the validation error contains the expected substring.""" assert context.validation_edge_error is not None, ( f"expected error containing '{expected}', but no error was reported" ) assert expected in context.validation_edge_error, ( f"expected '{expected}' in '{context.validation_edge_error}'" )