Compare commits
10
Commits
0f7dd83b33
...
2e7c69c5af
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2e7c69c5af | ||
|
|
c26c60f679 | ||
|
|
e9cfdd9e94 | ||
|
|
6ae350b48c | ||
|
|
805b28e344 | ||
|
|
130d06351d | ||
|
|
2250961d09 | ||
|
|
bf47218831 | ||
|
|
f28c1d6d7c | ||
|
|
d6b97e8b09 |
No files matched your search
@@ -135,3 +135,315 @@ result = runner.run()
|
||||
|
||||
print(result.passed) # True/False
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
# YAML Readers (yamlreaders/)
|
||||
|
||||
## Обзор
|
||||
|
||||
Модуль для парсинга тест-спецификаций из различных форматов и конвертации их во внутреннюю модель `TestModel`.
|
||||
|
||||
Поддерживаемые форматы:
|
||||
- **Кастомный YAML** — собственный формат проекта
|
||||
- **Postman Collection v2.1** — экспорт из Postman
|
||||
- **OpenAPI 3.0/3.1** — спецификация API
|
||||
|
||||
## Архитектура
|
||||
|
||||
```
|
||||
┌─────────────────────────────────────────────────┐
|
||||
│ YAML Readers │
|
||||
│ ┌─────────────┐ ┌─────────────┐ ┌──────────┐│
|
||||
│ │ YamlReader │ │ PostmanReader│ │OpenAPIReader│
|
||||
│ └──────┬──────┘ └──────┬──────┘ └─────┬────┘│
|
||||
│ └────────────────┼────────────────┘ │
|
||||
│ ▼ │
|
||||
│ TestModel (ITM) │
|
||||
└─────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
## Классы
|
||||
|
||||
### BaseReader (yamlreaders/base.py)
|
||||
|
||||
Абстрактный базовый класс для всех парсеров.
|
||||
|
||||
| Метод | Возвращаемое значение | Описание |
|
||||
|-------|----------------------|----------|
|
||||
| validate(source) | bool | Проверяет валидность файла |
|
||||
| parse(source) | TestModel | Конвертирует файл в TestModel |
|
||||
| load_yaml(source) | dict | Загружает YAML файл |
|
||||
|
||||
**Параметры:**
|
||||
- `source: str | Path` — путь к YAML файлу
|
||||
|
||||
**Исключения:**
|
||||
- `FileNotFoundError` — файл не найден
|
||||
- `ParseError` — ошибка парсинга YAML
|
||||
|
||||
---
|
||||
|
||||
### YamlReader (yamlreaders/yaml_reader.py)
|
||||
|
||||
Парсер для кастомного формата YAML проекта.
|
||||
|
||||
**Формат входного файла:**
|
||||
|
||||
```yaml
|
||||
name: "Test Name"
|
||||
base_url: http://localhost:8080
|
||||
variables:
|
||||
email: test@example.com
|
||||
password: Test123!
|
||||
|
||||
steps:
|
||||
- name: "Step Name"
|
||||
request:
|
||||
method: POST
|
||||
url: /api/endpoint
|
||||
headers:
|
||||
Content-Type: application/json
|
||||
body:
|
||||
key: value
|
||||
timeout: 30
|
||||
extract:
|
||||
- name: variable_name
|
||||
json_path: "$.field"
|
||||
source: body
|
||||
assert:
|
||||
- type: status
|
||||
value: 200
|
||||
- type: json
|
||||
path: "$.id"
|
||||
operator: exists
|
||||
```
|
||||
|
||||
**Методы:**
|
||||
|
||||
| Метод | Возвращаемое значение | Описание |
|
||||
|-------|----------------------|----------|
|
||||
| validate(source) | bool | Проверяет наличие steps, request, method |
|
||||
| parse(source) | TestModel | Парсит YAML в TestModel |
|
||||
|
||||
**Пример использования:**
|
||||
|
||||
```python
|
||||
from yamlreaders import YamlReader
|
||||
|
||||
reader = YamlReader()
|
||||
|
||||
# Проверка валидности
|
||||
if reader.validate("tests/api.yaml"):
|
||||
model = reader.parse("tests/api.yaml")
|
||||
print(model.name, len(model.steps))
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### PostmanReader (yamlreaders/postman_reader.py)
|
||||
|
||||
Парсер для Postman Collection v2/v2.1.
|
||||
|
||||
**Формат входного файла (YAML):**
|
||||
|
||||
```yaml
|
||||
info:
|
||||
name: "Collection Name"
|
||||
schema: "https://schema.getpostman.com/json/collection/v2.1.0/collection.json"
|
||||
|
||||
variable:
|
||||
- key: base_url
|
||||
value: http://localhost:8080
|
||||
- key: email
|
||||
value: test@example.com
|
||||
|
||||
item:
|
||||
- name: "Request Name"
|
||||
request:
|
||||
method: POST
|
||||
url:
|
||||
raw: "{{base_url}}/api/login"
|
||||
host:
|
||||
- "{{base_url}}"
|
||||
path:
|
||||
- api
|
||||
- login
|
||||
header:
|
||||
- key: Content-Type
|
||||
value: application/json
|
||||
body:
|
||||
mode: raw
|
||||
raw: '{"email": "{{email}}"}'
|
||||
event:
|
||||
- listen: test
|
||||
script:
|
||||
exec:
|
||||
- "pm.environment.set('token', pm.response.json().token);"
|
||||
```
|
||||
|
||||
**Методы:**
|
||||
|
||||
| Метод | Возвращаемое значение | Описание |
|
||||
|-------|----------------------|----------|
|
||||
| validate(source) | bool | Проверяет наличие info и item |
|
||||
| parse(source) | TestModel | Парсит Postman Collection в TestModel |
|
||||
|
||||
**Особенности парсинга:**
|
||||
|
||||
| Функция | Описание |
|
||||
|---------|----------|
|
||||
| Extract variables | Извлекает переменные из `variable[]` |
|
||||
| Extract base_url | Ищет переменную `base_url` |
|
||||
| Parse nested items | Рекурсивно обходит вложенные папки |
|
||||
| Parse request body | Конвертирует `raw`, `urlencoded`, `formdata` |
|
||||
| Parse test scripts | Извлекает `pm.environment.set()` для extraction |
|
||||
|
||||
**Пример использования:**
|
||||
|
||||
```python
|
||||
from yamlreaders import PostmanReader
|
||||
|
||||
reader = PostmanReader()
|
||||
|
||||
if reader.validate("exports/collection.yaml"):
|
||||
model = reader.parse("exports/collection.yaml")
|
||||
print(model.name, len(model.steps))
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### OpenAPIReader (yamlreaders/openapi_reader.py)
|
||||
|
||||
Парсер для OpenAPI 3.0/3.1 спецификаций.
|
||||
|
||||
**Формат входного файла (YAML):**
|
||||
|
||||
```yaml
|
||||
openapi: "3.0.0"
|
||||
info:
|
||||
title: "API Name"
|
||||
version: "1.0.0"
|
||||
|
||||
servers:
|
||||
- url: http://localhost:8080
|
||||
|
||||
paths:
|
||||
/api/login:
|
||||
post:
|
||||
operationId: "loginUser"
|
||||
summary: "Login user"
|
||||
requestBody:
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
properties:
|
||||
email:
|
||||
type: string
|
||||
password:
|
||||
type: string
|
||||
responses:
|
||||
"200":
|
||||
description: "Success"
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
properties:
|
||||
token:
|
||||
type: string
|
||||
```
|
||||
|
||||
**Методы:**
|
||||
|
||||
| Метод | Возвращаемое значение | Описание |
|
||||
|-------|----------------------|----------|
|
||||
| validate(source) | bool | Проверяет наличие openapi и paths |
|
||||
| parse(source) | TestModel | Парсит OpenAPI в TestModel |
|
||||
|
||||
**Особенности парсинга:**
|
||||
|
||||
| Функция | Описание |
|
||||
|---------|----------|
|
||||
| Extract base_url | Берёт URL из `servers[0]` |
|
||||
| Extract auth | Парсит `securitySchemes` (basic, bearer, apiKey, oauth2) |
|
||||
| Parse paths | Конвертирует все methods (GET, POST, PUT...) в шаги |
|
||||
| Generate request body | Создаёт example body из `schema` |
|
||||
| Generate asserts | Автоматически добавляет проверку status code |
|
||||
| Generate extracts | Извлекает поля из response schema |
|
||||
|
||||
**Пример использования:**
|
||||
|
||||
```python
|
||||
from yamlreaders import OpenAPIReader
|
||||
|
||||
reader = OpenAPIReader()
|
||||
|
||||
if reader.validate("docs/openapi.yaml"):
|
||||
model = reader.parse("docs/openapi.yaml")
|
||||
print(model.name, len(model.steps))
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Автоопределение формата
|
||||
|
||||
Функция `get_reader()` автоматически определяет формат файла и возвращаетappropriate reader.
|
||||
|
||||
```python
|
||||
from yamlreaders import get_reader
|
||||
|
||||
reader = get_reader("tests/api.yaml")
|
||||
model = reader.parse("tests/api.yaml")
|
||||
```
|
||||
|
||||
**Логика определения:**
|
||||
|
||||
| Признак | Формат | Reader |
|
||||
|---------|--------|--------|
|
||||
| `openapi:` в первой строке | OpenAPI | OpenAPIReader |
|
||||
| `"info"` и `"item"` в JSON | Postman | PostmanReader |
|
||||
| Иначе | Кастомный YAML | YamlReader |
|
||||
|
||||
---
|
||||
|
||||
## Исключения
|
||||
|
||||
### ParseError
|
||||
|
||||
Выбрасывается при ошибке парсинга YAML.
|
||||
|
||||
```python
|
||||
from yamlreaders.base import ParseError
|
||||
|
||||
try:
|
||||
model = reader.parse("invalid.yaml")
|
||||
except ParseError as e:
|
||||
print(f"Error at line {e.line}: {e}")
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Полный пример
|
||||
|
||||
```python
|
||||
from yamlreaders import get_reader
|
||||
from models import Context, TestRunner
|
||||
|
||||
# Определяем формат и парсим
|
||||
reader = get_reader("tests/api.yaml")
|
||||
model = reader.parse("tests/api.yaml")
|
||||
|
||||
# Создаём контекст и запускаем
|
||||
context = Context(model)
|
||||
runner = TestRunner(context)
|
||||
result = runner.run()
|
||||
|
||||
# Выводим результат
|
||||
for step in result.steps:
|
||||
status = "✓" if step.passed else "✗"
|
||||
print(f"{status} {step.name}")
|
||||
|
||||
print(f"\n{'PASSED' if result.passed else 'FAILED'}")
|
||||
```
|
||||
@@ -0,0 +1,48 @@
|
||||
name: "Test Authorization"
|
||||
base_url: http://localhost:8080
|
||||
variables:
|
||||
email: test@example.com
|
||||
password: Test123!
|
||||
|
||||
steps:
|
||||
- name: "Register User"
|
||||
request:
|
||||
method: POST
|
||||
url: /api/register
|
||||
headers:
|
||||
Content-Type: application/json
|
||||
body:
|
||||
email: "{{email}}"
|
||||
password: "{{password}}"
|
||||
extract:
|
||||
- name: user_id
|
||||
json_path: "$.id"
|
||||
assert:
|
||||
- type: status
|
||||
value: 201
|
||||
|
||||
- name: "Login"
|
||||
request:
|
||||
method: POST
|
||||
url: /api/login
|
||||
headers:
|
||||
Content-Type: application/json
|
||||
body:
|
||||
email: "{{email}}"
|
||||
password: "{{password}}"
|
||||
extract:
|
||||
- name: token
|
||||
json_path: "$.token"
|
||||
assert:
|
||||
- type: status
|
||||
value: 200
|
||||
|
||||
- name: "Get Profile"
|
||||
request:
|
||||
method: GET
|
||||
url: /api/profile
|
||||
headers:
|
||||
Authorization: "Bearer {{token}}"
|
||||
assert:
|
||||
- type: status
|
||||
value: 200
|
||||
@@ -0,0 +1,50 @@
|
||||
openapi: "3.0.0"
|
||||
info:
|
||||
title: "Test API"
|
||||
version: "1.0.0"
|
||||
|
||||
servers:
|
||||
- url: http://localhost:8080
|
||||
|
||||
paths:
|
||||
/api/login:
|
||||
post:
|
||||
operationId: "loginUser"
|
||||
summary: "Login user"
|
||||
requestBody:
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
properties:
|
||||
email:
|
||||
type: string
|
||||
password:
|
||||
type: string
|
||||
responses:
|
||||
"200":
|
||||
description: "Success"
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
properties:
|
||||
token:
|
||||
type: string
|
||||
|
||||
/api/profile:
|
||||
get:
|
||||
operationId: "getProfile"
|
||||
summary: "Get user profile"
|
||||
responses:
|
||||
"200":
|
||||
description: "Success"
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
properties:
|
||||
id:
|
||||
type: integer
|
||||
email:
|
||||
type: string
|
||||
@@ -0,0 +1,52 @@
|
||||
info:
|
||||
name: "Test Collection"
|
||||
schema: "https://schema.getpostman.com/json/collection/v2.1.0/collection.json"
|
||||
|
||||
variable:
|
||||
- key: base_url
|
||||
value: http://localhost:8080
|
||||
- key: email
|
||||
value: test@example.com
|
||||
|
||||
item:
|
||||
- name: "Login"
|
||||
request:
|
||||
method: POST
|
||||
url:
|
||||
raw: "{{base_url}}/api/login"
|
||||
host:
|
||||
- "{{base_url}}"
|
||||
path:
|
||||
- api
|
||||
- login
|
||||
header:
|
||||
- key: Content-Type
|
||||
value: application/json
|
||||
body:
|
||||
mode: raw
|
||||
raw: '{"email": "{{email}}", "password": "Test123!"}'
|
||||
event:
|
||||
- listen: test
|
||||
script:
|
||||
exec:
|
||||
- "pm.test('status is 200', function() { pm.response.to.have.status(200); });"
|
||||
- "pm.environment.set('token', pm.response.json().token);"
|
||||
|
||||
- name: "Get Profile"
|
||||
request:
|
||||
method: GET
|
||||
url:
|
||||
raw: "{{base_url}}/api/profile"
|
||||
host:
|
||||
- "{{base_url}}"
|
||||
path:
|
||||
- api
|
||||
- profile
|
||||
header:
|
||||
- key: Authorization
|
||||
value: "Bearer {{token}}"
|
||||
event:
|
||||
- listen: test
|
||||
script:
|
||||
exec:
|
||||
- "pm.test('status is 200', function() { pm.response.to.have.status(200); });"
|
||||
@@ -0,0 +1,730 @@
|
||||
# API Checker
|
||||
|
||||
Инструмент для автоматизированного end-to-end тестирования API через GitHub Actions.
|
||||
|
||||
---
|
||||
|
||||
## Содержание
|
||||
|
||||
- [Архитектура](#архитектура)
|
||||
- [Models](#models)
|
||||
- [AuthModel](#authmodel)
|
||||
- [RequestModel](#requestmodel)
|
||||
- [ExtractModel](#extractmodel)
|
||||
- [AssertModel](#assertmodel)
|
||||
- [StepModel](#stepmodel)
|
||||
- [TestModel](#testmodel)
|
||||
- [Context](#context)
|
||||
- [Response](#response)
|
||||
- [Executor](#executor)
|
||||
- [Extractor](#extractor)
|
||||
- [Validator](#validator)
|
||||
- [StepResult](#stepresult)
|
||||
- [TestResult](#testresult)
|
||||
- [TestRunner](#testrunner)
|
||||
- [YAML Readers](#yaml-readers)
|
||||
- [ParseError](#parseerror)
|
||||
- [BaseReader](#basereader)
|
||||
- [YamlReader](#yamlreader)
|
||||
- [PostmanReader](#postmanreader)
|
||||
- [OpenAPIReader](#openapireader)
|
||||
- [get_reader](#get_reader)
|
||||
- [Примеры использования](#примеры-использования)
|
||||
|
||||
---
|
||||
|
||||
## Архитектура
|
||||
|
||||
```
|
||||
YAML/OpenAPI/Postman → Parser → TestModel → Context → Runner → TestResult
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Models
|
||||
|
||||
### AuthModel
|
||||
|
||||
**Файл:** `models/itm.py`
|
||||
|
||||
Модель аутентификации.
|
||||
|
||||
| Поле | Тип | По умолчанию | Описание |
|
||||
|------|-----|--------------|----------|
|
||||
| `type` | `str` | `"none"` | Тип аутентификации: `none`, `basic`, `bearer`, `api_key`, `oauth2` |
|
||||
| `credentials` | `dict[str, str]` | `{}` | Данные для аутентификации |
|
||||
|
||||
**Когда использовать:** Когда API требует аутентификацию. Передаётся в `TestModel.auth`.
|
||||
|
||||
---
|
||||
|
||||
### RequestModel
|
||||
|
||||
**Файл:** `models/itm.py`
|
||||
|
||||
Модель HTTP запроса.
|
||||
|
||||
| Поле | Тип | По умолчанию | Описание |
|
||||
|------|-----|--------------|----------|
|
||||
| `method` | `str` | `"GET"` | HTTP метод: `GET`, `POST`, `PUT`, `PATCH`, `DELETE`, `HEAD`, `OPTIONS` |
|
||||
| `path` | `str` | `"/"` | Путь URL (поддерживает `{{variable}}`) |
|
||||
| `headers` | `dict[str, str]` | `{}` | HTTP заголовки |
|
||||
| `query` | `dict[str, str]` | `{}` | Query параметры |
|
||||
| `body` | `dict \| str \| None` | `None` | Тело запроса |
|
||||
| `body_type` | `str` | `"json"` | Тип тела: `json`, `form`, `raw` |
|
||||
| `timeout` | `int` | `30` | Таймаут в секундах |
|
||||
|
||||
**Когда использовать:** Определяет что отправить серверу. Передаётся в `StepModel.request`.
|
||||
|
||||
---
|
||||
|
||||
### ExtractModel
|
||||
|
||||
**Файл:** `models/itm.py`
|
||||
|
||||
Модель извлечения данных из ответа.
|
||||
|
||||
| Поле | Тип | По умолчанию | Описание |
|
||||
|------|-----|--------------|----------|
|
||||
| `name` | `str` | — | Имя переменной (обязательное) |
|
||||
| `json_path` | `str` | — | JSON-путь к значению: `$.token`, `$.user.id` (обязательное) |
|
||||
| `source` | `str` | `"body"` | Источник: `body`, `header`, `status` |
|
||||
|
||||
**Когда использовать:** Когда нужно сохранить данные из ответа для использования в следующих запросах. Передаётся в `StepModel.extract`.
|
||||
|
||||
---
|
||||
|
||||
### AssertModel
|
||||
|
||||
**Файл:** `models/itm.py`
|
||||
|
||||
Модель проверки ответа.
|
||||
|
||||
| Поле | Тип | По умолчанию | Описание |
|
||||
|------|-----|--------------|----------|
|
||||
| `type` | `str` | — | Тип проверки: `status`, `json`, `header` (обязательное) |
|
||||
| `path` | `str \| None` | `None` | JSON-путь к полю для проверки |
|
||||
| `operator` | `str` | `"equals"` | Оператор: `equals`, `exists`, `gt`, `lt`, `contains`, `matches` |
|
||||
| `value` | `Any` | `None` | Ожидаемое значение |
|
||||
|
||||
**Когда использовать:** Когда нужно проверить что ответ соответствует ожиданиям. Передаётся в `StepModel.assert_`.
|
||||
|
||||
---
|
||||
|
||||
### StepModel
|
||||
|
||||
**Файл:** `models/itm.py`
|
||||
|
||||
Модель одного шага теста.
|
||||
|
||||
| Поле | Тип | По умолчанию | Описание |
|
||||
|------|-----|--------------|----------|
|
||||
| `name` | `str` | — | Название шага (обязательное) |
|
||||
| `request` | `RequestModel` | — | HTTP запрос (обязательное) |
|
||||
| `extract` | `list[ExtractModel]` | `[]` | Извлечение данных |
|
||||
| `assert_` | `list[AssertModel]` | `[]` | Проверки |
|
||||
|
||||
**Когда использовать:** Один логический шаг теста (например, "Залогиниться"). Передаётся в `TestModel.steps`.
|
||||
|
||||
---
|
||||
|
||||
### TestModel
|
||||
|
||||
**Файл:** `models/itm.py`
|
||||
|
||||
Основная модель теста. Содержит всю конфигурацию.
|
||||
|
||||
| Поле | Тип | По умолчанию | Описание |
|
||||
|------|-----|--------------|----------|
|
||||
| `name` | `str` | `"Unnamed Test"` | Название теста |
|
||||
| `base_url` | `str` | `""` | Базовый URL API |
|
||||
| `variables` | `dict[str, str]` | `{}` | Статические переменные |
|
||||
| `auth` | `AuthModel \| None` | `None` | Аутентификация |
|
||||
| `steps` | `list[StepModel]` | `[]` | Шаги теста |
|
||||
|
||||
**Когда использовать:** Главная входная точка. Создаётся парсерами или вручную.
|
||||
|
||||
---
|
||||
|
||||
### Context
|
||||
|
||||
**Файл:** `models/context.py`
|
||||
|
||||
Контекст выполнения. Хранит состояние во время запуска теста.
|
||||
|
||||
| Поле | Тип | Описание |
|
||||
|------|-----|----------|
|
||||
| `config` | `TestModel` | Оригинальная конфигурация теста |
|
||||
| `variables` | `dict[str, str]` | Статические переменные из config |
|
||||
| `extracted` | `dict[str, Any]` | Извлечённые данные из ответов |
|
||||
|
||||
**Методы:**
|
||||
|
||||
| Метод | Вход | Возврат | Описание |
|
||||
|-------|------|---------|----------|
|
||||
| `__init__(config)` | `TestModel` | — | Создаёт контекст из конфигурации |
|
||||
| `resolve(template)` | `str` | `str` | Заменяет `{{var}}` на значение из variables или extracted |
|
||||
| `set_extracted(name, value)` | `str, Any` | — | Сохраняет извлечённое значение |
|
||||
|
||||
**Когда использовать:** Передаётся всем компонентам (Executor, Extractor, TestRunner) для доступа к данным.
|
||||
|
||||
**Пример:**
|
||||
|
||||
```python
|
||||
context = Context(TestModel(variables={"email": "test@x.com"}))
|
||||
context.set_extracted("token", "abc123")
|
||||
|
||||
context.resolve("{{email}}") # "test@x.com"
|
||||
context.resolve("Bearer {{token}}") # "Bearer abc123"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Response
|
||||
|
||||
**Файл:** `models/executor.py`
|
||||
|
||||
Модель HTTP ответа.
|
||||
|
||||
| Поле | Тип | Описание |
|
||||
|------|-----|----------|
|
||||
| `status` | `int` | HTTP статус-код (200, 404, ...) |
|
||||
| `body` | `dict` | Тело ответа (JSON) |
|
||||
| `headers` | `dict` | Заголовки ответа |
|
||||
|
||||
**Когда использовать:** Создаётся Executor после выполнения запроса. Передаётся в Extractor и Validator.
|
||||
|
||||
---
|
||||
|
||||
### Executor
|
||||
|
||||
**Файл:** `models/executor.py`
|
||||
|
||||
Выполняет HTTP запросы.
|
||||
|
||||
| Метод | Вход | Возврат | Описание |
|
||||
|-------|------|---------|----------|
|
||||
| `__init__(context)` | `Context` | — | Создаёт executor с контекстом |
|
||||
| `_resolve_url(request)` | `RequestModel` | `str` | Разрешает URL с переменными |
|
||||
| `execute(request)` | `RequestModel` | `Response` | Выполняет HTTP запрос |
|
||||
|
||||
**Когда использовать:** Отправляет запросы к API. Используется TestRunner.
|
||||
|
||||
**Пример:**
|
||||
|
||||
```python
|
||||
context = Context(model)
|
||||
executor = Executor(context)
|
||||
response = executor.execute(RequestModel(method="GET", path="/api/users"))
|
||||
print(response.status) # 200
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Extractor
|
||||
|
||||
**Файл:** `models/extractor.py`
|
||||
|
||||
Извлекает данные из HTTP ответов.
|
||||
|
||||
| Метод | Вход | Возврат | Описание |
|
||||
|-------|------|---------|----------|
|
||||
| `__init__(context)` | `Context` | — | Создаёт extractor с контекстом |
|
||||
| `extract(configs, response)` | `list[ExtractModel], Response` | — | Извлекает данные в context.extracted |
|
||||
|
||||
**Внутренние методы:**
|
||||
|
||||
| Метод | Вход | Возврат | Описание |
|
||||
|-------|------|---------|----------|
|
||||
| `_get_value(response, config)` | `Response, ExtractModel` | `Any` | Получает значение из ответа |
|
||||
| `_json_path(data, path)` | `dict, str` | `Any` | Извлекает значение по JSON-пути |
|
||||
|
||||
**Когда использовать:** Когда нужно сохранить данные из ответа (токен, ID, ...) для использования в следующих шагах.
|
||||
|
||||
**Пример:**
|
||||
|
||||
```python
|
||||
extractor = Extractor(context)
|
||||
response = Response(status=200, body={"token": "abc123"}, headers={})
|
||||
extractor.extract([ExtractModel(name="token", json_path="$.token")], response)
|
||||
|
||||
print(context.extracted["token"]) # "abc123"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Validator
|
||||
|
||||
**Файл:** `models/validator.py`
|
||||
|
||||
Проверяет HTTP ответы.
|
||||
|
||||
| Метод | Вход | Возврат | Описание |
|
||||
|-------|------|---------|----------|
|
||||
| `validate(asserts, response)` | `list[AssertModel], Response` | `bool` | Все ли проверки прошли |
|
||||
|
||||
**Внутренние методы:**
|
||||
|
||||
| Метод | Вход | Возврат | Описание |
|
||||
|-------|------|---------|----------|
|
||||
| `_check(assertion, response)` | `AssertModel, Response` | `bool` | Проверяет одно утверждение |
|
||||
|
||||
**Когда использовать:** Проверяет что ответ соответствует ожиданиям. Используется TestRunner.
|
||||
|
||||
**Пример:**
|
||||
|
||||
```python
|
||||
validator = Validator()
|
||||
response = Response(status=200, body={}, headers={})
|
||||
asserts = [AssertModel(type="status", value=200)]
|
||||
|
||||
print(validator.validate(asserts, response)) # True
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### StepResult
|
||||
|
||||
**Файл:** `models/runner.py`
|
||||
|
||||
Результат выполнения одного шага.
|
||||
|
||||
| Поле | Тип | Описание |
|
||||
|------|-----|----------|
|
||||
| `name` | `str` | Название шага |
|
||||
| `passed` | `bool` | Прошёл ли шаг |
|
||||
| `response` | `Response` | HTTP ответ |
|
||||
|
||||
**Когда использовать:** Создаётся TestRunner после выполнения шага. Возвращается в TestResult.
|
||||
|
||||
---
|
||||
|
||||
### TestResult
|
||||
|
||||
**Файл:** `models/runner.py`
|
||||
|
||||
Результат выполнения всего теста.
|
||||
|
||||
| Поле | Тип | Описание |
|
||||
|------|-----|----------|
|
||||
| `steps` | `list[StepResult]` | Результаты всех шагов |
|
||||
|
||||
**Свойства:**
|
||||
|
||||
| Свойство | Тип | Описание |
|
||||
|----------|-----|----------|
|
||||
| `passed` | `bool` | True если все шаги прошли |
|
||||
|
||||
**Когда использовать:** Возвращается TestRunner.run(). Используется для вывода результатов.
|
||||
|
||||
---
|
||||
|
||||
### TestRunner
|
||||
|
||||
**Файл:** `models/runner.py`
|
||||
|
||||
Оркестратор выполнения теста.
|
||||
|
||||
| Метод | Вход | Возврат | Описание |
|
||||
|-------|------|---------|----------|
|
||||
| `__init__(context)` | `Context` | — | Создаёт runner с контекстом |
|
||||
| `run()` | — | `TestResult` | Запускает все шаги теста |
|
||||
|
||||
**Когда использовать:** Главная точка входа для запуска теста.
|
||||
|
||||
**Пример:**
|
||||
|
||||
```python
|
||||
context = Context(model)
|
||||
runner = TestRunner(context)
|
||||
result = runner.run()
|
||||
|
||||
print(result.passed) # True/False
|
||||
for step in result.steps:
|
||||
print(f"{step.name}: {'OK' if step.passed else 'FAIL'}")
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## YAML Readers
|
||||
|
||||
### ParseError
|
||||
|
||||
**Файл:** `yamlreaders/base.py`
|
||||
|
||||
Исключение при ошибке парсинга.
|
||||
|
||||
| Поле | Тип | Описание |
|
||||
|------|-----|----------|
|
||||
| `line` | `int \| None` | Номер строки с ошибкой |
|
||||
|
||||
**Когда использовать:** Ловить при парсинге невалидных YAML файлов.
|
||||
|
||||
---
|
||||
|
||||
### BaseReader
|
||||
|
||||
**Файл:** `yamlreaders/base.py`
|
||||
|
||||
Абстрактный базовый класс для всех парсеров.
|
||||
|
||||
| Метод | Вход | Возврат | Описание |
|
||||
|-------|------|---------|----------|
|
||||
| `validate(source)` | `str \| Path` | `bool` | Проверяет валидность файла |
|
||||
| `parse(source)` | `str \| Path` | `TestModel` | Конвертирует файл в TestModel |
|
||||
| `load_yaml(source)` | `str \| Path` | `dict` | Загружает YAML файл |
|
||||
|
||||
**Исключения:**
|
||||
- `FileNotFoundError` — файл не найден
|
||||
- `ParseError` — ошибка парсинга
|
||||
|
||||
**Когда использовать:** Не используется напрямую. Используйте наследников (YamlReader, PostmanReader, OpenAPIReader).
|
||||
|
||||
---
|
||||
|
||||
### YamlReader
|
||||
|
||||
**Файл:** `yamlreaders/yaml_reader.py`
|
||||
|
||||
Парсер для кастомного формата YAML.
|
||||
|
||||
| Метод | Вход | Возврат | Описание |
|
||||
|-------|------|---------|----------|
|
||||
| `validate(source)` | `str \| Path` | `bool` | Проверяет наличие steps, request, method |
|
||||
| `parse(source)` | `str \| Path` | `TestModel` | Парсит YAML в TestModel |
|
||||
|
||||
**Внутренние методы:**
|
||||
|
||||
| Метод | Вход | Возврат | Описание |
|
||||
|-------|------|---------|----------|
|
||||
| `_parse_steps(steps)` | `list[dict]` | `list[StepModel]` | Парсит шаги |
|
||||
| `_parse_request(request)` | `dict` | `RequestModel` | Парсит запрос |
|
||||
| `_parse_extract(extract)` | `list[dict]` | `list[ExtractModel]` | Парсит извлечение |
|
||||
| `_parse_assert(asserts)` | `list[dict]` | `list[AssertModel]` | Парсит проверки |
|
||||
|
||||
**Формат YAML:**
|
||||
|
||||
```yaml
|
||||
name: "Test Name"
|
||||
base_url: http://localhost:8080
|
||||
variables:
|
||||
email: test@example.com
|
||||
|
||||
steps:
|
||||
- name: "Login"
|
||||
request:
|
||||
method: POST
|
||||
url: /api/login
|
||||
headers:
|
||||
Content-Type: application/json
|
||||
body:
|
||||
email: "{{email}}"
|
||||
extract:
|
||||
- name: token
|
||||
json_path: "$.token"
|
||||
assert:
|
||||
- type: status
|
||||
value: 200
|
||||
```
|
||||
|
||||
**Когда использовать:** Для собственных тест-спецификаций проекта.
|
||||
|
||||
---
|
||||
|
||||
### PostmanReader
|
||||
|
||||
**Файл:** `yamlreaders/postman_reader.py`
|
||||
|
||||
Парсер для Postman Collection v2/v2.1.
|
||||
|
||||
| Метод | Вход | Возврат | Описание |
|
||||
|-------|------|---------|----------|
|
||||
| `validate(source)` | `str \| Path` | `bool` | Проверяет наличие info и item |
|
||||
| `parse(source)` | `str \| Path` | `TestModel` | Парсит Postman Collection в TestModel |
|
||||
|
||||
**Внутренние методы:**
|
||||
|
||||
| Метод | Вход | Возврат | Описание |
|
||||
|-------|------|---------|----------|
|
||||
| `_extract_variables(data)` | `dict` | `dict[str, str]` | Извлекает переменные |
|
||||
| `_extract_base_url(data)` | `dict` | `str` | Ищет переменную base_url |
|
||||
| `_extract_auth(data)` | `dict` | `AuthModel \| None` | Парсит аутентификацию |
|
||||
| `_parse_items(items)` | `list[dict]` | `list[StepModel]` | Рекурсивно парсит items |
|
||||
| `_parse_item(item)` | `dict` | `StepModel \| None` | Парсит один request |
|
||||
| `_parse_request(request)` | `dict` | `RequestModel` | Парсит HTTP запрос |
|
||||
| `_parse_events(events)` | `list[dict]` | `list[ExtractModel]` | Парсит test scripts |
|
||||
| `_parse_test_script(line)` | `str` | `list[ExtractModel] \| None` | Извлекает pm.environment.set() |
|
||||
| `_convert_pm_to_json_path(pm_expr)` | `str` | `str` | Конвертирует pm表达式 в JSON-путь |
|
||||
|
||||
**Формат YAML:**
|
||||
|
||||
```yaml
|
||||
info:
|
||||
name: "Collection Name"
|
||||
|
||||
variable:
|
||||
- key: base_url
|
||||
value: http://localhost:8080
|
||||
|
||||
item:
|
||||
- name: "Login"
|
||||
request:
|
||||
method: POST
|
||||
url:
|
||||
raw: "{{base_url}}/api/login"
|
||||
path:
|
||||
- api
|
||||
- login
|
||||
header:
|
||||
- key: Content-Type
|
||||
value: application/json
|
||||
body:
|
||||
mode: raw
|
||||
raw: '{"email": "{{email}}"}'
|
||||
event:
|
||||
- listen: test
|
||||
script:
|
||||
exec:
|
||||
- "pm.environment.set('token', pm.response.json().token);"
|
||||
```
|
||||
|
||||
**Когда использовать:** Для импорта коллекций из Postman.
|
||||
|
||||
---
|
||||
|
||||
### OpenAPIReader
|
||||
|
||||
**Файл:** `yamlreaders/openapi_reader.py`
|
||||
|
||||
Парсер для OpenAPI 3.0/3.1 спецификаций.
|
||||
|
||||
| Метод | Вход | Возврат | Описание |
|
||||
|-------|------|---------|----------|
|
||||
| `validate(source)` | `str \| Path` | `bool` | Проверяет наличие openapi и paths |
|
||||
| `parse(source)` | `str \| Path` | `TestModel` | Парсит OpenAPI в TestModel |
|
||||
|
||||
**Внутренние методы:**
|
||||
|
||||
| Метод | Вход | Возврат | Описание |
|
||||
|-------|------|---------|----------|
|
||||
| `_extract_base_url(data)` | `dict` | `str` | Берёт URL из servers |
|
||||
| `_extract_auth(data)` | `dict` | `AuthModel \| None` | Парсит securitySchemes |
|
||||
| `_parse_paths(paths)` | `dict` | `list[StepModel]` | Конвертирует paths в шаги |
|
||||
| `_parse_operation(path, method, operation)` | `str, str, dict` | `StepModel` | Парсит один operation |
|
||||
| `_build_request(path, method, operation)` | `str, str, dict` | `RequestModel` | Создаёт RequestModel |
|
||||
| `_schema_to_example(schema)` | `dict` | `dict` | Создаёт example из schema |
|
||||
| `_generate_extracts(operation)` | `dict` | `list[ExtractModel]` | Генерирует extraction |
|
||||
| `_generate_asserts(operation)` | `dict` | `list[AssertModel]` | Генерирует asserts |
|
||||
|
||||
**Формат YAML:**
|
||||
|
||||
```yaml
|
||||
openapi: "3.0.0"
|
||||
info:
|
||||
title: "API Name"
|
||||
version: "1.0.0"
|
||||
|
||||
servers:
|
||||
- url: http://localhost:8080
|
||||
|
||||
paths:
|
||||
/api/login:
|
||||
post:
|
||||
operationId: "loginUser"
|
||||
requestBody:
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
properties:
|
||||
email:
|
||||
type: string
|
||||
responses:
|
||||
"200":
|
||||
description: "Success"
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
properties:
|
||||
token:
|
||||
type: string
|
||||
```
|
||||
|
||||
**Когда использовать:** Для импорта API из OpenAPI спецификаций.
|
||||
|
||||
---
|
||||
|
||||
### get_reader
|
||||
|
||||
**Файл:** `yamlreaders/__init__.py`
|
||||
|
||||
Функция для автоопределения формата и получения appropriate reader.
|
||||
|
||||
| Параметр | Тип | Описание |
|
||||
|----------|-----|----------|
|
||||
| `source` | `str` | Путь к YAML файлу |
|
||||
|
||||
**Возврат:** `BaseReader` (YamlReader, PostmanReader или OpenAPIReader)
|
||||
|
||||
**Исключения:**
|
||||
- `FileNotFoundError` — файл не найден
|
||||
|
||||
**Логика определения:**
|
||||
|
||||
| Признак | Формат | Reader |
|
||||
|---------|--------|--------|
|
||||
| `openapi:` в содержимом | OpenAPI | OpenAPIReader |
|
||||
| `info` и `item` в YAML | Postman | PostmanReader |
|
||||
| Иначе | Кастомный YAML | YamlReader |
|
||||
|
||||
**Когда использовать:** Когда нужно автоматически определить формат файла.
|
||||
|
||||
**Пример:**
|
||||
|
||||
```python
|
||||
from yamlreaders import get_reader
|
||||
|
||||
reader = get_reader("tests/api.yaml")
|
||||
model = reader.parse("tests/api.yaml")
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Примеры использования
|
||||
|
||||
### Базовый пример
|
||||
|
||||
```python
|
||||
from models import TestModel, StepModel, RequestModel, ExtractModel, AssertModel
|
||||
from models import Context, TestRunner
|
||||
|
||||
# Создаём модель теста
|
||||
model = TestModel(
|
||||
name="Тест логина",
|
||||
base_url="http://localhost:8080",
|
||||
variables={"email": "test@x.com", "password": "123456"},
|
||||
steps=[
|
||||
StepModel(
|
||||
name="Логин",
|
||||
request=RequestModel(
|
||||
method="POST",
|
||||
path="/api/login",
|
||||
body={"email": "{{email}}", "password": "{{password}}"}
|
||||
),
|
||||
extract=[
|
||||
ExtractModel(name="token", json_path="$.token")
|
||||
],
|
||||
assert_=[
|
||||
AssertModel(type="status", value=200)
|
||||
]
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
# Запускаем
|
||||
context = Context(model)
|
||||
runner = TestRunner(context)
|
||||
result = runner.run()
|
||||
|
||||
# Результат
|
||||
print(f"Тест: {result.passed}")
|
||||
```
|
||||
|
||||
### Парсинг из YAML
|
||||
|
||||
```python
|
||||
from yamlreaders import get_reader
|
||||
from models import Context, TestRunner
|
||||
|
||||
# Автоопределение формата
|
||||
reader = get_reader("tests/api.yaml")
|
||||
model = reader.parse("tests/api.yaml")
|
||||
|
||||
# Запуск
|
||||
context = Context(model)
|
||||
runner = TestRunner(context)
|
||||
result = runner.run()
|
||||
|
||||
print(result.passed)
|
||||
```
|
||||
|
||||
### Парсинг из Postman
|
||||
|
||||
```python
|
||||
from yamlreaders import PostmanReader
|
||||
from models import Context, TestRunner
|
||||
|
||||
reader = PostmanReader()
|
||||
model = reader.parse("exports/postman_collection.yaml")
|
||||
|
||||
context = Context(model)
|
||||
result = TestRunner(context).run()
|
||||
```
|
||||
|
||||
### Парсинг из OpenAPI
|
||||
|
||||
```python
|
||||
from yamlreaders import OpenAPIReader
|
||||
from models import Context, TestRunner
|
||||
|
||||
reader = OpenAPIReader()
|
||||
model = reader.parse("docs/openapi.yaml")
|
||||
|
||||
context = Context(model)
|
||||
result = TestRunner(context).run()
|
||||
```
|
||||
|
||||
### Использование переменных
|
||||
|
||||
```python
|
||||
model = TestModel(
|
||||
variables={"host": "localhost"},
|
||||
steps=[
|
||||
StepModel(
|
||||
name="Запрос",
|
||||
request=RequestModel(
|
||||
method="GET",
|
||||
path="http://{{host}}:8080/api/users"
|
||||
)
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
context = Context(model)
|
||||
context.resolve("{{host}}") # "localhost"
|
||||
```
|
||||
|
||||
### Извлечение данных
|
||||
|
||||
```python
|
||||
from models import Extractor, Response, ExtractModel
|
||||
|
||||
context = Context(model)
|
||||
extractor = Extractor(context)
|
||||
|
||||
response = Response(
|
||||
status=200,
|
||||
body={"user": {"id": 123, "name": "Test"}},
|
||||
headers={}
|
||||
)
|
||||
|
||||
extractor.extract([
|
||||
ExtractModel(name="user_id", json_path="$.user.id")
|
||||
], response)
|
||||
|
||||
print(context.extracted["user_id"]) # 123
|
||||
```
|
||||
|
||||
### Проверка ответов
|
||||
|
||||
```python
|
||||
from models import Validator, Response, AssertModel
|
||||
|
||||
validator = Validator()
|
||||
response = Response(status=200, body={}, headers={})
|
||||
|
||||
# Проверка статуса
|
||||
asserts = [AssertModel(type="status", value=200)]
|
||||
print(validator.validate(asserts, response)) # True
|
||||
```
|
||||
+5
-1
@@ -14,10 +14,14 @@ class Executor:
|
||||
def __init__(self, context: Context):
|
||||
self.context = context
|
||||
|
||||
def execute(self, request: RequestModel) -> Response:
|
||||
def _resolve_url(self, request: RequestModel) -> str:
|
||||
url = self.context.resolve(request.path)
|
||||
if not url.startswith("http"):
|
||||
url = f"{self.context.config.base_url}{url}"
|
||||
return url
|
||||
|
||||
def execute(self, request: RequestModel) -> Response:
|
||||
url = self._resolve_url(request)
|
||||
|
||||
with httpx.Client() as client:
|
||||
resp = client.request(
|
||||
|
||||
@@ -0,0 +1,281 @@
|
||||
import pytest
|
||||
from models import (
|
||||
AuthModel,
|
||||
RequestModel,
|
||||
ExtractModel,
|
||||
AssertModel,
|
||||
StepModel,
|
||||
TestModel,
|
||||
)
|
||||
from models.context import Context
|
||||
from models.executor import Executor, Response
|
||||
from models.extractor import Extractor
|
||||
from models.validator import Validator
|
||||
from models.runner import TestRunner, TestResult, StepResult
|
||||
|
||||
|
||||
class TestAuthModel:
|
||||
def test_default_values(self):
|
||||
auth = AuthModel()
|
||||
assert auth.type == "none"
|
||||
assert auth.credentials == {}
|
||||
|
||||
def test_with_credentials(self):
|
||||
auth = AuthModel(type="bearer", credentials={"token": "abc123"})
|
||||
assert auth.type == "bearer"
|
||||
assert auth.credentials["token"] == "abc123"
|
||||
|
||||
|
||||
class TestRequestModel:
|
||||
def test_default_values(self):
|
||||
req = RequestModel()
|
||||
assert req.method == "GET"
|
||||
assert req.path == "/"
|
||||
assert req.headers == {}
|
||||
assert req.query == {}
|
||||
assert req.body is None
|
||||
assert req.body_type == "json"
|
||||
assert req.timeout == 30
|
||||
|
||||
def test_custom_values(self):
|
||||
req = RequestModel(
|
||||
method="POST",
|
||||
path="/api/users",
|
||||
headers={"Content-Type": "application/json"},
|
||||
body={"name": "Test"},
|
||||
timeout=60,
|
||||
)
|
||||
assert req.method == "POST"
|
||||
assert req.path == "/api/users"
|
||||
assert req.headers["Content-Type"] == "application/json"
|
||||
assert req.body["name"] == "Test"
|
||||
assert req.timeout == 60
|
||||
|
||||
|
||||
class TestExtractModel:
|
||||
def test_values(self):
|
||||
ext = ExtractModel(name="token", json_path="$.token")
|
||||
assert ext.name == "token"
|
||||
assert ext.json_path == "$.token"
|
||||
assert ext.source == "body"
|
||||
|
||||
def test_header_source(self):
|
||||
ext = ExtractModel(name="auth", json_path="Authorization", source="header")
|
||||
assert ext.source == "header"
|
||||
|
||||
|
||||
class TestAssertModel:
|
||||
def test_status_assert(self):
|
||||
assertion = AssertModel(type="status", value=200)
|
||||
assert assertion.type == "status"
|
||||
assert assertion.value == 200
|
||||
|
||||
def test_json_assert(self):
|
||||
assertion = AssertModel(type="json", path="$.id", operator="exists")
|
||||
assert assertion.type == "json"
|
||||
assert assertion.path == "$.id"
|
||||
assert assertion.operator == "exists"
|
||||
|
||||
|
||||
class TestStepModel:
|
||||
def test_step_with_request(self):
|
||||
step = StepModel(
|
||||
name="Login",
|
||||
request=RequestModel(method="POST", path="/login"),
|
||||
)
|
||||
assert step.name == "Login"
|
||||
assert step.request.method == "POST"
|
||||
assert step.extract == []
|
||||
assert step.assert_ == []
|
||||
|
||||
def test_step_with_extracts_and_asserts(self):
|
||||
step = StepModel(
|
||||
name="Login",
|
||||
request=RequestModel(method="POST", path="/login"),
|
||||
extract=[ExtractModel(name="token", json_path="$.token")],
|
||||
assert_=[AssertModel(type="status", value=200)],
|
||||
)
|
||||
assert len(step.extract) == 1
|
||||
assert len(step.assert_) == 1
|
||||
|
||||
|
||||
class TestTestModel:
|
||||
def test_default_values(self):
|
||||
model = TestModel()
|
||||
assert model.name == "Unnamed Test"
|
||||
assert model.base_url == ""
|
||||
assert model.variables == {}
|
||||
assert model.auth is None
|
||||
assert model.steps == []
|
||||
|
||||
def test_full_model(self):
|
||||
model = TestModel(
|
||||
name="API Test",
|
||||
base_url="http://localhost:8080",
|
||||
variables={"email": "test@x.com"},
|
||||
auth=AuthModel(type="bearer"),
|
||||
steps=[
|
||||
StepModel(
|
||||
name="Step1",
|
||||
request=RequestModel(method="GET", path="/"),
|
||||
)
|
||||
],
|
||||
)
|
||||
assert model.name == "API Test"
|
||||
assert model.base_url == "http://localhost:8080"
|
||||
assert model.variables["email"] == "test@x.com"
|
||||
assert model.auth.type == "bearer"
|
||||
assert len(model.steps) == 1
|
||||
|
||||
|
||||
class TestContext:
|
||||
def test_resolve_simple(self):
|
||||
model = TestModel(variables={"name": "Kirill"})
|
||||
ctx = Context(model)
|
||||
assert ctx.resolve("Hello {{name}}") == "Hello Kirill"
|
||||
|
||||
def test_resolve_extracted(self):
|
||||
model = TestModel()
|
||||
ctx = Context(model)
|
||||
ctx.set_extracted("token", "abc123")
|
||||
assert ctx.resolve("Bearer {{token}}") == "Bearer abc123"
|
||||
|
||||
def test_resolve_multiple(self):
|
||||
model = TestModel(variables={"host": "localhost"})
|
||||
ctx = Context(model)
|
||||
ctx.set_extracted("port", "8080")
|
||||
assert ctx.resolve("{{host}}:{{port}}") == "localhost:8080"
|
||||
|
||||
def test_resolve_no_variable(self):
|
||||
model = TestModel()
|
||||
ctx = Context(model)
|
||||
assert ctx.resolve("No variables here") == "No variables here"
|
||||
|
||||
|
||||
class TestExtractor:
|
||||
def test_extract_from_body(self):
|
||||
model = TestModel()
|
||||
ctx = Context(model)
|
||||
extractor = Extractor(ctx)
|
||||
|
||||
response = Response(status=200, body={"token": "abc123"}, headers={})
|
||||
configs = [ExtractModel(name="token", json_path="$.token")]
|
||||
|
||||
extractor.extract(configs, response)
|
||||
|
||||
assert ctx.extracted["token"] == "abc123"
|
||||
|
||||
def test_extract_nested(self):
|
||||
model = TestModel()
|
||||
ctx = Context(model)
|
||||
extractor = Extractor(ctx)
|
||||
|
||||
response = Response(
|
||||
status=200,
|
||||
body={"user": {"id": 123, "name": "Test"}},
|
||||
headers={},
|
||||
)
|
||||
configs = [ExtractModel(name="user_id", json_path="$.user.id")]
|
||||
|
||||
extractor.extract(configs, response)
|
||||
|
||||
assert ctx.extracted["user_id"] == 123
|
||||
|
||||
def test_extract_from_status(self):
|
||||
model = TestModel()
|
||||
ctx = Context(model)
|
||||
extractor = Extractor(ctx)
|
||||
|
||||
response = Response(status=201, body={}, headers={})
|
||||
configs = [ExtractModel(name="status", json_path="status", source="status")]
|
||||
|
||||
extractor.extract(configs, response)
|
||||
|
||||
assert ctx.extracted["status"] == 201
|
||||
|
||||
def test_extract_from_header(self):
|
||||
model = TestModel()
|
||||
ctx = Context(model)
|
||||
extractor = Extractor(ctx)
|
||||
|
||||
response = Response(status=200, body={}, headers={"X-Request-Id": "req-123"})
|
||||
configs = [ExtractModel(name="req_id", json_path="X-Request-Id", source="header")]
|
||||
|
||||
extractor.extract(configs, response)
|
||||
|
||||
assert ctx.extracted["req_id"] == "req-123"
|
||||
|
||||
|
||||
class TestValidator:
|
||||
def test_validate_status_success(self):
|
||||
validator = Validator()
|
||||
response = Response(status=200, body={}, headers={})
|
||||
asserts = [AssertModel(type="status", value=200)]
|
||||
|
||||
assert validator.validate(asserts, response) is True
|
||||
|
||||
def test_validate_status_failure(self):
|
||||
validator = Validator()
|
||||
response = Response(status=404, body={}, headers={})
|
||||
asserts = [AssertModel(type="status", value=200)]
|
||||
|
||||
assert validator.validate(asserts, response) is False
|
||||
|
||||
def test_validate_empty_asserts(self):
|
||||
validator = Validator()
|
||||
response = Response(status=200, body={}, headers={})
|
||||
|
||||
assert validator.validate([], response) is True
|
||||
|
||||
|
||||
class TestExecutor:
|
||||
def test_resolve_url(self):
|
||||
model = TestModel(base_url="http://localhost:8080")
|
||||
ctx = Context(model)
|
||||
executor = Executor(ctx)
|
||||
|
||||
request = RequestModel(method="GET", path="/api/users")
|
||||
assert executor._resolve_url(request) == "http://localhost:8080/api/users"
|
||||
|
||||
def test_resolve_url_with_variables(self):
|
||||
model = TestModel(base_url="http://localhost:8080")
|
||||
ctx = Context(model)
|
||||
ctx.set_extracted("user_id", "123")
|
||||
executor = Executor(ctx)
|
||||
|
||||
request = RequestModel(method="GET", path="/api/users/{{user_id}}")
|
||||
assert executor._resolve_url(request) == "http://localhost:8080/api/users/123"
|
||||
|
||||
|
||||
class TestStepResult:
|
||||
def test_step_result(self):
|
||||
response = Response(status=200, body={}, headers={})
|
||||
result = StepResult(name="Test", passed=True, response=response)
|
||||
|
||||
assert result.name == "Test"
|
||||
assert result.passed is True
|
||||
assert result.response.status == 200
|
||||
|
||||
|
||||
class TestTestResult:
|
||||
def test_all_passed(self):
|
||||
response = Response(status=200, body={}, headers={})
|
||||
steps = [
|
||||
StepResult(name="S1", passed=True, response=response),
|
||||
StepResult(name="S2", passed=True, response=response),
|
||||
]
|
||||
result = TestResult(steps=steps)
|
||||
assert result.passed is True
|
||||
|
||||
def test_one_failed(self):
|
||||
response = Response(status=200, body={}, headers={})
|
||||
steps = [
|
||||
StepResult(name="S1", passed=True, response=response),
|
||||
StepResult(name="S2", passed=False, response=response),
|
||||
]
|
||||
result = TestResult(steps=steps)
|
||||
assert result.passed is False
|
||||
|
||||
def test_empty_steps(self):
|
||||
result = TestResult(steps=[])
|
||||
assert result.passed is True
|
||||
@@ -0,0 +1,267 @@
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import os
|
||||
|
||||
from yamlreaders import YamlReader, PostmanReader, OpenAPIReader, get_reader
|
||||
from yamlreaders.base import BaseReader, ParseError
|
||||
|
||||
|
||||
TESTS_DIR = Path(__file__).parent
|
||||
EXAMPLES_DIR = TESTS_DIR.parent / "examples"
|
||||
|
||||
|
||||
class TestBaseReader:
|
||||
def test_is_abstract(self):
|
||||
with pytest.raises(TypeError):
|
||||
BaseReader()
|
||||
|
||||
|
||||
class TestYamlReader:
|
||||
def test_validate_valid(self):
|
||||
reader = YamlReader()
|
||||
assert reader.validate(EXAMPLES_DIR / "custom.yaml") is True
|
||||
|
||||
def test_validate_invalid(self):
|
||||
reader = YamlReader()
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
|
||||
f.write("invalid: yaml\ncontent: here")
|
||||
f.flush()
|
||||
result = reader.validate(f.name)
|
||||
os.unlink(f.name)
|
||||
assert result is False
|
||||
|
||||
def test_parse(self):
|
||||
reader = YamlReader()
|
||||
model = reader.parse(EXAMPLES_DIR / "custom.yaml")
|
||||
|
||||
assert model.name == "Test Authorization"
|
||||
assert model.base_url == "http://localhost:8080"
|
||||
assert model.variables["email"] == "test@example.com"
|
||||
assert model.variables["password"] == "Test123!"
|
||||
assert len(model.steps) == 3
|
||||
|
||||
def test_parse_step_name(self):
|
||||
reader = YamlReader()
|
||||
model = reader.parse(EXAMPLES_DIR / "custom.yaml")
|
||||
|
||||
assert model.steps[0].name == "Register User"
|
||||
assert model.steps[1].name == "Login"
|
||||
assert model.steps[2].name == "Get Profile"
|
||||
|
||||
def test_parse_request(self):
|
||||
reader = YamlReader()
|
||||
model = reader.parse(EXAMPLES_DIR / "custom.yaml")
|
||||
|
||||
req = model.steps[0].request
|
||||
assert req.method == "POST"
|
||||
assert req.path == "/api/register"
|
||||
assert req.headers["Content-Type"] == "application/json"
|
||||
assert req.body["email"] == "{{email}}"
|
||||
|
||||
def test_parse_extract(self):
|
||||
reader = YamlReader()
|
||||
model = reader.parse(EXAMPLES_DIR / "custom.yaml")
|
||||
|
||||
extract = model.steps[0].extract
|
||||
assert len(extract) == 1
|
||||
assert extract[0].name == "user_id"
|
||||
assert extract[0].json_path == "$.id"
|
||||
|
||||
def test_parse_assert(self):
|
||||
reader = YamlReader()
|
||||
model = reader.parse(EXAMPLES_DIR / "custom.yaml")
|
||||
|
||||
asserts = model.steps[0].assert_
|
||||
assert len(asserts) == 1
|
||||
assert asserts[0].type == "status"
|
||||
assert asserts[0].value == 201
|
||||
|
||||
def test_parse_step_without_name_fails(self):
|
||||
reader = YamlReader()
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
|
||||
f.write("steps:\n - request:\n method: GET\n url: /")
|
||||
f.flush()
|
||||
with pytest.raises(ParseError):
|
||||
reader.parse(f.name)
|
||||
os.unlink(f.name)
|
||||
|
||||
def test_parse_step_without_request_fails(self):
|
||||
reader = YamlReader()
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
|
||||
f.write("steps:\n - name: Test")
|
||||
f.flush()
|
||||
with pytest.raises(ParseError):
|
||||
reader.parse(f.name)
|
||||
os.unlink(f.name)
|
||||
|
||||
|
||||
class TestPostmanReader:
|
||||
def test_validate_valid(self):
|
||||
reader = PostmanReader()
|
||||
assert reader.validate(EXAMPLES_DIR / "postman.yaml") is True
|
||||
|
||||
def test_parse(self):
|
||||
reader = PostmanReader()
|
||||
model = reader.parse(EXAMPLES_DIR / "postman.yaml")
|
||||
|
||||
assert model.name == "Test Collection"
|
||||
assert model.base_url == "http://localhost:8080"
|
||||
assert model.variables["email"] == "test@example.com"
|
||||
assert len(model.steps) == 2
|
||||
|
||||
def test_parse_step_names(self):
|
||||
reader = PostmanReader()
|
||||
model = reader.parse(EXAMPLES_DIR / "postman.yaml")
|
||||
|
||||
assert model.steps[0].name == "Login"
|
||||
assert model.steps[1].name == "Get Profile"
|
||||
|
||||
def test_parse_request_method(self):
|
||||
reader = PostmanReader()
|
||||
model = reader.parse(EXAMPLES_DIR / "postman.yaml")
|
||||
|
||||
assert model.steps[0].request.method == "POST"
|
||||
assert model.steps[1].request.method == "GET"
|
||||
|
||||
def test_parse_request_headers(self):
|
||||
reader = PostmanReader()
|
||||
model = reader.parse(EXAMPLES_DIR / "postman.yaml")
|
||||
|
||||
headers = model.steps[0].request.headers
|
||||
assert headers["Content-Type"] == "application/json"
|
||||
|
||||
def test_parse_request_body(self):
|
||||
reader = PostmanReader()
|
||||
model = reader.parse(EXAMPLES_DIR / "postman.yaml")
|
||||
|
||||
body = model.steps[0].request.body
|
||||
assert body["email"] == "{{email}}"
|
||||
assert body["password"] == "Test123!"
|
||||
|
||||
def test_parse_nested_items(self):
|
||||
reader = PostmanReader()
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
|
||||
f.write("""
|
||||
info:
|
||||
name: Nested
|
||||
|
||||
item:
|
||||
- name: Folder
|
||||
item:
|
||||
- name: Nested Request
|
||||
request:
|
||||
method: GET
|
||||
url: /nested
|
||||
""")
|
||||
f.flush()
|
||||
model = reader.parse(f.name)
|
||||
os.unlink(f.name)
|
||||
|
||||
assert len(model.steps) == 1
|
||||
assert model.steps[0].name == "Nested Request"
|
||||
|
||||
def test_validate_invalid(self):
|
||||
reader = PostmanReader()
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
|
||||
f.write("info: {}\nother: data")
|
||||
f.flush()
|
||||
assert reader.validate(f.name) is False
|
||||
os.unlink(f.name)
|
||||
|
||||
|
||||
class TestOpenAPIReader:
|
||||
def test_validate_valid(self):
|
||||
reader = OpenAPIReader()
|
||||
assert reader.validate(EXAMPLES_DIR / "openapi.yaml") is True
|
||||
|
||||
def test_parse(self):
|
||||
reader = OpenAPIReader()
|
||||
model = reader.parse(EXAMPLES_DIR / "openapi.yaml")
|
||||
|
||||
assert model.name == "Test API"
|
||||
assert model.base_url == "http://localhost:8080"
|
||||
assert len(model.steps) == 2
|
||||
|
||||
def test_parse_step_names(self):
|
||||
reader = OpenAPIReader()
|
||||
model = reader.parse(EXAMPLES_DIR / "openapi.yaml")
|
||||
|
||||
names = [s.name for s in model.steps]
|
||||
assert "loginUser" in names
|
||||
assert "getProfile" in names
|
||||
|
||||
def test_parse_request_methods(self):
|
||||
reader = OpenAPIReader()
|
||||
model = reader.parse(EXAMPLES_DIR / "openapi.yaml")
|
||||
|
||||
methods = [s.request.method for s in model.steps]
|
||||
assert "POST" in methods
|
||||
assert "GET" in methods
|
||||
|
||||
def test_parse_request_paths(self):
|
||||
reader = OpenAPIReader()
|
||||
model = reader.parse(EXAMPLES_DIR / "openapi.yaml")
|
||||
|
||||
paths = [s.request.path for s in model.steps]
|
||||
assert "/api/login" in paths
|
||||
assert "/api/profile" in paths
|
||||
|
||||
def test_parse_generates_asserts(self):
|
||||
reader = OpenAPIReader()
|
||||
model = reader.parse(EXAMPLES_DIR / "openapi.yaml")
|
||||
|
||||
for step in model.steps:
|
||||
assert len(step.assert_) > 0
|
||||
assert step.assert_[0].type == "status"
|
||||
assert step.assert_[0].value == 200
|
||||
|
||||
def test_parse_with_servers(self):
|
||||
reader = OpenAPIReader()
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
|
||||
f.write("""
|
||||
openapi: "3.0.0"
|
||||
info:
|
||||
title: Test
|
||||
version: "1.0"
|
||||
servers:
|
||||
- url: https://api.example.com/v1
|
||||
paths:
|
||||
/test:
|
||||
get:
|
||||
operationId: test
|
||||
responses:
|
||||
"200":
|
||||
description: OK
|
||||
""")
|
||||
f.flush()
|
||||
model = reader.parse(f.name)
|
||||
os.unlink(f.name)
|
||||
|
||||
assert model.base_url == "https://api.example.com/v1"
|
||||
|
||||
def test_validate_invalid(self):
|
||||
reader = OpenAPIReader()
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
|
||||
f.write("swagger: '2.0'\ninfo:\n title: Test")
|
||||
f.flush()
|
||||
assert reader.validate(f.name) is False
|
||||
os.unlink(f.name)
|
||||
|
||||
|
||||
class TestGetReader:
|
||||
def test_auto_detect_yaml(self):
|
||||
reader = get_reader(EXAMPLES_DIR / "custom.yaml")
|
||||
assert isinstance(reader, YamlReader)
|
||||
|
||||
def test_auto_detect_postman(self):
|
||||
reader = get_reader(EXAMPLES_DIR / "postman.yaml")
|
||||
assert isinstance(reader, PostmanReader)
|
||||
|
||||
def test_auto_detect_openapi(self):
|
||||
reader = get_reader(EXAMPLES_DIR / "openapi.yaml")
|
||||
assert isinstance(reader, OpenAPIReader)
|
||||
|
||||
def test_file_not_found(self):
|
||||
with pytest.raises(FileNotFoundError):
|
||||
get_reader("nonexistent.yaml")
|
||||
@@ -0,0 +1,35 @@
|
||||
from .base import BaseReader, ParseError
|
||||
from .yaml_reader import YamlReader
|
||||
from .postman_reader import PostmanReader
|
||||
from .openapi_reader import OpenAPIReader
|
||||
|
||||
|
||||
def get_reader(source: str) -> BaseReader:
|
||||
from pathlib import Path
|
||||
import yaml
|
||||
|
||||
path = Path(source)
|
||||
|
||||
if not path.exists():
|
||||
raise FileNotFoundError(f"File not found: {path}")
|
||||
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
content = f.read(2048)
|
||||
|
||||
if "openapi:" in content:
|
||||
try:
|
||||
data = yaml.safe_load(content)
|
||||
if isinstance(data, dict) and "openapi" in data:
|
||||
return OpenAPIReader()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
try:
|
||||
data = yaml.safe_load(content)
|
||||
if isinstance(data, dict):
|
||||
if "info" in data and "item" in data:
|
||||
return PostmanReader()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return YamlReader()
|
||||
@@ -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
|
||||
@@ -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
|
||||
Reference in new issue
Block a user