Как в python тестировать асинхронный http-stream?

1

pip install aiohttp pytest-aiohttp

1
2
3

import pytest
from aiohttp import web
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

class TestHTTPStream(AioHTTPTestCase):
    async def setUpAsync(self):
        # Дополнительная инициализация перед запуском каждого тестового метода
        pass

    async def tearDownAsync(self):
        # Дополнительные действия после завершения каждого тестового метода
        pass

    async def get_application(self):
        # Возвращает экземпляр объекта веб-приложения aiohttp
        app = web.Application()
        app.router.add_get('/stream', self.stream_handler)
        return app

    @unittest_run_loop
    async def test_http_stream(self):
        # Асинхронный тестовый метод для проверки http-потока
        async with self.client.get('/stream') as response:
            assert response.status == 200
            # Дополнительные проверки, чтение данных из потока и т. д.

    async def stream_handler(self, request):
        # Обработчик потока с данными
        response = web.StreamResponse()
        await response.prepare(request)

        # Здесь вы можете отправлять данные в поток
        await response.write(b'Hello, World!')

        await response.write_eof()
        return response

1

pytest test_http_stream.py