14.08.2025 06:47
sound_right 0 213 Источник

Вступление

Иногда кажется, что добавление async/await в тесты — это почти «бесплатный способ» сделать их быстрее. Мы ведь знаем, что тесты часто тратят время на ожидание ответов от серверов или UI‑действий, и в голову сразу приходит мысль: «А что, если пока один тест ждёт ответа, другой начнёт выполняться?» Кажется логичным, что так можно сэкономить минуты, а то и десятки минут прогона.

Но на практике всё оказывается не так однозначно. Асинхронность — это не магическая кнопка, делающая тесты параллельными «по щелчку». async/await в Python работает внутри одного потока и не создаёт реального параллелизма. Да, можно выстроить корутины так, что пока одна ждёт ответа, другая начинает выполняться, и на маленьких изолированных примерах это действительно выглядит впечатляюще. Но в реальных проектах всё сложнее: нужно переписывать фикстуры, страницы, клиенты, поддерживать асинхронные версии библиотек и отлавливать ошибки, которые при синхронном подходе просто не возникли бы.

В этой статье мы разберём, как выглядит переписывание автотестов на async/await, используя Python и Pytest. Для UI тестов будем использовать Playwright, для API — HTTPX. Мы посмотрим, какие изменения нужно вносить в тестовый фреймворк, что происходит с PageObject и API‑клиентами, как меняется структура тестов и фикстур. И что важнее — мы обсудим, когда такой подход имеет смысл, а когда он скорее создаёт лишние сложности, чем даёт реальный выигрыш.

Отдельно отмечу: цель этой статьи — не призывать к массовому переходу на асинхронные тесты, а показать, как это можно сделать и зачем иногда это действительно нужно. Мы не будем подробно повторять создание тестового фреймворка с нуля (этот материал уже разобран в предыдущих статьях), а сосредоточимся на отличиях между синхронной и асинхронной реализацией.

Ссылки на материалы с базовой архитектурой я оставлю ниже, чтобы при желании можно было погрузиться в детали:

Технологии

Для реализации асинхронных тестов мы будем использовать привычные инструменты, которые и в обычных синхронных тестах применяются ежедневно. Основой, как и прежде, остаётся pytest — он удобен, поддерживает фикстуры, расширяемость и гибкую организацию тестов. Чтобы добавить поддержку async/await, достаточно установить плагин pytest-asyncio, который умеет запускать тестовые функции как корутины и управлять асинхронным event loop внутри Pytest.

Для UI тестов мы будем использовать Playwright. Этот инструмент изначально проектировался с поддержкой асинхронного API, что упрощает миграцию: вместо синхронных вызовов браузера и страниц (playwright.sync_api) мы просто переходим на асинхронную версию (playwright.async_api). Фактически, это не «хаки» и не «обходные пути», а штатный способ работы с Playwright.

Для API автотестов мы будем использовать библиотеку HTTPX, у которой уже есть полноценная поддержка асинхронных запросов через AsyncClient. Это упрощает задачу: переписывание клиента сводится к добавлению async и await в методы, без необходимости городить дополнительные обёртки или подменять низкоуровневый транспорт.

Остальная архитектура тестов (фикстуры, PageObject, логгирование, валидация схем) остаётся прежней. Если вы уже знакомы с моими предыдущими материалами по построению фреймворка, то увидите, что структура остаётся привычной — меняется лишь способ вызова методов и работы с асинхронными библиотеками.

Подробно сам подход к построению UI и API фреймворков мы разбирали ранее:

UI автотесты

Начнём с UI тестов и посмотрим, как можно преобразовать синхронный код в асинхронный.

Playwright в этом плане сильно упрощает задачу: изначально он спроектирован как асинхронная библиотека, и синхронная версия (playwright.sync_api) — это скорее обёртка над основным асинхронным API. Поэтому переход к async/await не требует радикальной переделки фреймворка: вы просто переключаетесь на другой модуль (playwright.async_api) и добавляете ключевые слова async и await в соответствующих местах.

На практике это означает, что основная структура тестов остаётся прежней: фикстуры, PageObject и PageComponent продолжают работать по тем же принципам, просто становятся асинхронными. Этот момент важен, потому что многие ожидают, что придётся переписывать «всё и сразу». На самом деле меняется только способ вызова методов браузера и взаимодействия со страницами.

Полные примеры реализации UI тестов вы можете найти в репозиториях:

Фикстуры

Первое, что нужно изменить при переходе на async/await, — это фикстуры. В обычном Pytest они синхронные и работают в стандартном потоке выполнения. Для асинхронных тестов нужен собственный событийный цикл, поэтому используется плагин pytest-asyncio, который добавляет поддержку асинхронных фикстур и тестов.

Синхронная реализация: /fixtures/pages.py

import uuid

import allure
import pytest
from playwright.sync_api import Playwright, Page, expect

from config import Settings
from pages.dashboard_page import DashboardPage
from pages.registration_page import RegistrationPage


@pytest.fixture
def chromium_page(playwright: Playwright, settings: Settings) -> Page:
    expect.set_options(timeout=settings.expect_timeout)

    browser = playwright.chromium.launch(headless=settings.headless)
    context = browser.new_context(
        base_url=f"{settings.app_url}/",
        record_video_dir=settings.videos_dir
    )
    context.tracing.start(screenshots=True, snapshots=True, sources=True)

    page = context.new_page()
    yield page

    tracing_file = settings.tracing_dir.joinpath(f'{uuid.uuid4()}.zip')
    context.tracing.stop(path=tracing_file)
    browser.close()

    allure.attach.file(tracing_file, name='trace', extension='zip')
    allure.attach.file(page.video.path(), name='video', attachment_type=allure.attachment_type.WEBM)


@pytest.fixture
def dashboard_page(chromium_page: Page) -> DashboardPage:
    return DashboardPage(page=chromium_page)


@pytest.fixture
def registration_page(chromium_page: Page) -> RegistrationPage:
    return RegistrationPage(page=chromium_page)

Асинхронная реализация: /fixtures/pages.py

import uuid

import allure
import pytest_asyncio
from playwright.async_api import Page, expect, async_playwright

from config import Settings
from pages.dashboard_page import DashboardPage
from pages.registration_page import RegistrationPage


@pytest_asyncio.fixture
async def chromium_page(settings: Settings) -> Page:
    async with async_playwright() as playwright:
        expect.set_options(timeout=settings.expect_timeout)

        browser = await playwright.chromium.launch(headless=settings.headless)
        context = await browser.new_context(
            base_url=f"{settings.app_url}/",
            record_video_dir=settings.videos_dir
        )
        await context.tracing.start(screenshots=True, snapshots=True, sources=True)

        page = await context.new_page()
        yield page

        video_file = await page.video.path()
        tracing_file = settings.tracing_dir.joinpath(f'{uuid.uuid4()}.zip')
        await context.tracing.stop(path=tracing_file)
        await browser.close()

        allure.attach.file(tracing_file, name='trace', extension='zip')
        allure.attach.file(video_file, name='video', attachment_type=allure.attachment_type.WEBM)


@pytest_asyncio.fixture
async def dashboard_page(chromium_page: Page) -> DashboardPage:
    return DashboardPage(page=chromium_page)


@pytest_asyncio.fixture
async def registration_page(chromium_page: Page) -> RegistrationPage:
    return RegistrationPage(page=chromium_page)

В синхронной версии мы использовали стандартный pytest.fixture и синхронный Playwright API (playwright.sync_api). Запуск браузера, создание контекста и открытие страницы выполнялись пошагово и блокировали поток до завершения каждого шага.

В асинхронной версии всё меняется следующим образом:

  • Вместо pytest.fixture используется pytest_asyncio.fixture, который позволяет объявить функцию как async def. Это значит, что фикстура сама становится корутиной и может вызывать асинхронные операции без блокировки основного потока.

  • Вместо синхронного API Playwright используется модуль playwright.async_api, который предоставляет асинхронные версии всех методов. Теперь запуск браузера (launch), создание контекста (new_context) и открытие страницы (new_page) выполняются с ключевым словом await.

  • Встроенная фикстура playwright из плагина pytest-playwright использоваться не может, так как она инициализирует синхронный API (sync_playwright). Поэтому мы самостоятельно открываем контекст через async with async_playwright() as playwright:. Это гарантирует корректное управление ресурсами и завершение всех подключений при выходе из фикстуры.

  • Закрытие браузера (browser.close()), остановка трейсинга (context.tracing.stop()) и работа с видео (page.video.path()) теперь тоже выполняются с использованием await, потому что эти операции в асинхронном API тоже не блокируют поток.

  • Фикстуры, возвращающие страницы (dashboard_page, registration_page), объявлены асинхронными не «ради галочки», а потому что они напрямую зависят от асинхронной фикстуры chromium_page. Pytest требует, чтобы вся цепочка зависимостей оставалась асинхронной, иначе ожидание значений из корутин просто не сработает.

С точки зрения использования в тестах ничего не меняется — фикстуры по-прежнему доступны по имени и подставляются в параметры тестовых функций. Но «под капотом» они работают в событийном цикле, создаваемом pytest-asyncio, и больше не блокируют выполнение, что особенно заметно при большом количестве асинхронных вызовов.

PageObject, PageComponent, PageFactory

Поскольку мы переходим на использование playwright.async_api, необходимо адаптировать всю работу со страницами: начиная с базовых элементов (PageFactory и отдельные элементы), далее компоненты интерфейса (PageComponent), и завершая полноценными объектами страниц (PageObject). Рассмотрим это на примерах, начиная с самого низкого уровня — базового элемента.

PageFactory

Синхронная реализация: /elements/base_element.py

import allure
from playwright.sync_api import Page, Locator, expect

from tools.logger import get_logger

logger = get_logger("BASE_ELEMENT")


class BaseElement:
    def __init__(self, page: Page, locator: str, name: str) -> None:
        self.page = page
        self.name = name
        self.locator = locator

    @property
    def type_of(self) -> str:
        return "base element"

    def get_locator(self, nth: int = 0, **kwargs) -> Locator:
        locator = self.locator.format(**kwargs)
        step = f'Getting locator with "data-testid={locator}" at index "{nth}"'

        with allure.step(step):
            logger.info(step)
            return self.page.get_by_test_id(locator).nth(nth)

    def click(self, nth: int = 0, **kwargs):
        step = f'Clicking {self.type_of} "{self.name}"'

        with allure.step(step):
            locator = self.get_locator(nth, **kwargs)
            logger.info(step)
            locator.click()

    def check_visible(self, nth: int = 0, **kwargs):
        step = f'Checking that {self.type_of} "{self.name}" is visible'

        with allure.step(step):
            locator = self.get_locator(nth, **kwargs)
            logger.info(step)
            expect(locator).to_be_visible()

    def check_have_text(self, text: str, nth: int = 0, **kwargs):
        step = f'Checking that {self.type_of} "{self.name}" has text "{text}"'

        with allure.step(step):
            locator = self.get_locator(nth, **kwargs)
            logger.info(step)
            expect(locator).to_have_text(text)

Асинхронная реализация: /elements/base_element.py

import allure
from playwright.async_api import Page, Locator, expect

from tools.logger import get_logger

logger = get_logger("BASE_ELEMENT")


class BaseElement:
    def __init__(self, page: Page, locator: str, name: str) -> None:
        self.page = page
        self.name = name
        self.locator = locator

    @property
    def type_of(self) -> str:
        return "base element"

    def get_locator(self, nth: int = 0, **kwargs) -> Locator:
        locator = self.locator.format(**kwargs)
        step = f'Getting locator with "data-testid={locator}" at index "{nth}"'

        with allure.step(step):
            logger.info(step)
            return self.page.get_by_test_id(locator).nth(nth)

    async def click(self, nth: int = 0, **kwargs):
        step = f'Clicking {self.type_of} "{self.name}"'

        with allure.step(step):
            locator = self.get_locator(nth, **kwargs)
            logger.info(step)
            await locator.click()

    async def check_visible(self, nth: int = 0, **kwargs):
        step = f'Checking that {self.type_of} "{self.name}" is visible'

        with allure.step(step):
            locator = self.get_locator(nth, **kwargs)
            logger.info(step)
            await expect(locator).to_be_visible()

    async def check_have_text(self, text: str, nth: int = 0, **kwargs):
        step = f'Checking that {self.type_of} "{self.name}" has text "{text}"'

        with allure.step(step):
            locator = self.get_locator(nth, **kwargs)
            logger.info(step)
            await expect(locator).to_have_text(text)

В исходной (синхронной) версии методы, такие как click, check_visible, check_have_text, выполнялись блокирующе. После миграции они стали async def и внутри вызывают асинхронные методы Playwright (await locator.click(), await expect(locator).to_be_visible()). Это значит, что каждый шаг теперь отдаёт управление обратно в событийный цикл, и выполнение может переключаться на другие задачи, пока Playwright ждёт ответа от браузера. Интерфейс остался прежним (снаружи всё так же выглядит как «кликнуть», «проверить»), но теперь выполнение не блокирует поток.

PageComponent

Далее посмотрим на пример компонента. Сравним, как выглядела синхронная версия и как она реализуется после перехода на асинхронный API.

Синхронная реализация: /components/navbar_component.py

import allure
from playwright.sync_api import Page

from components.base_component import BaseComponent
from elements.text import Text


class NavbarComponent(BaseComponent):
    def __init__(self, page: Page):
        super().__init__(page)

        self.app_title = Text(page, 'navigation-navbar-app-title-text', 'App title')
        self.welcome_title = Text(page, 'navigation-navbar-welcome-title-text', 'Welcome title')

    @allure.step("Check visible navbar")
    def check_visible(self, username: str):
        self.app_title.check_visible()
        self.app_title.check_have_text('UI Course')

        self.welcome_title.check_visible()
        self.welcome_title.check_have_text(f'Welcome, {username}!')

Асинхронная реализация: /components/navbar_component.py

import allure
from playwright.async_api import Page

from components.base_component import BaseComponent
from elements.text import Text


class NavbarComponent(BaseComponent):
    def __init__(self, page: Page):
        super().__init__(page)

        self.app_title = Text(page, 'navigation-navbar-app-title-text', 'App title')
        self.welcome_title = Text(page, 'navigation-navbar-welcome-title-text', 'Welcome title')

    @allure.step("Check visible navbar")
    async def check_visible(self, username: str):
        await self.app_title.check_visible()
        await self.app_title.check_have_text('UI Course')

        await self.welcome_title.check_visible()
        await self.welcome_title.check_have_text(f'Welcome, {username}!')

Компоненты, которые использовали элементы, тоже стали асинхронными. Методы вроде check_visible теперь объявлены через async def, а вызовы методов элементов выполняются с await. Главное отличие: если хотя бы один метод внутри компонента асинхронный, вся цепочка вызовов должна быть асинхронной. Это заставляет разработчиков использовать await в тестах и поддерживать единообразие API.

PageObject

И наконец разберём, как изменяются сами объекты страниц. Для примера возьмём страницу регистрации нового пользователя и посмотрим, как она переписывается под использование async/await.

Синхронная реализация: /pages/registration_page.py

import re

from playwright.sync_api import Page

from components.registration_form_component import RegistrationFormComponent
from elements.button import Button
from elements.link import Link
from pages.base_page import BasePage


class RegistrationPage(BasePage):
    def __init__(self, page: Page):
        super().__init__(page)

        self.registration_form = RegistrationFormComponent(page)

        self.login_link = Link(page, "registration-page-login-link", "Login")
        self.registration_button = Button(page, "registration-page-registration-button", "Registration")

    def click_registration_button(self):
        self.registration_button.click()
        self.check_current_url(re.compile(".*/#/dashboard"))

Асинхронная реализация: /pages/registration_page.py

import re

from playwright.async_api import Page

from components.registration_form_component import RegistrationFormComponent
from elements.button import Button
from elements.link import Link
from pages.base_page import BasePage


class RegistrationPage(BasePage):
    def __init__(self, page: Page):
        super().__init__(page)

        self.registration_form = RegistrationFormComponent(page)

        self.login_link = Link(page, "registration-page-login-link", "Login")
        self.registration_button = Button(page, "registration-page-registration-button", "Registration")

    async def click_registration_button(self):
        await self.registration_button.click()
        await self.check_current_url(re.compile(".*/#/dashboard"))

Те же изменения применяются и к PageObject: если страница вызывает асинхронные компоненты или элементы, её методы тоже должны быть асинхронными (async def click_registration_button). Более того, базовые проверки (check_current_url) или переходы по страницам также должны быть переписаны на async‑варианты, если внутри они обращаются к Playwright API.

Итог

После этих изменений структура проекта осталась прежней (есть элементы → компоненты → страницы → тесты), но вся цепочка вызовов стала асинхронной. Это ключевой момент: нельзя просто сделать один метод async и оставить остальные синхронными — чтобы не блокировать поток, вся «вертикаль» вызовов должна быть переписана на async. При этом использование остаётся привычным: в тестах вы по‑прежнему пишете «нажать кнопку, проверить текст», но добавляете await, и тесты теперь могут выполняться эффективнее при большом числе взаимодействий с браузером.

Тесты

Перед тем как перейти к самим тестам, нужно немного изменить конфигурацию pytest. В файле pytest.ini появляется параметр asyncio_mode = auto, чтобы pytest-asyncio корректно обрабатывал асинхронные тесты. Также стоит добавить настройку asyncio_default_fixture_loop_scope = function, чтобы каждый тест выполнялся в собственном событийном цикле, изолированном от других тестов.

pytest.ini

[pytest]
addopts = -s -v
asyncio_mode = auto
python_files = *_tests.py test_*.py
python_classes = Test*
python_functions = test_*
asyncio_default_fixture_loop_scope = function
markers =
    regression: Маркировка для регрессионных тестов.
    registration: Маркировка для тестов, связанных с регистрацией пользователей.

Теперь разберём, что именно изменилось в самих тестах:

Синхронная реализация: /tests/test_registration.py

import allure
import pytest

from pages.dashboard_page import DashboardPage
from pages.registration_page import RegistrationPage
from tools.routes import AppRoute


@pytest.mark.regression
@pytest.mark.registration
class TestRegistration:
    @allure.title("Successful registration")
    def test_successful_registration(
            self,
            dashboard_page: DashboardPage,
            registration_page: RegistrationPage
    ):
        registration_page.visit(AppRoute.REGISTRATION)
        registration_page.registration_form.check_visible(email="", username="", password="")
        registration_page.registration_form.fill(
            email="user@example.com",
            username="Playwright",
            password="qwerty"
        )
        registration_page.click_registration_button()

        dashboard_page.navbar.check_visible("Playwright")
        dashboard_page.dashboard_toolbar_view.check_visible()
        dashboard_page.check_visible_scores_chart()
        dashboard_page.check_visible_courses_chart()
        dashboard_page.check_visible_students_chart()
        dashboard_page.check_visible_activities_chart()

Асинхронная реализация: /tests/test_registration.py

import allure
import pytest

from pages.dashboard_page import DashboardPage
from pages.registration_page import RegistrationPage
from tools.routes import AppRoute


@pytest.mark.asyncio
@pytest.mark.regression
@pytest.mark.registration
class TestRegistration:
    @allure.title("Successful registration")
    async def test_successful_registration(
            self,
            dashboard_page: DashboardPage,
            registration_page: RegistrationPage
    ):
        await registration_page.visit(AppRoute.REGISTRATION)
        await registration_page.registration_form.check_visible(email="", username="", password="")
        await registration_page.registration_form.fill(
            email="user@example.com",
            username="Playwright",
            password="qwerty"
        )
        await registration_page.click_registration_button()

        await dashboard_page.navbar.check_visible("Playwright")
        await dashboard_page.dashboard_toolbar_view.check_visible()
        await dashboard_page.check_visible_scores_chart()
        await dashboard_page.check_visible_courses_chart()
        await dashboard_page.check_visible_students_chart()
        await dashboard_page.check_visible_activities_chart()

В синхронной версии тест просто вызывал методы объектов страниц. Все операции (переход по маршруту, заполнение формы, клик по кнопке, проверки элементов) выполнялись последовательно и блокировали поток до завершения каждого действия.

В асинхронной версии ключевые изменения следующие:

  • Методы теста и все вызываемые им операции теперь асинхронные, поэтому тестовая функция объявляется как async def, а все вызовы методов, которые возвращают корутины, оборачиваются в await.

  • Появилась маркировка @pytest.mark.asyncio, которая сообщает pytest, что этот тест должен выполняться внутри событийного цикла.

  • Все взаимодействия с элементами страницы (переход на страницу, заполнение формы, нажатие кнопки, проверки) теперь не блокируют основной поток — вместо этого управление возвращается в event loop, что позволяет при желании выполнять несколько асинхронных операций параллельно (например, если запускать несколько таких тестов).

Важно подчеркнуть, что с точки зрения бизнес-логики теста ничего не изменилось: сценарий по-прежнему описывает успешную регистрацию пользователя и проверки отображения элементов. Однако теперь все действия выполняются асинхронно и совместимы с другими async-фикстурами и компонентами, которые мы адаптировали ранее.

Итог

Несмотря на переход на асинхронный стиль, сам тест выполняется примерно за то же время, что и его синхронная версия. Это ожидаемый результат: сценарий состоит из шагов, каждый из которых зависит от завершения предыдущего (переход на страницу, заполнение формы, клик, проверки), и они выполняются строго последовательно.

Асинхронность в данном случае не делает тесты «мгновенно быстрыми». Её задача другая — обеспечить совместимость с другими асинхронными компонентами (например, с асинхронными API-клиентами или библиотеками), чтобы в проекте можно было использовать единый событийный цикл. Это полезно, когда тестируемый код сам по себе работает в async-контексте, либо когда тесты действительно выполняют параллельные I/O‑операции внутри одного теста.

В случае же классических UI-сценариев, где всё строится как цепочка шагов «открыть → ввести данные → нажать кнопку → проверить результат», выигрыш во времени практически отсутствует — сценарий остаётся линейным, и каждый шаг должен завершиться прежде, чем начнётся следующий.

API автотесты

Если в случае с UI тестами мы фактически только подстраивались под уже существующий асинхронный API Playwright, то в API тестах ситуация немного иная: именно наши API‑клиенты управляют сетью и формируют большую часть времени выполнения теста. И это кажется более логичным местом для применения async/await, ведь сетевые вызовы — это типичные I/O bound операции.

Однако важно понимать, что сама по себе перепись клиентов на async и замена HTTP‑библиотеки на асинхронную (в нашем случае httpx.AsyncClient) не делает тесты «магически быстрыми». Если тесты по‑прежнему идут один за другим (а pytest именно так их и выполняет), выигрыш в скорости может и не появиться. Асинхронный код в этом случае полезен прежде всего для:

  • ситуаций, когда сам тестируемый код уже асинхронный (например, WebSocket или стриминг);

  • случаев, когда вам нужно запустить несколько операций в рамках одного теста параллельно (например, моделировать несколько одновременных клиентов).

В остальном структура работы очень похожа на UI: фикстуры становятся асинхронными, HTTP‑клиенты используют await, а тесты помечаются маркером @pytest.mark.asyncio.

Полные примеры реализации API тестов вы можете найти в репозиториях:

API клиенты

Движком API тестов являются API‑клиенты. Именно они выполняют сетевые вызовы и, как следствие, являются I/O‑bound операциями. Такие вызовы не нагружают процессор, а в основном ждут ответа от сервера, поэтому именно на них асинхронность влияет больше всего.

BaseClient

Мы используем библиотеку HTTPX, которая уже поддерживает асинхронный режим работы через класс AsyncClient. Поэтому первым шагом для миграции на async/await является адаптация базового клиента: перевод его методов на асинхронные и использование асинхронного клиента вместо синхронного.

Синхронная реализация: /clients/base_client.py

from typing import Any

import allure
from httpx import Client, URL, Response, QueryParams
from httpx._types import RequestData, RequestFiles

from clients.event_hooks import log_request_event_hook, log_response_event_hook
from config import HTTPClientConfig


class BaseClient:
    def __init__(self, client: Client):
        self.client = client

    @allure.step("Make GET request to {url}")
    def get(self, url: URL | str, params: QueryParams | None = None) -> Response:
        return self.client.get(url, params=params)

    @allure.step("Make POST request to {url}")
    def post(
            self,
            url: URL | str,
            json: Any | None = None,
            data: RequestData | None = None,
            files: RequestFiles | None = None
    ) -> Response:
        return self.client.post(url, json=json, data=data, files=files)

    @allure.step("Make PATCH request to {url}")
    def patch(self, url: URL | str, json: Any | None = None) -> Response:
        return self.client.patch(url, json=json)

    @allure.step("Make DELETE request to {url}")
    def delete(self, url: URL | str) -> Response:
        return self.client.delete(url)


def get_http_client(config: HTTPClientConfig) -> Client:
    return Client(
        timeout=config.timeout,
        base_url=config.client_url,
        event_hooks={
            "request": [log_request_event_hook],
            "response": [log_response_event_hook]
        }
    )

Асинхронная реализация: /clients/base_client.py

from typing import Any

import allure
from httpx import AsyncClient, URL, Response, QueryParams
from httpx._types import RequestData, RequestFiles

from clients.event_hooks import log_request_event_hook, log_response_event_hook
from config import HTTPClientConfig


class BaseClient:
    def __init__(self, client: AsyncClient):
        self.client = client

    @allure.step("Make GET request to {url}")
    async def get(self, url: URL | str, params: QueryParams | None = None) -> Response:
        return await self.client.get(url, params=params)

    @allure.step("Make POST request to {url}")
    async def post(
            self,
            url: URL | str,
            json: Any | None = None,
            data: RequestData | None = None,
            files: RequestFiles | None = None
    ) -> Response:
        return await self.client.post(url, json=json, data=data, files=files)

    @allure.step("Make PATCH request to {url}")
    async def patch(self, url: URL | str, json: Any | None = None) -> Response:
        return await self.client.patch(url, json=json)

    @allure.step("Make DELETE request to {url}")
    async def delete(self, url: URL | str) -> Response:
        return await self.client.delete(url)


def get_http_client(config: HTTPClientConfig) -> AsyncClient:
    return AsyncClient(
        timeout=config.timeout,
        base_url=config.client_url,
        event_hooks={
            "request": [log_request_event_hook],
            "response": [log_response_event_hook]
        }
    )

В синхронной реализации мы использовали httpx.Client, а вызовы методов (get, post, patch, delete) блокировали выполнение теста до получения ответа. В асинхронной версии:

  • используется httpx.AsyncClient, и все методы клиента стали асинхронными (async def …);

  • каждый сетевой вызов теперь выполняется с ключевым словом await, чтобы не блокировать event loop;

  • фабрика клиента (get_http_client) теперь возвращает асинхронный клиент, но интерфейс для тестов остаётся прежним — сигнатуры методов не меняются, меняется только способ вызова.

Благодаря этому тесты, написанные с использованием этого клиента, можно выполнять внутри асинхронного event loop без переписывания всей бизнес‑логики.

Event hooks

Следующий момент, о котором легко забыть — это event hooks. В HTTPX можно подписываться на события отправки запроса и получения ответа. Эти хуки тоже нужно адаптировать под асинхронную работу, иначе они могут блокировать выполнение event loop.

Синхронная реализация: /clients/event_hooks.py

from httpx import Request, Response

from tools.logger import get_logger

logger = get_logger("HTTP_CLIENT")


def log_request_event_hook(request: Request):
    logger.info(f'Make {request.method} request to {request.url}')


def log_response_event_hook(response: Response):
    logger.info(
        f"Got response {response.status_code} {response.reason_phrase} from {response.url}"
    )

Асинхронная реализация: /clients/event_hooks.py

from httpx import Request, Response

from tools.logger import get_logger

logger = get_logger("HTTP_CLIENT")


async def log_request_event_hook(request: Request):
    logger.info(f'Make {request.method} request to {request.url}')


async def log_response_event_hook(response: Response):
    logger.info(
        f"Got response {response.status_code} {response.reason_phrase} from {response.url}"
    )

В исходной реализации хуки log_request_event_hook и log_response_event_hook были обычными синхронными функциями. Теперь они объявлены с async def, что позволяет корректно интегрироваться с AsyncClient, который также работает в асинхронном контексте. Фактически логика внутри функций не изменилась (мы по‑прежнему просто логируем события), но сигнатура должна быть асинхронной, чтобы HTTPX мог корректно их вызывать в event loop и не блокировать выполнение других корутин.

OperationsClient

После адаптации базового клиента и хуков можно переходить к конкретным API‑клиентам. На примере OperationsClient видно, что изменения здесь минимальны, но важны: все сетевые вызовы теперь выполняются асинхронно.

Синхронная реализация: /clients/operations_client.py

import allure
from httpx import Response

from clients.base_client import BaseClient, get_http_client
from config import Settings
from schema.operations import CreateOperationSchema, UpdateOperationSchema, OperationSchema
from tools.routes import APIRoutes


class OperationsClient(BaseClient):
    @allure.step("Get list of operations")
    def get_operations_api(self) -> Response:
        return self.get(APIRoutes.OPERATIONS)

    @allure.step("Get operation by id {operation_id}")
    def get_operation_api(self, operation_id: int) -> Response:
        return self.get(f"{APIRoutes.OPERATIONS}/{operation_id}")

    @allure.step("Create operation")
    def create_operation_api(self, operation: CreateOperationSchema) -> Response:
        return self.post(
            APIRoutes.OPERATIONS,
            json=operation.model_dump(mode='json', by_alias=True)
        )

    @allure.step("Update operation by id {operation_id}")
    def update_operation_api(
            self,
            operation_id: int,
            operation: UpdateOperationSchema
    ) -> Response:
        return self.patch(
            f"{APIRoutes.OPERATIONS}/{operation_id}",
            json=operation.model_dump(mode='json', by_alias=True, exclude_none=True)
        )

    @allure.step("Delete operation by id {operation_id}")
    def delete_operation_api(self, operation_id: int) -> Response:
        return self.delete(f"{APIRoutes.OPERATIONS}/{operation_id}")

    def create_operation(self) -> OperationSchema:
        request = CreateOperationSchema()
        response = self.create_operation_api(request)
        return OperationSchema.model_validate_json(response.text)


def get_operations_client(settings: Settings) -> OperationsClient:
    return OperationsClient(client=get_http_client(settings.fake_bank_http_client))

Асинхронная реализация: /clients/operations_client.py

import allure
from httpx import Response

from clients.base_client import BaseClient, get_http_client
from config import Settings
from schema.operations import CreateOperationSchema, UpdateOperationSchema, OperationSchema
from tools.routes import APIRoutes


class OperationsClient(BaseClient):
    @allure.step("Get list of operations")
    async def get_operations_api(self) -> Response:
        return await self.get(APIRoutes.OPERATIONS)

    @allure.step("Get operation by id {operation_id}")
    async def get_operation_api(self, operation_id: int) -> Response:
        return await self.get(f"{APIRoutes.OPERATIONS}/{operation_id}")

    @allure.step("Create operation")
    async def create_operation_api(self, operation: CreateOperationSchema) -> Response:
        return await self.post(
            APIRoutes.OPERATIONS,
            json=operation.model_dump(mode='json', by_alias=True)
        )

    @allure.step("Update operation by id {operation_id}")
    async def update_operation_api(
            self,
            operation_id: int,
            operation: UpdateOperationSchema
    ) -> Response:
        return await self.patch(
            f"{APIRoutes.OPERATIONS}/{operation_id}",
            json=operation.model_dump(mode='json', by_alias=True, exclude_none=True)
        )

    @allure.step("Delete operation by id {operation_id}")
    async def delete_operation_api(self, operation_id: int) -> Response:
        return await self.delete(f"{APIRoutes.OPERATIONS}/{operation_id}")

    async def create_operation(self) -> OperationSchema:
        request = CreateOperationSchema()
        response = await self.create_operation_api(request)
        return OperationSchema.model_validate_json(response.text)


def get_operations_client(settings: Settings) -> OperationsClient:
    return OperationsClient(client=get_http_client(settings.fake_bank_http_client))

  • Все методы, которые делают HTTP‑запросы (get_operations_api, get_operation_api, create_operation_api, update_operation_api, delete_operation_api) стали асинхронными и используют await для вызова базовых методов клиента.

  • Вспомогательный метод create_operation, который внутри себя вызывает API, также стал асинхронным и ждёт выполнения сетевого вызова.

  • Фабрика клиента get_operations_client не изменила интерфейс, но теперь возвращает клиент на основе AsyncClient.

Ключевой момент: интерфейс использования клиента в тестах остаётся почти прежним, меняется только то, что все вызовы нужно выполнять с await. Это позволяет легко перейти на новый режим, не меняя общую архитектуру тестового фреймворка.

Фикстуры

Так как API‑клиент теперь полностью асинхронный, необходимо адаптировать и связанные с ним фикстуры. В противном случае тесты не смогут корректно работать с await‑вызовами, и pytest выдаст ошибку выполнения. Давайте посмотрим, как изменить фикстуры, чтобы они поддерживали асинхронность.

Синхронная реализация: /fixtures/operations.py

import pytest

from clients.operations_client import OperationsClient, get_operations_client
from config import Settings
from schema.operations import OperationSchema


@pytest.fixture
def operations_client(settings: Settings) -> OperationsClient:
    return get_operations_client(settings)


@pytest.fixture
def function_operation(operations_client: OperationsClient) -> OperationSchema:
    operation = operations_client.create_operation()
    yield operation

    operations_client.delete_operation_api(operation.id)

Асинхронная реализация: /fixtures/operations.py

import pytest
import pytest_asyncio

from clients.operations_client import OperationsClient, get_operations_client
from config import Settings
from schema.operations import OperationSchema


@pytest.fixture
def operations_client(settings: Settings) -> OperationsClient:
    return get_operations_client(settings)


@pytest_asyncio.fixture
async def function_operation(operations_client: OperationsClient) -> OperationSchema:
    operation = await operations_client.create_operation()
    yield operation

    await operations_client.delete_operation_api(operation.id)

После перехода на асинхронные API‑клиенты нам пришлось изменить подход и к фикстурам. В синхронной версии фикстура function_operation вызывала методы клиента напрямую, и тесты выполнялись в обычном синхронном режиме.

В асинхронной версии изменений немного, но они принципиальны:

  • Для фикстуры function_operation используется декоратор @pytest_asyncio.fixture вместо стандартного @pytest.fixture. Это необходимо, потому что сама фикстура теперь объявлена как async def и внутри себя вызывает асинхронные методы клиента.

  • Создание и удаление операции теперь выполняются с ключевым словом await, что позволяет не блокировать event loop.

  • Фикстура operations_client остаётся синхронной, потому что она возвращает уже готовый клиент, который не требует асинхронной инициализации.

Ключевая идея: Мы не переписываем все фикстуры подряд, а только те, которые напрямую взаимодействуют с асинхронным кодом (в нашем случае с вызовами API). Всё остальное может оставаться синхронным, что упрощает миграцию и делает её постепенной.

Тесты

Так как теперь и клиент, и фикстуры работают асинхронно, тесты тоже нужно адаптировать.

Синхронная реализация: /tests/test_operations.py

from http import HTTPStatus

import allure
import pytest

from clients.operations_client import OperationsClient
from schema.operations import OperationsSchema, OperationSchema, CreateOperationSchema, UpdateOperationSchema
from tools.assertions.base import assert_status_code
from tools.assertions.operations import assert_operation, assert_create_operation
from tools.assertions.schema import validate_json_schema


@pytest.mark.operations
@pytest.mark.regression
class TestOperations:
    @allure.title("Get operations")
    def test_get_operations(self, operations_client: OperationsClient):
        response = operations_client.get_operations_api()

        assert_status_code(response.status_code, HTTPStatus.OK)
        validate_json_schema(response.json(), OperationsSchema.model_json_schema())

    @allure.title("Get operation")
    def test_get_operation(
            self,
            operations_client: OperationsClient,
            function_operation: OperationSchema
    ):
        response = operations_client.get_operation_api(function_operation.id)
        operation = OperationSchema.model_validate_json(response.text)

        assert_status_code(response.status_code, HTTPStatus.OK)
        assert_operation(operation, function_operation)

        validate_json_schema(response.json(), operation.model_json_schema())

    @allure.title("Create operation")
    def test_create_operation(self, operations_client: OperationsClient):
        request = CreateOperationSchema()
        response = operations_client.create_operation_api(request)
        operation = OperationSchema.model_validate_json(response.text)

        assert_status_code(response.status_code, HTTPStatus.CREATED)
        assert_create_operation(operation, request)

        validate_json_schema(response.json(), operation.model_json_schema())

    @allure.title("Update operation")
    def test_update_operation(
            self,
            operations_client: OperationsClient,
            function_operation: OperationSchema
    ):
        request = UpdateOperationSchema()
        response = operations_client.update_operation_api(function_operation.id, request)
        operation = OperationSchema.model_validate_json(response.text)

        assert_status_code(response.status_code, HTTPStatus.OK)
        assert_create_operation(operation, request)

        validate_json_schema(response.json(), operation.model_json_schema())

    @allure.title("Delete operation")
    def test_delete_operation(
            self,
            operations_client: OperationsClient,
            function_operation: OperationSchema
    ):
        delete_response = operations_client.delete_operation_api(function_operation.id)
        assert_status_code(delete_response.status_code, HTTPStatus.OK)

        get_response = operations_client.get_operation_api(function_operation.id)
        assert_status_code(get_response.status_code, HTTPStatus.NOT_FOUND)

Асинхронная реализация: /tests/test_operations.py

from http import HTTPStatus

import allure
import pytest

from clients.operations_client import OperationsClient
from schema.operations import OperationsSchema, OperationSchema, CreateOperationSchema, UpdateOperationSchema
from tools.assertions.base import assert_status_code
from tools.assertions.operations import assert_operation, assert_create_operation
from tools.assertions.schema import validate_json_schema


@pytest.mark.asyncio
@pytest.mark.operations
@pytest.mark.regression
class TestOperations:
    @allure.title("Get operations")
    async def test_get_operations(self, operations_client: OperationsClient):
        response = await operations_client.get_operations_api()

        assert_status_code(response.status_code, HTTPStatus.OK)
        validate_json_schema(response.json(), OperationsSchema.model_json_schema())

    @allure.title("Get operation")
    async def test_get_operation(
            self,
            operations_client: OperationsClient,
            function_operation: OperationSchema
    ):
        response = await operations_client.get_operation_api(function_operation.id)
        operation = OperationSchema.model_validate_json(response.text)

        assert_status_code(response.status_code, HTTPStatus.OK)
        assert_operation(operation, function_operation)

        validate_json_schema(response.json(), operation.model_json_schema())

    @allure.title("Create operation")
    async def test_create_operation(self, operations_client: OperationsClient):
        request = CreateOperationSchema()
        response = await operations_client.create_operation_api(request)
        operation = OperationSchema.model_validate_json(response.text)

        assert_status_code(response.status_code, HTTPStatus.CREATED)
        assert_create_operation(operation, request)

        validate_json_schema(response.json(), operation.model_json_schema())

    @allure.title("Update operation")
    async def test_update_operation(
            self,
            operations_client: OperationsClient,
            function_operation: OperationSchema
    ):
        request = UpdateOperationSchema()
        response = await operations_client.update_operation_api(function_operation.id, request)
        operation = OperationSchema.model_validate_json(response.text)

        assert_status_code(response.status_code, HTTPStatus.OK)
        assert_create_operation(operation, request)

        validate_json_schema(response.json(), operation.model_json_schema())

    @allure.title("Delete operation")
    async def test_delete_operation(
            self,
            operations_client: OperationsClient,
            function_operation: OperationSchema
    ):
        delete_response = await operations_client.delete_operation_api(function_operation.id)
        assert_status_code(delete_response.status_code, HTTPStatus.OK)

        get_response = await operations_client.get_operation_api(function_operation.id)
        assert_status_code(get_response.status_code, HTTPStatus.NOT_FOUND)

Главные изменения:

  1. Асинхронные функции тестов. Методы класса тестов стали async def, чтобы внутри них можно было использовать ключевое слово await. Это обязательное условие: без него нельзя вызвать асинхронные методы API‑клиента (get_operations_api, create_operation_api и т. д.).

  2. Добавление маркера @pytest.mark.asyncio. Этот маркер сообщает pytest, что тест должен выполняться в событийном цикле, который предоставляет плагин pytest‑asyncio. Без маркера тест упадёт с ошибкой вида «RuntimeError: no running event loop».

  3. Асинхронные вызовы API‑клиента. Все обращения к клиенту теперь сопровождаются await, потому что методы клиента (get, post, patch, delete) и их обёртки в OperationsClient стали асинхронными. Соответственно, выполнение запросов не блокирует event loop, и другие корутины в нём могут выполняться параллельно.

  4. Логика тестов осталась прежней. Проверки ответов (assert_status_code, validate_json_schema, сравнение схем и данных) не изменились — они остаются синхронными, потому что это просто операции в памяти. Меняется только то, как вызываются методы клиента и как сам тест «подвешен» к event loop.

Важно: как и в случае с UI‑тестами, время выполнения тестов при переходе на async/await не изменится «автоматически». Асинхронность нужна только потому, что базовые вызовы клиента теперь сами по себе асинхронные (работают через httpx.AsyncClient). Если бы клиент оставался синхронным, никакого выигрыша по времени не было бы, а наоборот — возникали бы проблемы (нельзя вызывать блокирующий код из асинхронных тестов без дополнительных обёрток).

Делаем выводы

Теперь, когда у нас перед глазами есть два полноценных примера — синхронные и асинхронные UI и API тесты, — можно трезво оценить, что даёт эта «магия async/await» на практике.

Скажу честно: никакого «бесплатного ускорения» мы не получили. Я прогнал оба набора тестов на CI/CD, и они завершились за одно и то же время, что в синхронной, что в асинхронной версии. Результаты можно посмотреть в GitHub Actions:

Почему так?

Async/await — это не параллельное выполнение тестов. Это механизм кооперативной многозадачности: пока одна корутина ждёт I/O (например, ответа от API), event loop может переключиться на другую. Но всё это всё равно работает в одном потоке, а Pytest по своей природе запускает тесты последовательно. Даже если добавить плагины, позволяющие группировать корутины и исполнять их «пакетами», весь прогон от этого не становится параллельным.

На первый взгляд кажется логичным: «Пусть тесты в момент ожидания ответа от медленного эндпоинта выполняют другие тесты». Но на практике это требует полной перестройки фреймворка, дополнительных плагинов и жизни с ограничениями одного event loop. Результат — рост числа одновременных соединений, нестабильность, сложность отладки и логирования. Выигрыш (если он есть) — ограничен и почти никогда не перекрывает стоимость поддержки.

А если пойти ещё дальше и использовать плагины, которые позволяют выполнять несколько тестов внутри одного event loop (например, pytest-asyncio-concurrent), то можно быстро оказаться в так называемом async hell. Когда десятки корутин одновременно держат открытые соединения, одно подвисшее или неправильно реализованное действие способно «заморозить» весь поток. Диагностика ошибок превращается в кошмар, а падение одной корутины может обрушить всю группу тестов сразу. Итог: рост нестабильности и сложности поддержки, а выигрыш по времени остаётся минимальным.

Когда async действительно уместен?

Тогда, когда ваш тестируемый код изначально асинхронный: WebSocket, gRPC streaming, свой asyncio-клиент, когда сама используемая библиотека изначально построена на асинхронной архитектуре, или когда проект сам живёт в async-контексте. В таком случае логично писать и тесты в том же стиле, чтобы не блокировать event loop. Но использовать async «ради ускорения» на обычных HTTP-тестах — это иллюзия бесплатного прироста.

Итого

Если ваша цель — скорость прогона, то async вам её не даст. Для этого используйте многопроцессный параллелизм (xdist) или оптимизируйте сами тесты и инфраструктуру. Async в тестах — инструмент для совместимости, а не для ускорения.

Заключение

Асинхронность в тестах выглядит как интересный эксперимент и может казаться «бесплатным» способом ускорить выполнение. На практике это не так. Да, мы можем переписать UI и API тесты под async/await, и они будут работать корректно. Мы даже получим плюс в тех случаях, когда сам код, который мы тестируем, изначально асинхронный (например, WebSocket-клиенты или стриминговые API). Но в классических сценариях — вызовы REST API, UI‑тесты через Playwright — прироста нет.

Причина проста: Pytest по умолчанию выполняет тесты последовательно, и даже с async/await они идут один за другим. Чтобы реально ускорить выполнение, чаще всего используют pytest-xdist, который запускает несколько воркеров в отдельных процессах и реально распределяет нагрузку по ядрам процессора.

Асинхронный подход может быть оправдан в проектах, которые и так работают в async-контексте (например, весь код на FastAPI или aiohttp). В таких случаях тесты будут выглядеть естественнее, а не «оборачивать асинхронное API в синхронные вызовы». Но переписывать стабильные синхронные тесты на async только ради гипотетического ускорения — затея сомнительная.

В этом эксперименте мы показали:

  • как переписать тестовый фреймворк на async/await;

  • что это действительно работает и интегрируется с Pytest и Allure;

  • что без изменения парадигмы запуска (xdist, event-loop batching) ускорения не происходит.

Итог: async/await в тестах — это не волшебная кнопка «ускорить всё». Это инструмент для специфических задач, который имеет смысл использовать, когда того требует архитектура, а не ради «попробовать модный синтаксис».

Все примеры кода и результаты прогонов CI/CD доступны на моём GitHub:

Комментарии (0)