feat(yamlreaders): Added OpenAPIReader for OpenAPI 3.0/3.1

Parses OpenAPI specifications:
- Extracts base_url from servers
- Parses paths and operations
- Generates request body from schema
- Auto-generates asserts from responses

yamlreaders/openapi_reader.py

@@ -0,0 +1,215 @@
+from pathlib import Path
+
+from models import (
+    AuthModel,
+    ExtractModel,
+    RequestModel,
+    StepModel,
+    TestModel,
+)
+
+from .base import BaseReader, ParseError
+
+
+class OpenAPIReader(BaseReader):
+    def validate(self, source: str | Path) -> bool:
+        try:
+            data = self.load_yaml(source)
+
+            if "openapi" not in data:
+                return False
+
+            version = data["openapi"]
+            if not version.startswith("3."):
+                return False
+
+            if "paths" not in data:
+                return False
+
+            return True
+
+        except Exception:
+            return False
+
+    def parse(self, source: str | Path) -> TestModel:
+        data = self.load_yaml(source)
+
+        info = data.get("info", {})
+        name = info.get("title", "Unnamed API")
+
+        base_url = self._extract_base_url(data)
+        auth = self._extract_auth(data)
+        steps = self._parse_paths(data.get("paths", {}))
+
+        return TestModel(
+            name=name,
+            base_url=base_url,
+            auth=auth,
+            steps=steps,
+        )
+
+    def _extract_base_url(self, data: dict) -> str:
+        servers = data.get("servers", [])
+
+        if servers:
+            url = servers[0].get("url", "")
+
+            if not url.startswith("http"):
+                url = f"http://localhost{url}" if url.startswith("/") else f"http://localhost/{url}"
+
+            return url.rstrip("/")
+
+        return "http://localhost"
+
+    def _extract_auth(self, data: dict) -> AuthModel | None:
+        security_schemes = data.get("components", {}).get("securitySchemes", {})
+
+        if not security_schemes:
+            return None
+
+        for scheme_name, scheme in security_schemes.items():
+            scheme_type = scheme.get("type", "")
+
+            if scheme_type == "http":
+                auth_type = scheme.get("scheme", "bearer")
+                return AuthModel(
+                    type=auth_type,
+                    credentials={"scheme": auth_type},
+                )
+
+            elif scheme_type == "apiKey":
+                return AuthModel(
+                    type="api_key",
+                    credentials={
+                        "in": scheme.get("in", "header"),
+                        "name": scheme.get("name", ""),
+                    },
+                )
+
+            elif scheme_type == "oauth2":
+                return AuthModel(
+                    type="oauth2",
+                    credentials={},
+                )
+
+        return None
+
+    def _parse_paths(self, paths: dict) -> list[StepModel]:
+        steps = []
+
+        for path, path_item in paths.items():
+            for method in ["get", "post", "put", "patch", "delete", "head", "options"]:
+                operation = path_item.get(method)
+                if operation:
+                    step = self._parse_operation(path, method, operation)
+                    if step:
+                        steps.append(step)
+
+        return steps
+
+    def _parse_operation(self, path: str, method: str, operation: dict) -> StepModel | None:
+        name = operation.get("operationId", f"{method.upper()} {path}")
+
+        request = self._build_request(path, method, operation)
+        extract = self._generate_extracts(operation)
+
+        return StepModel(
+            name=name,
+            request=request,
+            extract=extract,
+            assert_=self._generate_asserts(operation),
+        )
+
+    def _build_request(self, path: str, method: str, operation: dict) -> RequestModel:
+        headers = {}
+        query = {}
+        body = None
+
+        for param in operation.get("parameters", []):
+            param_in = param.get("in", "")
+            param_name = param.get("name", "")
+
+            if param_in == "header":
+                headers[param_name] = ""
+            elif param_in == "query":
+                query[param_name] = ""
+
+        request_body = operation.get("requestBody", {})
+        if request_body:
+            content = request_body.get("content", {})
+            json_content = content.get("application/json", {})
+            schema = json_content.get("schema", {})
+
+            if schema:
+                body = self._schema_to_example(schema)
+
+        return RequestModel(
+            method=method.upper(),
+            path=path,
+            headers=headers,
+            query=query,
+            body=body,
+        )
+
+    def _schema_to_example(self, schema: dict) -> dict:
+        if "example" in schema:
+            return schema["example"]
+
+        if schema.get("type") == "object":
+            properties = schema.get("properties", {})
+            example = {}
+
+            for prop_name, prop_schema in properties.items():
+                example[prop_name] = self._schema_to_example(prop_schema)
+
+            return example
+
+        if schema.get("type") == "string":
+            return ""
+
+        if schema.get("type") == "integer":
+            return 0
+
+        if schema.get("type") == "boolean":
+            return False
+
+        return {}
+
+    def _generate_extracts(self, operation: dict) -> list[ExtractModel]:
+        extracts = []
+
+        responses = operation.get("responses", {})
+        for status_code, response in responses.items():
+            if status_code.startswith("2"):
+                content = response.get("content", {})
+                json_content = content.get("application/json", {})
+                schema = json_content.get("schema", {})
+
+                if schema and schema.get("type") == "object":
+                    for prop_name in schema.get("properties", {}).keys():
+                        extracts.append(ExtractModel(
+                            name=f"{operation.get('operationId', 'response')}_{prop_name}",
+                            json_path=f"$.{prop_name}",
+                            source="body",
+                        ))
+
+                break
+
+        return extracts
+
+    def _generate_asserts(self, operation: dict) -> list:
+        from models import AssertModel
+
+        asserts = []
+
+        responses = operation.get("responses", {})
+        for status_code in responses.keys():
+            if status_code.startswith("2"):
+                asserts.append(AssertModel(
+                    type="status",
+                    operator="equals",
+                    value=int(status_code) if status_code.isdigit() else 200,
+                ))
+                break
+
+        return asserts