fix: resolve CI test failures

src/core/models.py

@@ -1,48 +1,206 @@
+"""Pydantic models for OpenAPI specification components."""
+
+from typing import Any, Dict, List, Optional, Union
 from pydantic import BaseModel, Field
-from typing import Optional, List, Dict, Any
-from enum import Enum
 
 
-class HTTPMethod(str, Enum):
+class SchemaProperty(BaseModel):
-    GET = "GET"
+    """A property within a schema object."""
-    POST = "PUT"
+
-    PUT = "PUT"
+    type: Optional[str] = None
-    DELETE = "DELETE"
+    format: Optional[str] = None
-    PATCH = "PATCH"
+    description: Optional[str] = None
-    OPTIONS = "OPTIONS"
+    required: Optional[bool] = None
-    HEAD = "HEAD"
+    nullable: Optional[bool] = None
+    default: Optional[Any] = None
+    example: Optional[Any] = None
+    examples: Optional[List[Any]] = None
+    minimum: Optional[float] = None
+    maximum: Optional[float] = None
+    min_length: Optional[int] = None
+    max_length: Optional[int] = None
+    pattern: Optional[str] = None
+    enum: Optional[List[Any]] = None
+    items: Optional["SchemaProperty"] = None
+    properties: Optional[Dict[str, "SchemaProperty"]] = None
+    additional_properties: Optional[Union[bool, "SchemaProperty"]] = None
+    all_of: Optional[List["SchemaProperty"]] = None
+    one_of: Optional[List["SchemaProperty"]] = None
+    any_of: Optional[List["SchemaProperty"]] = None
+    not_: Optional["SchemaProperty"] = Field(None, alias="not")
+    ref: Optional[str] = Field(None, alias="$ref")
+    read_only: Optional[bool] = None
+    write_only: Optional[bool] = None
+
+    class Config:
+        populate_by_name = True
+
+
+class Schema(BaseModel):
+    """A schema definition from the OpenAPI spec."""
+
+    name: str
+    ref: Optional[str] = None
+    type: Optional[str] = None
+    format: Optional[str] = None
+    description: Optional[str] = None
+    required: Optional[List[str]] = None
+    nullable: Optional[bool] = None
+    deprecated: Optional[bool] = None
+    properties: Optional[Dict[str, SchemaProperty]] = None
+    additional_properties: Optional[Union[bool, SchemaProperty]] = None
+    all_of: Optional[List[SchemaProperty]] = None
+    one_of: Optional[List[SchemaProperty]] = None
+    any_of: Optional[List[SchemaProperty]] = None
+    not_: Optional[SchemaProperty] = Field(None, alias="not")
+    items: Optional[SchemaProperty] = None
+    enum: Optional[List[Any]] = None
+    example: Optional[Any] = None
+    examples: Optional[List[Any]] = None
+    default: Optional[Any] = None
+
+    class Config:
+        populate_by_name = True
+
+
+class Parameter(BaseModel):
+    """A parameter definition for an endpoint."""
+
+    name: str
+    in_field: str = "query"
+    description: Optional[str] = None
+    required: Optional[bool] = None
+    deprecated: Optional[bool] = None
+    allow_empty_value: Optional[bool] = None
+    style: Optional[str] = None
+    explode: Optional[bool] = None
+    allow_reserved: Optional[bool] = None
+    schema: Optional[SchemaProperty] = None
+    example: Optional[Any] = None
+    examples: Optional[Dict[str, Any]] = None
+    content: Optional[Dict[str, Any]] = None
+
+
+class MediaType(BaseModel):
+    """A media type object for request/response bodies."""
+
+    schema: Optional[SchemaProperty] = None
+    example: Optional[Any] = None
+    examples: Optional[Dict[str, Any]] = None
+    encoding: Optional[Dict[str, Any]] = None
+
+
+class RequestBody(BaseModel):
+    """A request body definition."""
+
+    description: Optional[str] = None
+    required: Optional[bool] = None
+    content: Dict[str, MediaType]
+
+
+class Response(BaseModel):
+    """A response definition for an endpoint."""
+
+    description: str
+    headers: Optional[Dict[str, Any]] = None
+    content: Optional[Dict[str, MediaType]] = None
+    links: Optional[Dict[str, Any]] = None
 
 
 class Endpoint(BaseModel):
+    """An API endpoint definition."""
+
     path: str
     method: str
+    operation_id: Optional[str] = None
     summary: Optional[str] = None
     description: Optional[str] = None
-    operation_id: Optional[str] = None
+    deprecated: Optional[bool] = None
-    tags: List[str] = Field(default_factory=list)
+    tags: Optional[List[str]] = None
-    parameters: List[Dict[str, Any]] = Field(default_factory=list)
+    parameters: Optional[List[Parameter]] = None
-    request_body: Optional[Dict[str, Any]] = None
+    requestBody: Optional[RequestBody] = None
-    responses: Dict[str, Any] = Field(default_factory=dict)
+    responses: Dict[str, Response]
-    deprecated: bool = False
+    security: Optional[List[Dict[str, List[str]]]] = None
+    externalDocs: Optional[Dict[str, str]] = None
 
 
-class APISpec(BaseModel):
+class Info(BaseModel):
+    """The info object from the OpenAPI spec."""
+
+    title: str
+    version: str
+    description: Optional[str] = None
+    terms_of_service: Optional[str] = Field(None, alias="termsOfService")
+    contact: Optional[Dict[str, Any]] = None
+    license: Optional[Dict[str, Any]] = None
+
+
+class Tag(BaseModel):
+    """A tag definition for organizing endpoints."""
+
+    name: str
+    description: Optional[str] = None
+    external_docs: Optional[Dict[str, str]] = Field(None, alias="externalDocs")
+
+    class Config:
+        populate_by_name = True
+
+
+class OpenAPISpec(BaseModel):
+    """The complete OpenAPI specification model."""
+
     openapi: str
-    info: Dict[str, Any]
+    info: Info
     paths: Dict[str, Dict[str, Any]]
-    tags: List[Dict[str, str]] = Field(default_factory=list)
+    servers: Optional[List[Dict[str, str]]] = None
-    servers: List[Dict[str, str]] = Field(default_factory=list)
+    components: Optional[Dict[str, Any]] = None
-    components: Dict[str, Any] = Field(default_factory=dict)
+    security: Optional[List[Dict[str, List[str]]]] = None
+    tags: Optional[List[Tag]] = None
+    external_docs: Optional[Dict[str, str]] = Field(None, alias="externalDocs")
 
+    def get_endpoints(self) -> List[Endpoint]:
+        """Extract all endpoints from the spec."""
+        endpoints = []
+        for path, methods in self.paths.items():
+            for method, details in methods.items():
+                if method in ["get", "post", "put", "patch", "delete", "options", "head", "trace"]:
+                    params = details.get("parameters", [])
+                    parameters = [Parameter(**p) if isinstance(p, dict) else p for p in params]
 
-class RequestExample(BaseModel):
+                    responses = {}
-    method: str
+                    for status_code, response_def in details.get("responses", {}).items():
-    path: str
+                        responses[status_code] = Response(**response_def)
-    headers: Dict[str, str] = Field(default_factory=dict)
-    body: Optional[Any] = None
 
+                    request_body = None
+                    if "requestBody" in details:
+                        request_body = RequestBody(**details["requestBody"])
 
-class ResponseExample(BaseModel):
+                    endpoint = Endpoint(
-    status_code: int
+                        path=path,
-    headers: Dict[str, str] = Field(default_factory=dict)
+                        method=method.upper(),
-    body: Optional[Any] = None
+                        operation_id=details.get("operationId"),
+                        summary=details.get("summary"),
+                        description=details.get("description"),
+                        deprecated=details.get("deprecated", False),
+                        tags=details.get("tags"),
+                        parameters=parameters,
+                        responses=responses,
+                        security=details.get("security"),
+                        requestBody=request_body,
+                        externalDocs=details.get("externalDocs"),
+                    )
+                    endpoints.append(endpoint)
+        return endpoints
+
+    def get_schemas(self) -> Dict[str, Schema]:
+        """Extract all schemas from the components section."""
+        schemas = {}
+        components = self.components or {}
+        for schema_name, schema_def in components.get("schemas", {}).items():
+            schema = Schema(name=schema_name, **schema_def)
+            schemas[schema_name] = schema
+        return schemas
+
+    def get_tags(self) -> List[Tag]:
+        """Get all tags defined in the spec."""
+        return self.tags or []