如何将 `url_route` 键/值添加到 `scope` 进行测试？

Question

我正在测试我的消费者，它使用scope['url_route']，但使用HttpCommunicator 或ApplicationCommunicator，该参数未设置。如何设置此参数？测试文档非常有限，没有关于此https://channels.readthedocs.io/en/latest/topics/testing.html 的任何文档。

test.py

from channels.testing import HttpCommunicator
import pytest

from my_app import consumers

@pytest.mark.asyncio
async def test_consumers():
    communicator = HttpCommunicator(consumers.MyConsumer, 'GET', '/project/1')
    response = await communicator.get_response()
    assert response['status'] == 400

my_app.py

import json

from channels.db import database_sync_to_async
from channels.generic.http import AsyncHttpConsumer

from projects import model

class MyConsumer(AsyncHttpConsumer):
    async def handle(self, body):
        _id = int(self.scope['url_route']['kwargs'].get('project_id', 1))
        record = await database_sync_to_async(models.Project.objects.get)(id=_id)
        data = json.dumps({"id": _id})
        await self.send_response(200, data.encode('utf8'))

Answer 1

在文档中找到了解决方案，但它位于 Websocket 测试部分，而不是 HTTP 部分。导入您的 asgi 应用程序或将使用者包装在 URLRouter 中，然后在 HttpCommunicator 中使用它。原来有一些流在URLRouter 中添加了scope['url_route']。

Answer 2

您还可以更改 channels.testing 提供的 WebsocketCommunicator 以包含您需要的额外信息。我无法获得 URLRouter 选项以将正确的项目添加到我的实例中的范围。

例如，我的 会话 中需要一些对象，所以我这样做是为了适应：

import json
from urllib.parse import unquote, urlparse

from asgiref.testing import ApplicationCommunicator


class WebsocketCommunicator(ApplicationCommunicator):
    """
    ApplicationCommunicator subclass that has WebSocket shortcut methods.

    It will construct the scope for you, so you need to pass the application
    (uninstantiated) along with the initial connection parameters.
    """

    def __init__(self, application, path, headers=None, subprotocols=None, property=None, client=None):
        if not isinstance(path, str):
            raise TypeError("Expected str, got {}".format(type(path)))
        parsed = urlparse(path)
        self.scope = {
            "type": "websocket",
            "path": unquote(parsed.path),
            "query_string": parsed.query.encode("utf-8"),
            "headers": headers or [],
            "subprotocols": subprotocols or [],
            # needed for connect() in consumer
            "session": {
                # create client and property with passed objects
                'client': client,
                'property': property
            },
        }
        super().__init__(application, self.scope)

    async def connect(self, timeout=1):
        """
        Trigger the connection code.

        On an accepted connection, returns (True, <chosen-subprotocol>)
        On a rejected connection, returns (False, <close-code>)
        """
        await self.send_input({"type": "websocket.connect"})
        response = await self.receive_output(timeout)
        if response["type"] == "websocket.close":
            return (False, response.get("code", 1000))
        else:
            return (True, response.get("subprotocol", None))

    async def send_to(self, text_data=None, bytes_data=None):
        """
        Sends a WebSocket frame to the application.
        """
        # Make sure we have exactly one of the arguments
        assert bool(text_data) != bool(
            bytes_data
        ), "You must supply exactly one of text_data or bytes_data"
        # Send the right kind of event
        if text_data:
            assert isinstance(text_data, str), "The text_data argument must be a str"
            await self.send_input({"type": "websocket.receive", "text": text_data})
        else:
            assert isinstance(
                bytes_data, bytes
            ), "The bytes_data argument must be bytes"
            await self.send_input({"type": "websocket.receive", "bytes": bytes_data})

    async def send_json_to(self, data):
        """
        Sends JSON data as a text frame
        """
        await self.send_to(text_data=json.dumps(data))

    async def receive_from(self, timeout=1):
        """
        Receives a data frame from the view. Will fail if the connection
        closes instead. Returns either a bytestring or a unicode string
        depending on what sort of frame you got.
        """
        response = await self.receive_output(timeout)
        # Make sure this is a send message
        assert response["type"] == "websocket.send"
        # Make sure there's exactly one key in the response
        assert ("text" in response) != (
            "bytes" in response
        ), "The response needs exactly one of 'text' or 'bytes'"
        # Pull out the right key and typecheck it for our users
        if "text" in response:
            assert isinstance(response["text"], str), "Text frame payload is not str"
            return response["text"]
        else:
            assert isinstance(
                response["bytes"], bytes
            ), "Binary frame payload is not bytes"
            return response["bytes"]

    async def receive_json_from(self, timeout=1):
        """
        Receives a JSON text frame payload and decodes it
        """
        payload = await self.receive_from(timeout)
        assert isinstance(payload, str), "JSON data is not a text frame"
        return json.loads(payload)

    async def disconnect(self, code=1000, timeout=1):
        """
        Closes the socket
        """
        await self.send_input({"type": "websocket.disconnect", "code": code})
        await self.wait(timeout)

然后，例如，您可以编写一个轻松连接的测试：

import pytest
from channels.routing import URLRouter
from django.conf import settings
from django.conf.urls import url

from apps.chat.consumers import PropertyConsumer
# this is the edited communicator
from utils.testing.WebsocketCommunicator import WebsocketCommunicator

@pytest.mark.asyncio
class TestWebsockets:
    async def test_can_connect(self, client, slug, property, admin):
        # Use in-memory channel layers for testing.
        settings.CHANNEL_LAYERS = {
            'default': {
                'BACKEND': 'channels.layers.InMemoryChannelLayer',
            },
        }
    
        application = URLRouter([
            url(r'(?P<client_url>\w+)/chat/property/(?P<property_id>\w+)/ws/$', PropertyConsumer),
        ])
        
        communicator = WebsocketCommunicator(application, f'{slug.url_base}/chat/property/{property.id}/ws/', property=property, client=slug)
        connected, subprotocol = await communicator.connect()
        assert connected
        await communicator.disconnect()