81 lines
2.8 KiB
Python
81 lines
2.8 KiB
Python
"""HTTP API 的 Pydantic 契约。"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, List, Literal, Optional
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
class PreflightBody(BaseModel):
|
|
dut_id: Optional[str] = None
|
|
host_id: Optional[str] = None
|
|
firmware_id: Optional[str] = None
|
|
discovered_serial: Optional[str] = None
|
|
|
|
|
|
class CreateRunBody(BaseModel):
|
|
name: str = "固件 A/B 无人值守回归"
|
|
workflow_id: Optional[str] = None
|
|
dut_id: Optional[str] = None
|
|
host_id: Optional[str] = None
|
|
firmware_a_id: Optional[str] = None
|
|
firmware_b_id: Optional[str] = None
|
|
loops: int = Field(default=8, ge=1, le=500)
|
|
inject_failure: bool = True
|
|
execution_mode: Literal["simulated", "agent_dry_run"] = "simulated"
|
|
created_by: str = "console-demo"
|
|
|
|
|
|
class HumanActionBody(BaseModel):
|
|
actor: str = "console-user"
|
|
reason: str = "人工操作"
|
|
|
|
|
|
class AgentRegisterBody(BaseModel):
|
|
host_id: Optional[str] = None
|
|
host_name: Optional[str] = None
|
|
agent_version: str = Field(min_length=1, max_length=32)
|
|
os_family: str = Field(default="linux", min_length=1, max_length=32)
|
|
os_version: Optional[str] = Field(default=None, max_length=128)
|
|
kernel: Optional[str] = Field(default=None, max_length=128)
|
|
cpu: Optional[str] = Field(default=None, max_length=128)
|
|
motherboard: Optional[str] = Field(default=None, max_length=128)
|
|
bios_version: Optional[str] = Field(default=None, max_length=64)
|
|
ip_address: Optional[str] = Field(default=None, max_length=64)
|
|
toolchain: Dict[str, Any] = Field(default_factory=dict)
|
|
labels: Dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class AgentHeartbeatBody(BaseModel):
|
|
healthy: bool = True
|
|
agent_version: Optional[str] = Field(default=None, max_length=32)
|
|
ip_address: Optional[str] = Field(default=None, max_length=64)
|
|
toolchain: Dict[str, Any] = Field(default_factory=dict)
|
|
device_probe: Dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class AgentLeaseBody(BaseModel):
|
|
wait_timeout_s: float = Field(default=0.0, ge=0.0, le=20.0)
|
|
|
|
|
|
class AgentStepAttemptBody(BaseModel):
|
|
attempt: int = Field(ge=1)
|
|
|
|
|
|
class AgentStepEvent(BaseModel):
|
|
message: str = Field(min_length=1, max_length=2000)
|
|
severity: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = "INFO"
|
|
payload: Dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class AgentStepEventsBody(AgentStepAttemptBody):
|
|
events: List[AgentStepEvent] = Field(min_length=1, max_length=100)
|
|
|
|
|
|
class AgentStepCompleteBody(AgentStepAttemptBody):
|
|
status: Literal["SUCCEEDED", "FAILED"]
|
|
exit_code: Optional[int] = None
|
|
error_class: Optional[str] = Field(default=None, max_length=32)
|
|
result: Dict[str, Any] = Field(default_factory=dict)
|
|
log_uri: Optional[str] = Field(default=None, max_length=512)
|