feat(yamlreaders): Added PostmanReader for Postman Collection v2.1

Parses Postman collections:
- Extracts variables and base_url
- Parses nested items and folders
- Converts request body and headers
- Extracts test scripts for variable extraction

yamlreaders/postman_reader.py

@@ -0,0 +1,211 @@
+from pathlib import Path
+
+import yaml
+
+from models import (
+    AuthModel,
+    ExtractModel,
+    RequestModel,
+    StepModel,
+    TestModel,
+)
+
+from .base import BaseReader, ParseError
+
+
+class PostmanReader(BaseReader):
+    def validate(self, source: str | Path) -> bool:
+        try:
+            data = self.load_yaml(source)
+
+            if "info" not in data:
+                return False
+
+            if "item" 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("name", "Unnamed Collection")
+
+        variables = self._extract_variables(data)
+        base_url = self._extract_base_url(data)
+        auth = self._extract_auth(data)
+
+        steps = self._parse_items(data.get("item", []))
+
+        return TestModel(
+            name=name,
+            base_url=base_url,
+            variables=variables,
+            auth=auth,
+            steps=steps,
+        )
+
+    def _extract_variables(self, data: dict) -> dict[str, str]:
+        variables = {}
+
+        for var in data.get("variable", []):
+            key = var.get("key", "")
+            value = var.get("value", "")
+            if key:
+                variables[key] = str(value)
+
+        return variables
+
+    def _extract_base_url(self, data: dict) -> str:
+        for var in data.get("variable", []):
+            if var.get("key") == "base_url":
+                return str(var.get("value", ""))
+
+        return ""
+
+    def _extract_auth(self, data: dict) -> AuthModel | None:
+        auth_data = data.get("auth")
+        if not auth_data:
+            return None
+
+        auth_type = auth_data.get("type", "none")
+        credentials = {}
+
+        for item in auth_data.get("basic", auth_data.get("apikey", [])):
+            key = item.get("key", "")
+            value = item.get("value", "")
+            credentials[key] = str(value)
+
+        return AuthModel(type=auth_type, credentials=credentials)
+
+    def _parse_items(self, items: list[dict]) -> list[StepModel]:
+        steps = []
+
+        for item in items:
+            if "item" in item:
+                steps.extend(self._parse_items(item["item"]))
+            elif "request" in item:
+                step = self._parse_item(item)
+                if step:
+                    steps.append(step)
+
+        return steps
+
+    def _parse_item(self, item: dict) -> StepModel | None:
+        name = item.get("name", "Unnamed Request")
+        request_data = item.get("request", {})
+
+        if isinstance(request_data, str):
+            request_data = {"method": "GET", "url": request_data}
+
+        request = self._parse_request(request_data)
+        extract = self._parse_events(item.get("event", []))
+
+        return StepModel(
+            name=name,
+            request=request,
+            extract=extract,
+            assert_=[],
+        )
+
+    def _parse_request(self, request: dict) -> RequestModel:
+        url = request.get("url", "")
+
+        if isinstance(url, dict):
+            raw = url.get("raw", "")
+            path_parts = url.get("path", [])
+            url = "/" + "/".join(path_parts) if path_parts else raw
+
+        headers = {}
+        for header in request.get("header", []):
+            key = header.get("key", "")
+            value = header.get("value", "")
+            if key:
+                headers[key] = value
+
+        body = None
+        body_type = "json"
+        body_data = request.get("body", {})
+
+        if body_data:
+            mode = body_data.get("mode", "raw")
+
+            if mode == "raw":
+                raw = body_data.get("raw", "")
+                try:
+                    import json
+                    body = json.loads(raw)
+                except Exception:
+                    body = raw
+            elif mode == "urlencoded":
+                body = {}
+                for item in body_data.get("urlencoded", []):
+                    body[item.get("key", "")] = item.get("value", "")
+            elif mode == "formdata":
+                body = {}
+                for item in body_data.get("formdata", []):
+                    body[item.get("key", "")] = item.get("value", "")
+
+        return RequestModel(
+            method=request.get("method", "GET").upper(),
+            path=url,
+            headers=headers,
+            body=body,
+            body_type=body_type,
+        )
+
+    def _parse_events(self, events: list[dict]) -> list[ExtractModel]:
+        extract = []
+
+        for event in events:
+            if event.get("listen") == "test":
+                script = event.get("script", {})
+                exec_lines = script.get("exec", [])
+
+                for line in exec_lines:
+                    extracted = self._parse_test_script(line)
+                    if extracted:
+                        extract.extend(extracted)
+
+        return extract
+
+    def _parse_test_script(self, line: str) -> list[ExtractModel] | None:
+        import re
+
+        patterns = [
+            r"pm\.environment\.set\(['\"](\w+)['\"],\s*(.+?)\)",
+            r"pm\.variables\.set\(['\"](\w+)['\"],\s*(.+?)\)",
+            r"pm\.collectionVariables\.set\(['\"](\w+)['\"],\s*(.+?)\)",
+        ]
+
+        result = []
+
+        for pattern in patterns:
+            match = re.search(pattern, line)
+            if match:
+                name = match.group(1)
+                json_path = match.group(2).strip()
+
+                if json_path.startswith("pm."):
+                    json_path = self._convert_pm_to_json_path(json_path)
+
+                result.append(ExtractModel(
+                    name=name,
+                    json_path=json_path,
+                    source="body",
+                ))
+
+        return result if result else None
+
+    def _convert_pm_to_json_path(self, pm_expr: str) -> str:
+        import re
+
+        match = re.match(r"pm\.response\.json\(\)\.(\w+)", pm_expr)
+        if match:
+            return f"$.{match.group(1)}"
+
+        return pm_expr