Тестирование

Тестирование потребителей Channels немного сложнее, чем тестирование обычных представлений Django, из-за их асинхронной природы.

Чтобы помочь в тестировании, Channels предоставляет тестовые помощники под названием Коммуникаторы, которые позволяют вам обернуть ASGI-приложение (например, потребитель) в его собственный цикл событий и задать ему вопросы.

Since Django 3.1, you can test asynchronous code с помощью TestCase от Django. В качестве альтернативы вы можете использовать pytest с его плагином asyncio.

Настройка асинхронных тестов

Чтобы использовать TestCase Django, достаточно определить async def метод тестирования, чтобы обеспечить соответствующий асинхронный контекст:

from django.test import TestCase
from channels.testing import HttpCommunicator
from myproject.myapp.consumers import MyConsumer

class MyTests(TestCase):
    async def test_my_consumer(self):
        communicator = HttpCommunicator(MyConsumer, "GET", "/test/")
        response = await communicator.get_response()
        self.assertEqual(response["body"], b"test response")
        self.assertEqual(response["status"], 200)

Чтобы использовать pytest, вам нужно установить его с поддержкой асинхронных тестов и, предположительно, с поддержкой тестов Django. Вы можете сделать это, установив пакеты pytest-django и << 2 >>>:

python -m pip install -U pytest-django pytest-asyncio

Затем вам нужно украсить тесты, которые вы хотите запускать async с помощью pytest.mark.asyncio. Обратите внимание, что вы не можете смешивать это с подклассами unittest.TestCase; вы должны писать тесты async как тестовые функции верхнего уровня в родном стиле pytest:

import pytest
from channels.testing import HttpCommunicator
from myproject.myapp.consumers import MyConsumer

@pytest.mark.asyncio
async def test_my_consumer():
    communicator = HttpCommunicator(MyConsumer, "GET", "/test/")
    response = await communicator.get_response()
    assert response["body"] == b"test response"
    assert response["status"] == 200

Существует несколько вариантов коммуникатора - обычный для общего использования, и по одному для HTTP и WebSockets, которые имеют методы быстрого доступа,

ПриложениеКоммуникатор

ApplicationCommunicator - это общий помощник тестирования для любого ASGI-приложения. Он предоставляет несколько основных методов для взаимодействия, как описано ниже.

Этот общий класс нужен только для тестов, не связанных с HTTP/WebSocket, хотя, возможно, вам придется вернуться к нему, если вы тестируете такие вещи, как HTTP chunked responses или long-polling, которые пока не поддерживаются в HttpCommunicator.

Примечание

ApplicationCommunicator фактически предоставляется базовым пакетом asgiref, но мы позволяем вам импортировать его из channels.testing для удобства.

Чтобы создать его, передайте ему приложение и область видимости:

from channels.testing import ApplicationCommunicator
communicator = ApplicationCommunicator(MyConsumer, {"type": "http", ...})

send_input

Вызовите его, чтобы отправить событие в приложение:

await communicator.send_input({
    "type": "http.request",
    "body": b"chunk one \x01 chunk two",
})

receive_output

Вызовите его для получения события от приложения:

event = await communicator.receive_output(timeout=1)
assert event["type"] == "http.response.start"

получать_ничего

Вызовите его, чтобы проверить, что нет события, ожидающего получения от приложения:

assert await communicator.receive_nothing(timeout=0.1, interval=0.01) is False
# Receive the rest of the http request from above
event = await communicator.receive_output()
assert event["type"] == "http.response.body"
assert event.get("more_body") is True
event = await communicator.receive_output()
assert event["type"] == "http.response.body"
assert event.get("more_body") is None
# Check that there isn't another event
assert await communicator.receive_nothing() is True
# You could continue to send and receive events
# await communicator.send_input(...)

Метод имеет два необязательных параметра:

  • timeout: количество секунд ожидания, чтобы убедиться, что очередь пуста. По умолчанию равно 0,1.

  • interval: количество секунд для ожидания очередной проверки на наличие новых событий. По умолчанию равно 0,01.

подождите

Вызовите его для ожидания выхода приложения (вам нужно будет либо сделать это, либо подождать, пока оно отправит вам вывод, прежде чем вы сможете увидеть, что оно сделало, используя макеты или инспекцию):

await communicator.wait(timeout=1)

Если вы ожидаете, что ваше приложение вызовет исключение, используйте pytest.raises вокруг wait:

with pytest.raises(ValueError):
    await communicator.wait()

HttpCommunicator

HttpCommunicator - это подкласс ApplicationCommunicator, специально разработанный для HTTP-запросов. Вам нужно только инстанцировать его с нужными вам параметрами:

from channels.testing import HttpCommunicator
communicator = HttpCommunicator(MyHttpConsumer, "GET", "/test/")

А затем дождитесь его реакции:

response = await communicator.get_response()
assert response["body"] == b"test response"

Вы можете передать конструктору следующие аргументы:

  • method: Имя метода HTTP (строка unicode, обязательно)

  • path: HTTP-путь (строка unicode, обязательно)

  • body: тело HTTP (байтстринг, необязательно)

Ответом метода get_response будет диктант со следующими ключами:

  • status: код статуса HTTP (целое число)

  • headers: Список заголовков в виде кортежей (имя, значение) (оба байтстринги)

  • body: тело ответа HTTP (байтстринг)

WebsocketCommunicator

WebsocketCommunicator позволяет вам легче тестировать потребителей WebSocket. Он предоставляет несколько удобных методов для взаимодействия с WebSocket-приложением, как показано в этом примере:

from channels.testing import WebsocketCommunicator
communicator = WebsocketCommunicator(SimpleWebsocketApp.as_asgi(), "/testws/")
connected, subprotocol = await communicator.connect()
assert connected
# Test sending text
await communicator.send_to(text_data="hello")
response = await communicator.receive_from()
assert response == "hello"
# Close
await communicator.disconnect()

Примечание

Все эти методы являются корутинами, что означает, что вы должны await их выполнять. Если вы этого не сделаете, ваш тест либо прервется (если вы забыли ожидать send), либо попытается сравнить вещи с объектом coroutine (если вы забыли ожидать receive).

Важно

Если вы не вызовете WebsocketCommunicator.disconnect() перед завершением набора тестов, вы можете получить RuntimeWarnings о том, что ничего не дождались, поскольку вы завершаете свое приложение в середине его жизненного цикла. Однако не нужно вызывать disconnect(), если ваше приложение уже вызвало ошибку.

Вы также можете передать application, построенный с URLRouter, вместо обычного потребительского класса. Это позволяет тестировать приложения, требующие позиционных или ключевых аргументов в scope:

from channels.testing import WebsocketCommunicator
application = URLRouter([
    url(r"^testws/(?P<message>\w+)/$", KwargsWebSocketApp.as_asgi()),
])
communicator = WebsocketCommunicator(application, "/testws/test/")
connected, subprotocol = await communicator.connect()
assert connected
# Test on connection welcome message
message = await communicator.receive_from()
assert message == 'test'
# Close
await communicator.disconnect()

Примечание

Поскольку класс WebsocketCommunicator принимает URL в своем конструкторе, один коммуникатор может тестировать только один URL. Если вы хотите проверить несколько различных URL, используйте несколько коммуникаторов.

подключить

Запускает фазу соединения WebSocket и ждет, пока приложение примет или отклонит соединение. Не принимает никаких параметров и возвращает любой из них:

  • (True, <chosen_subprotocol>) если сокет был принят. chosen_subprotocol по умолчанию None.

  • (False, <close_code>) если сокет был отклонен. close_code по умолчанию 1000.

отправить_к

Отправляет кадр данных в приложение. Принимает в качестве параметров ровно одно из bytes_data или text_data и ничего не возвращает:

await communicator.send_to(bytes_data=b"hi\0")

Этот метод проверит тип ваших параметров, чтобы убедиться, что то, что вы отправляете, действительно является текстом или байтами.

send_json_to

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

await communicator.send_json_to({"hello": "world"})

получать_от

Получает кадр от приложения и возвращает вам либо bytes, либо text в зависимости от типа кадра:

response = await communicator.receive_from()

Принимает необязательный аргумент timeout с числом секунд ожидания до завершения работы, которое по умолчанию равно 1. Он также будет проверять типы ответов вашего приложения, чтобы убедиться, что текстовые фреймы содержат текстовые данные, а двоичные фреймы - двоичные данные.

получать_json_от

Получает текстовый кадр от приложения и декодирует его для вас:

response = await communicator.receive_json_from()
assert response == {"hello": "world"}

Принимает необязательный аргумент timeout с числом секунд ожидания до завершения тайм-аута, которое по умолчанию равно 1.

получать_ничего

Проверяет, что нет кадра, ожидающего получения от приложения. Подробнее см. в разделе ApplicationCommunicator.

отключить

Закрывает сокет со стороны клиента. Ничего не принимает и ничего не возвращает.

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

ChannelsLiveServerTestCase

Если вы просто хотите запустить стандартные Selenium или другие тесты, которые требуют запуска веб-сервера для внешних программ, вы можете использовать ChannelsLiveServerTestCase, который является заменой стандартного Django LiveServerTestCase:

from channels.testing import ChannelsLiveServerTestCase

class SomeLiveTests(ChannelsLiveServerTestCase):

    def test_live_stuff(self):
        call_external_testing_thing(self.live_server_url)

Примечание

Вы не можете использовать базу данных in-memory для ваших живых тестов. Поэтому включите в настройки имя файла тестовой базы данных, чтобы указать Django на использование файловой базы данных, если вы используете SQLite:

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.sqlite3",
        "NAME": os.path.join(BASE_DIR, "db.sqlite3"),
        "TEST": {
            "NAME": os.path.join(BASE_DIR, "db_test.sqlite3"),
        },
    },
}

serve_static

Подкласс ChannelsLiveServerTestCase с serve_static = True для обслуживания статических файлов (сравнимо с StaticLiveServerTestCase Django, вам не нужно запускать collectstatic до или как часть настройки тестов).

Безопасность
Рабочие и фоновые задачи
Вернуться на верх