Ссылаясь на: Как мне избежать аргумента цикла
Я пытаюсь написать клиент SockJS на python, и я хотел бы, чтобы этот код имел несколько автоматических тестов.
Вот код:
from urllib.parse import urlparse
import websockets
class SockJsClient:
def __init__(self, base_url):
self.base_url = urlparse(base_url)
async def connect(self):
uri = self.base_url.geturl() + "/websocket"
self.websocket = await websockets.connect(uri)
async def disconnect(self):
await self.websocket.close()
и вот мои автоматизированные тесты:
# -*- coding: utf-8 -*-
from .context import sockjs_client
import unittest
import logging
import asyncio
import threading
from aiohttp import web
import sockjs
import time
connected = False
def aiohttp_server():
async def handler(msg, session):
global connected
if msg.type == sockjs.MSG_OPEN:
connected = True
if msg.type == sockjs.MSG_CLOSE:
connected = False
app = web.Application()
sockjs.add_endpoint(app, handler)
return web.AppRunner(app)
async def run_server(ready):
logging.basicConfig(level=logging.DEBUG,
format='%(asctime)s %(levelname)s %(message)s')
runner = aiohttp_server()
await runner.setup()
site = web.TCPSite(runner, 'localhost', 8080)
await site.start()
ready.set()
# emulates loop.run_forever()
await asyncio.get_running_loop().create_future()
def start_server():
ready = threading.Event()
threading.Thread(target=asyncio.run, args=(run_server(ready),),
daemon=True).start()
ready.wait()
class SockJsClientTestSuite(unittest.TestCase):
"""Advanced test cases."""
@classmethod
def setUpClass(cls):
logger = logging.getLogger('websockets')
logger.setLevel(logging.INFO)
logger.addHandler(logging.StreamHandler())
start_server()
def test(self):
client = sockjs_client.SockJsClient("ws://localhost:8080/sockjs")
assert "ws" == client.base_url.scheme
assert "localhost" == client.base_url.hostname
assert 8080 == client.base_url.port
assert "/sockjs" == client.base_url.path
async def _connect(self):
client = sockjs_client.SockJsClient("ws://localhost:8080/sockjs")
await client.connect()
assert connected
await client.disconnect()
def test_connect(self):
asyncio.new_event_loop().run_until_complete(self._connect())
async def _disconnect(self):
client = sockjs_client.SockJsClient("ws://localhost:8080/sockjs")
await client.connect()
await client.disconnect()
assert not connected
def test_disconnect(self):
asyncio.new_event_loop().run_until_complete(self._disconnect())
if __name__ == '__main__':
unittest.main()
В большинстве случаев тесты проходят:
python3.7 -m unittest
.2019-10-23 11:56:08,279 DEBUG Using selector: EpollSelector
2019-10-23 11:56:08,280 DEBUG open session: 136606092
2019-10-23 11:56:08,281 INFO close session: 136606092
2019-10-23 11:56:08,281 INFO session closed: 136606092
2019-10-23 11:56:08,281 INFO ::1 [23/Oct/2019:14:56:08 +0000] "GET /sockjs/websocket HTTP/1.1" 101 0 "-" "Python/3.7 websockets/8.0.2"
.2019-10-23 11:56:08,282 DEBUG Using selector: EpollSelector
2019-10-23 11:56:08,283 DEBUG open session: 794699575
2019-10-23 11:56:08,283 INFO close session: 794699575
2019-10-23 11:56:08,283 INFO session closed: 794699575
2019-10-23 11:56:08,284 INFO ::1 [23/Oct/2019:14:56:08 +0000] "GET /sockjs/websocket HTTP/1.1" 101 0 "-" "Python/3.7 websockets/8.0.2"
.
----------------------------------------------------------------------
Ran 3 tests in 0.010s
OK
но иногда они терпят неудачу:
python3.7 -m unittest
.2019-10-23 11:46:44,011 DEBUG Using selector: EpollSelector
2019-10-23 11:46:44,013 DEBUG open session: 1793450030
F2019-10-23 11:46:44,014 DEBUG Using selector: EpollSelector
2019-10-23 11:46:44,015 DEBUG open session: 1673843711
2019-10-23 11:46:44,015 INFO close session: 1673843711
2019-10-23 11:46:44,016 INFO session closed: 1673843711
2019-10-23 11:46:44,016 INFO ::1 [23/Oct/2019:14:46:44 +0000] "GET /sockjs/websocket HTTP/1.1" 101 0 "-" "Python/3.7 websockets/8.0.2"
.
======================================================================
FAIL: test_connect (tests.test_sockjs_client.SockJsClientTestSuite)
----------------------------------------------------------------------
Traceback (most recent call last):
File "/home/peter/ownCloud/NavBlue/eclipse-workspace-acars-mock/sockjs_client/tests/test_sockjs_client.py", line 73, in test_connect
asyncio.new_event_loop().run_until_complete(self._connect())
File "/usr/local/lib/python3.7/asyncio/base_events.py", line 579, in run_until_complete
return future.result()
File "/home/peter/ownCloud/NavBlue/eclipse-workspace-acars-mock/sockjs_client/tests/test_sockjs_client.py", line 69, in _connect
assert connected
AssertionError
----------------------------------------------------------------------
Ran 3 tests in 0.010s
FAILED (failures=1)
2019-10-23 11:46:44,022 ERROR Task was destroyed but it is pending!
task: <Task pending coro=<WebSocketCommonProtocol.transfer_data() running at /home/peter/.local/lib/python3.7/site-packages/websockets/protocol.py:795> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7fb0900d0090>()]> cb=[<TaskWakeupMethWrapper object at 0x7fb0900d00d0>()]>
2019-10-23 11:46:44,022 ERROR Task was destroyed but it is pending!
task: <Task pending coro=<WebSocketCommonProtocol.keepalive_ping() running at /home/peter/.local/lib/python3.7/site-packages/websockets/protocol.py:1084> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7fb0900c3b10>()]>>
2019-10-23 11:46:44,022 ERROR Task was destroyed but it is pending!
task: <Task pending coro=<WebSocketCommonProtocol.close_connection() running at /home/peter/.local/lib/python3.7/site-packages/websockets/protocol.py:1129> wait_for=<Task pending coro=<WebSocketCommonProtocol.transfer_data() running at /home/peter/.local/lib/python3.7/site-packages/websockets/protocol.py:795> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7fb0900d0090>()]> cb=[<TaskWakeupMethWrapper object at 0x7fb0900d00d0>()]>>
Exception ignored in: <coroutine object WebSocketCommonProtocol.close_connection at 0x7fb091311dd0>
Traceback (most recent call last):
File "/home/peter/.local/lib/python3.7/site-packages/websockets/protocol.py", line 1169, in close_connection
self.writer.close()
File "/usr/local/lib/python3.7/asyncio/streams.py", line 317, in close
return self._transport.close()
File "/usr/local/lib/python3.7/asyncio/selector_events.py", line 653, in close
self._loop.call_soon(self._call_connection_lost, None)
File "/usr/local/lib/python3.7/asyncio/base_events.py", line 683, in call_soon
self._check_closed()
File "/usr/local/lib/python3.7/asyncio/base_events.py", line 475, in _check_closed
raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed
make: *** [test] Error 1