Conftest.py в Pytest: мощный инструмент для организации тестов

# conftest.py
import pytest
import requests

@pytest.fixture
def api_client():
base_url = "https://api.example.com/v1"
session = requests.Session()
session.headers.update({
"Content-Type": "application/json",
"Authorization": "Bearer test_token"
})
yield session
session.close()

# test_users.py
def test_get_user(api_client):
response = api_client.get("/users/1")
assert response.status_code == 200
assert response.json()["name"] == "Test User"

# conftest.py
import pytest

@pytest.fixture
def db_connection():
# Настройка соединения с БД
connection = create_db_connection()
yield connection
connection.close()

@pytest.fixture
def user_repository(db_connection):
# Используем существующую фикстуру для создания более специализированной
return UserRepository(db_connection)

@pytest.fixture
def test_user(user_repository):
# Создаём тестового пользователя
user = user_repository.create(name="Test User", email="test@example.com")
yield user
# Очистка после теста
user_repository.delete(user.id)

# conftest.py
import pytest

def user_factory(role):
@pytest.fixture
def _user():
# Создаём пользователя с указанной ролью
return {"id": 1, "name": "Test User", "role": role}
return _user

# Создаём конкретные фикстуры из фабрики
admin_user = user_factory("admin")
regular_user = user_factory("user")
guest_user = user_factory("guest")

# conftest.py
import pytest
from selenium import webdriver

@pytest.fixture(scope="function")
def temp_file():
# Создаётся для каждого теста
path = "temp_test_data.txt"
with open(path, "w") as f:
f.write("Test data")
yield path
import os
os.remove(path) # Очистка после каждого теста

@pytest.fixture(scope="module")
def db_connection():
# Создаётся один раз на модуль
print("Opening DB connection")
conn = create_connection()
yield conn
print("Closing DB connection")
conn.close()

@pytest.fixture(scope="session")
def web_driver():
# Создаётся один раз на всю сессию тестирования
print("Starting WebDriver")
driver = webdriver.Chrome()
driver.implicitly_wait(10)
yield driver
print("Shutting down WebDriver")
driver.quit()

# conftest.py
import pytest
import os

def determine_scope(fixture_name, config):
if os.environ.get("CI") == "true":
# В CI-окружении используем более широкую область 
# для ускорения тестирования
return "module"
# На локальной машине предпочитаем лучшую изоляцию
return "function"

@pytest.fixture(scope=determine_scope)
def dynamic_scoped_fixture():
# Область видимости определяется динамически
return expensive_resource()

# conftest.py
import pytest
import os
import logging

# Настройка логгирования для всех тестов
def pytest_configure(config):
"""
Вызывается в начале тестовой сессии перед сбором тестов
"""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
handlers=[
logging.FileHandler('test_run.log'),
logging.StreamHandler()
]
)
logging.info("Starting test session")

# Регистрация кастомного маркера
config.addinivalue_line(
"markers", "slow: marks test as slow running"
)

# Пропуск медленных тестов на CI
def pytest_runtest_setup(item):
"""
Вызывается перед каждым тестом
"""
if os.environ.get("CI") == "true" and "slow" in item.keywords:
pytest.skip("Skipping slow test in CI environment")

# Модификация сообщения об ошибке
def pytest_assertion_pass(item, lineno, orig):
"""
Вызывается при успешной проверке assert
"""
logging.info(f"Assertion passed in {item.name}")

def pytest_assertion_fail(item, message):
"""
Вызывается при провале assert
"""
logging.error(f"Assertion failed in {item.name}: {message}")

# Кастомизация сбора тестов
def pytest_collect_file(parent, path):
"""
Вызывается для каждого файла при сборе тестов
"""
if path.basename.startswith("integration_") and path.ext == ".py":
# Можно вернуть специальный коллектор для таких файлов
return parent.session.collect_file(path)

# conftest.py
def pytest_addoption(parser):
"""
Добавление кастомных параметров командной строки
"""
parser.addoption(
"--env",
choices=["dev", "staging", "prod"],
default="dev",
help="Specify the test environment"
)
parser.addoption(
"--api-version", 
default="v1",
help="API version to test"
)

# Затем доступ к параметрам через фикстуру
@pytest.fixture(scope="session")
def env(request):
return request.config.getoption("--env")

@pytest.fixture(scope="session")
def api_base_url(env):
"""
Фикстура, использующая параметр командной строки
"""
urls = {
"dev": "https://dev-api.example.com",
"staging": "https://staging-api.example.com",
"prod": "https://api.example.com"
}
return urls[env]

# conftest.py
def pytest_terminal_summary(terminalreporter, exitstatus, config):
"""
Вызывается в конце тестовой сессии для вывода финального отчёта
"""
print("\n=== Custom Test Summary ===")
print(f"Total tests: {terminalreporter.stats.get('passed', 0) + terminalreporter.stats.get('failed', 0)}")
print(f"Passed: {len(terminalreporter.stats.get('passed', []))}")
print(f"Failed: {len(terminalreporter.stats.get('failed', []))}")
print(f"Skipped: {len(terminalreporter.stats.get('skipped', []))}")
print(f"Execution time: {terminalreporter.config.duration:.2f} seconds")

# Можно добавить интеграцию с внешними системами
if exitstatus != 0 and os.environ.get("NOTIFY_FAILURES"):
send_notification("Test failures detected!")

# conftest.py
def pytest_xdist_node_collection_finished(node, ids):
"""
Вызывается на каждом узле при завершении сбора тестов при использовании pytest-xdist
"""
print(f"Node {node.gateway.id} collected {len(ids)} tests")

def pytest_configure_node(node):
"""
Вызывается для настройки удалённого узла при использовании pytest-xdist
"""
node.workerinput["node_id"] = node.gateway.id
# Можно передать информацию удалённому процессу
if node.gateway.id == "w0":
node.workerinput["tests_type"] = "api"
elif node.gateway.id == "w1":
node.workerinput["tests_type"] = "ui"

# conftest.py
import pytest
import json
import os

def pytest_addoption(parser):
parser.addoption("--env", default="dev")

@pytest.fixture(scope="session")
def config(request):
# Загружаем конфигурацию в зависимости от окружения
env = request.config.getoption("--env")
config_path = f"configs/{env}_config.json"

with open(config_path) as f:
config = json.load(f)

# Обогащаем конфигурацию динамическими данными
config["test_run_id"] = os.environ.get("TEST_RUN_ID", "local_run")

return config

@pytest.fixture(scope="session")
def api_client(config):
"""Клиент API с настройками для указанного окружения"""
client = APIClient(
base_url=config["api_url"],
timeout=config["timeout"],
headers={"X-Environment": config["env_name"]}
)
return client

@pytest.fixture(scope="function")
def test_data(config):
"""Загрузка тестовых данных для конкретного окружения"""
env = config["env_name"]
with open(f"test_data/{env}_data.json") as f:
return json.load(f)

# conftest.py
import pytest
from auth_service import get_token

# Создаём фабрику авторизованных клиентов
def create_authorized_client(role):
@pytest.fixture(scope="function")
def _authorized_client(api_client):
token = get_token(role)
api_client.headers.update({"Authorization": f"Bearer {token}"})
return api_client
return _authorized_client

# Фикстуры для разных ролей
admin_client = create_authorized_client("admin")
user_client = create_authorized_client("user")
guest_client = create_authorized_client("guest")

# Затем в тестах:
# def test_admin_access(admin_client):
# ...
# def test_user_permissions(user_client):
# ...

# conftest.py
import pytest
import allure
import datetime
import requests

@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item, call):
"""Создаёт расширенный отчёт о прохождении теста"""
outcome = yield
report = outcome.get_result()

# Добавляем дополнительные метаданные к отчету
report.start_time = datetime.datetime.now().isoformat()
report.environment = item.config.getoption("--env", default="dev")

if report.when == "call":
# Только для фазы выполнения теста
allure.attach(
body=str(report.duration),
name="Test Duration",
attachment_type=allure.attachment_type.TEXT
)

@pytest.fixture(scope="function", autouse=True)
def report_test_metrics(request):
"""Автоматически отправляет метрики о тестах в систему мониторинга"""
start_time = datetime.datetime.now()

yield

duration = (datetime.datetime.now() – start_time).total_seconds()
test_name = request.node.name
result = "passed" if not request.node.rep_call.failed else "failed"

# Отправка метрик в внешнюю систему (например, Prometheus, Graphite)
try:
requests.post(
"http://metrics-collector.example.com/submit",
json={
"test_name": test_name,
"duration": duration,
"result": result,
"environment": request.config.getoption("--env", default="dev")
},
timeout=1
)
except Exception as e:
print(f"Failed to send metrics: {e}")

# conftest.py
def pytest_configure(config):
"""Настраивает метаданные тестов для правильной параллелизации"""
if config.getoption("--xdist-workers") not in [None, "auto"]:
# Регистрируем кастомный атрибут для тестов
config.addinivalue_line("markers", 
"resource(name): mark test as requiring specific resource")

def pytest_xdist_node_collection_finished(node, ids):
"""Распределяет тесты по узлам в зависимости от используемых ресурсов"""
import random

# Получаем общее количество узлов
total_nodes = int(node.config.getoption("--numprocesses"))
node_id = int(node.gateway.id.replace('gw', ''))

# Определяем ресурсы для этого узла
# (например, базы данных, API-эндпоинты и т.д.)
resources_per_node = {
0: ["database_a", "api_user"],
1: ["database_b", "api_product"],
# ...и так далее для каждого узла
}

assigned_resources = resources_per_node.get(node_id, [])

# Для каждого узла фильтруем только тесты, использующие его ресурсы
selected_tests = []
for test_id in ids:
for marker in node.session.items_by_id[test_id].iter_markers(name="resource"):
if marker.args[0] in assigned_resources:
selected_tests.append(test_id)
break
else:
# Если тест не требует особых ресурсов, распределяем случайно
if hash(test_id) % total_nodes == node_id:
selected_tests.append(test_id)

# Оставляем для этого узла только отобранные тесты
node.slaveinput["selected_tests"] = selected_tests

# conftest.py
import pytest
import json

def pytest_generate_tests(metafunc):
"""Динамически генерирует тестовые случаи из файла данных"""
if "api_endpoint" in metafunc.fixturenames:
with open("api_endpoints.json") as f:
endpoints = json.load(f)

# Параметризуем тест для каждого эндпоинта
metafunc.parametrize(
"api_endpoint,expected_status", 
[(e["url"], e["status"]) for e in endpoints]
)