AI-Driven Remediation Testing

"""Pydantic models for scenario definition.""" from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field class Meta(BaseModel): """Scenario metadata.""" id: str = Field(..., description="Unique scenario identifier") title: str = Field(..., description="Scenario title") owner: str = Field(..., description="Scenario owner") class Defaults(BaseModel): """Default settings for scenario.""" model: Optional[str] = Field(default="gpt-4", description="Default AI model") timeout: Optional[int] = Field(default=300, description="Default timeout in seconds") policies: Optional[Dict[str, Any]] = Field( default_factory=dict, description="Default policies" ) class SignalFlowCheck(BaseModel): """SignalFlow-based check.""" program: str = Field(..., description="SignalFlow program") assert_rules: List[str] = Field( default_factory=list, description="Assertion rules" ) class Precheck(BaseModel): """Pre-execution checks.""" signalflow: Optional[List[SignalFlowCheck]] = Field( default_factory=list, description="SignalFlow checks" ) class Fault(BaseModel): """Fault injection definition.""" type: str = Field(..., description="Fault type (e.g., pod_kill, network_delay)") params: Dict[str, Any] = Field( default_factory=dict, description="Fault-specific parameters" ) class WaitFor(BaseModel): """Wait/polling conditions.""" detector: Optional[str] = Field(default=None, description="Detector ID to wait for") timeout: int = Field(default=300, description="Timeout in seconds") class Stabilize(BaseModel): """Stabilization period after fault injection.""" wait_for: WaitFor = Field(..., description="Wait conditions") class AssistantExpectation(BaseModel): """Expectations for assistant response.""" references: List[str] = Field( default_factory=list, description="Expected references" ) metrics: List[str] = Field(default_factory=list, description="Expected metrics") guards: Optional[List[Dict[str, str]]] = Field( default_factory=list, description="Validation guards (regex, json_schema)" ) class AssistantStep(BaseModel): """Assistant interaction step (RCA or Remedy).""" model: Optional[str] = Field(default=None, description="AI model override") system: str = Field(..., description="System prompt") user: str = Field(..., description="User prompt") expect: AssistantExpectation = Field(..., description="Expected response criteria") class Sandbox(BaseModel): """Execution sandbox configuration.""" service_account: Optional[str] = Field( default=None, description="Service account for execution" ) namespace: Optional[str] = Field(default=None, description="Kubernetes namespace") policies: Optional[Dict[str, List[str]]] = Field( default_factory=lambda: {"deny_patterns": []}, description="Execution policies", ) class Command(BaseModel): """Command to execute.""" name: str = Field(..., description="Command name/description") cmd: str = Field(..., description="Command to execute") args: List[str] = Field(default_factory=list, description="Command arguments") class ExecuteRemedy(BaseModel): """Remedy execution configuration.""" sandbox: Sandbox = Field(..., description="Execution sandbox") commands: List[Command] = Field(..., description="Commands to execute") class DetectorCheck(BaseModel): """Detector state check.""" detector_id: str = Field(..., description="Detector identifier") expected_state: str = Field(default="clear", description="Expected detector state") class Verify(BaseModel): """Post-execution verification.""" signalflow: Optional[List[SignalFlowCheck]] = Field( default_factory=list, description="SignalFlow checks" ) detector_clear: Optional[List[DetectorCheck]] = Field( default_factory=list, description="Detector clear checks" ) class CleanupCommand(BaseModel): """Cleanup command.""" name: str = Field(..., description="Cleanup task name") when_failed: bool = Field( default=False, description="Only run if scenario failed" ) cmd: str = Field(..., description="Command to execute") args: List[str] = Field(default_factory=list, description="Command arguments") class Report(BaseModel): """Report generation configuration.""" formats: List[str] = Field( default_factory=lambda: ["json"], description="Report formats (json, xml)" ) store: Optional[str] = Field(default=None, description="Storage location") class Scenario(BaseModel): """Complete scenario definition.""" meta: Meta = Field(..., description="Scenario metadata") defaults: Optional[Defaults] = Field( default_factory=Defaults, description="Default settings" ) bindings: Dict[str, str] = Field( default_factory=dict, description="Variable bindings" ) prechecks: Optional[Precheck] = Field( default_factory=Precheck, description="Pre-execution checks" ) fault: Fault = Field(..., description="Fault injection") stabilize: Stabilize = Field(..., description="Stabilization period") assistant_rca: AssistantStep = Field(..., description="RCA assistant interaction") assistant_remedy: AssistantStep = Field( ..., description="Remedy assistant interaction" ) execute_remedy: ExecuteRemedy = Field(..., description="Remedy execution") verify: Optional[Verify] = Field( default_factory=Verify, description="Post-execution verification" ) cleanup: Optional[List[CleanupCommand]] = Field( default_factory=list, description="Cleanup commands" ) report: Optional[Report] = Field( default_factory=Report, description="Report configuration" ) class Config: """Pydantic configuration.""" json_schema_extra = { "example": { "meta": { "id": "scenario-001", "title": "Pod Crash Recovery", "owner": "platform-team", }, "fault": {"type": "pod_kill", "params": {"namespace": "production"}}, } }