12.2AI-Powered Drift Detection

from dataclasses import dataclass

@dataclass
class Drift:
    type: str           # UNDOCUMENTED_FIELD, MISSING_FIELD, TYPE_MISMATCH, etc.
    field: str
    severity: str       # HIGH, MEDIUM, LOW
    message: str

class SchemaDriftDetector:
    """Detect differences between documented API schema and actual behavior."""

    def __init__(self, openapi_spec: dict, base_url: str):
        self.spec = openapi_spec
        self.base_url = base_url

    def detect_response_drift(self, endpoint: str, method: str) -> list[Drift]:
        """Compare actual API response against documented schema."""
        # Get documented schema
        documented = self.spec["paths"][endpoint][method]["responses"]["200"]
        documented_schema = documented["content"]["application/json"]["schema"]

        # Get actual response
        response = httpx.request(method.upper(), f"{self.base_url}{endpoint}")
        actual_body = response.json()

        drifts = []

        # Check for undocumented fields
        actual_fields = set(self._flatten_keys(actual_body))
        documented_fields = set(self._flatten_keys(documented_schema.get("properties", {})))

        undocumented = actual_fields - documented_fields
        for field in undocumented:
            drifts.append(Drift(
                type="UNDOCUMENTED_FIELD",
                field=field,
                severity="LOW",
                message=f"Field '{field}' present in response but not in schema"
            ))

        # Check for missing documented fields
        missing = documented_fields - actual_fields
        for field in missing:
            required = field in documented_schema.get("required", [])
            drifts.append(Drift(
                type="MISSING_FIELD",
                field=field,
                severity="HIGH" if required else "LOW",
                message=f"Field '{field}' documented but missing from response"
            ))

        # Check for type mismatches
        for field in actual_fields & documented_fields:
            actual_type = type(actual_body.get(field)).__name__
            doc_type = documented_schema["properties"].get(field, {}).get("type")
            if doc_type and not self._types_match(actual_type, doc_type):
                drifts.append(Drift(
                    type="TYPE_MISMATCH",
                    field=field,
                    severity="HIGH",
                    message=f"Field '{field}': documented as '{doc_type}', "
                            f"actual is '{actual_type}'"
                ))

        return drifts

    def _flatten_keys(self, obj, prefix="") -> list[str]:
        """Flatten a nested dict/schema into dot-notation keys."""
        keys = []
        if isinstance(obj, dict):
            for key in obj:
                full_key = f"{prefix}.{key}" if prefix else key
                keys.append(full_key)
        return keys

    def _types_match(self, python_type: str, openapi_type: str) -> bool:
        """Check if a Python type matches an OpenAPI type."""
        mapping = {
            "str": "string",
            "int": "integer",
            "float": "number",
            "bool": "boolean",
            "list": "array",
            "dict": "object",
            "NoneType": "null",
        }
        return mapping.get(python_type) == openapi_type