Compare commits
10
Commits
0f7dd83b33
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
 Black_Raven
|
2e7c69c5af | ||
|
Black_Raven
|
c26c60f679 | ||
|
Black_Raven
|
e9cfdd9e94 | ||
|
Black_Raven
|
6ae350b48c | ||
|
Black_Raven
|
805b28e344 | ||
|
Black_Raven
|
130d06351d | ||
|
Black_Raven
|
2250961d09 | ||
|
Black_Raven
|
bf47218831 | ||
|
Black_Raven
|
f28c1d6d7c | ||
|
Black_Raven
|
d6b97e8b09 |
No files matched your search
@@ -135,3 +135,315 @@ result = runner.run()
|
|||||||
|
|
||||||
print(result.passed) # True/False
|
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):
|
def __init__(self, context: Context):
|
||||||
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)
|
url = self.context.resolve(request.path)
|
||||||
if not url.startswith("http"):
|
if not url.startswith("http"):
|
||||||
url = f"{self.context.config.base_url}{url}"
|
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:
|
with httpx.Client() as client:
|
||||||
resp = client.request(
|
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