diff --git a/yamlreaders/yaml_reader.py b/yamlreaders/yaml_reader.py new file mode 100644 index 0000000..f78e7ee --- /dev/null +++ b/yamlreaders/yaml_reader.py @@ -0,0 +1,108 @@ +from pathlib import Path + +from models import ( + AssertModel, + ExtractModel, + RequestModel, + StepModel, + TestModel, +) + +from .base import BaseReader, ParseError + + +class YamlReader(BaseReader): + def validate(self, source: str | Path) -> bool: + try: + data = self.load_yaml(source) + + if "steps" not in data: + return False + + if not isinstance(data["steps"], list): + return False + + for step in data["steps"]: + if "name" not in step or "request" not in step: + return False + + request = step["request"] + if "method" not in request: + return False + + return True + + except Exception: + return False + + def parse(self, source: str | Path) -> TestModel: + data = self.load_yaml(source) + + return TestModel( + name=data.get("name", "Unnamed Test"), + base_url=data.get("base_url", ""), + variables=data.get("variables", {}), + steps=self._parse_steps(data.get("steps", [])), + ) + + def _parse_steps(self, steps: list[dict]) -> list[StepModel]: + result = [] + + for i, step in enumerate(steps): + if "name" not in step: + raise ParseError(f"Step {i} missing 'name'") + if "request" not in step: + raise ParseError(f"Step {i} missing 'request'") + + result.append(StepModel( + name=step["name"], + request=self._parse_request(step["request"]), + extract=self._parse_extract(step.get("extract", [])), + assert_=self._parse_assert(step.get("assert", [])), + )) + + return result + + def _parse_request(self, request: dict) -> RequestModel: + return RequestModel( + method=request.get("method", "GET").upper(), + path=request.get("url", request.get("path", "/")), + headers=request.get("headers", {}), + query=request.get("query", {}), + body=request.get("body"), + body_type=request.get("body_type", "json"), + timeout=request.get("timeout", 30), + ) + + def _parse_extract(self, extract: list[dict]) -> list[ExtractModel]: + result = [] + + for item in extract: + if "name" not in item: + raise ParseError("Extract missing 'name'") + if "json_path" not in item: + raise ParseError(f"Extract '{item.get('name')}' missing 'json_path'") + + result.append(ExtractModel( + name=item["name"], + json_path=item["json_path"], + source=item.get("source", "body"), + )) + + return result + + def _parse_assert(self, asserts: list[dict]) -> list[AssertModel]: + result = [] + + for item in asserts: + if "type" not in item: + raise ParseError("Assert missing 'type'") + + result.append(AssertModel( + type=item["type"], + path=item.get("path"), + operator=item.get("operator", "equals"), + value=item.get("value"), + )) + + return result