From d6b97e8b09210e52bca9e40913987c9e0408ca9c Mon Sep 17 00:00:00 2001 From: Kirill Date: Mon, 24 Aug 2026 16:03:32 +0300 Subject: [PATCH] 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 | 211 ++++++++++++++++++++++++++++++++++ 1 file changed, 211 insertions(+) create mode 100644 yamlreaders/postman_reader.py diff --git a/yamlreaders/postman_reader.py b/yamlreaders/postman_reader.py new file mode 100644 index 0000000..4fbdd4a --- /dev/null +++ b/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