Black_Raven
0f7dd83b33
Parses custom YAML test specifications: - Steps with requests - Extract variables from responses - Assert response status and fields
109 lines
3.2 KiB
Python
109 lines
3.2 KiB
Python
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
|