Was ist der beste Weg, um Unit-Tests für Code mithilfe der Python 3.4- asyncio
Bibliothek zu schreiben ? Angenommen, ich möchte einen TCP-Client testen ( SocketConnection
):
import asyncio
import unittest
class TestSocketConnection(unittest.TestCase):
def setUp(self):
self.mock_server = MockServer("localhost", 1337)
self.socket_connection = SocketConnection("localhost", 1337)
@asyncio.coroutine
def test_sends_handshake_after_connect(self):
yield from self.socket_connection.connect()
self.assertTrue(self.mock_server.received_handshake())
Wenn Sie diesen Testfall mit dem Standardtestläufer ausführen, ist der Test immer erfolgreich, da die Methode nur bis zur ersten yield from
Anweisung ausgeführt wird. Danach wird sie zurückgegeben, bevor Zusicherungen ausgeführt werden. Dies führt dazu, dass Tests immer erfolgreich sind.
Gibt es einen vorgefertigten Testläufer, der so asynchronen Code verarbeiten kann?
python
unit-testing
python-3.x
python-unittest
python-asyncio
Marvin töten
quelle
quelle
loop.run_until_complete()
anstelle von verwendenyield from
. Siehe auchasyncio.test_utils
.async def
und zurawait
Syntax finden Sie unter: stackoverflow.com/questions/41263988/…Antworten:
Ich habe das Problem vorübergehend mit einem von Tornados gen_test inspirierten Dekorateur gelöst :
def async_test(f): def wrapper(*args, **kwargs): coro = asyncio.coroutine(f) future = coro(*args, **kwargs) loop = asyncio.get_event_loop() loop.run_until_complete(future) return wrapper
Wie von JF Sebastian vorgeschlagen, wird dieser Dekorateur blockieren, bis die Coroutine der Testmethode abgeschlossen ist. Dadurch kann ich Testfälle wie folgt schreiben:
class TestSocketConnection(unittest.TestCase): def setUp(self): self.mock_server = MockServer("localhost", 1337) self.socket_connection = SocketConnection("localhost", 1337) @async_test def test_sends_handshake_after_connect(self): yield from self.socket_connection.connect() self.assertTrue(self.mock_server.received_handshake())
Bei dieser Lösung fehlen wahrscheinlich einige Randfälle.
Ich denke , eine Einrichtung wie diese Python-Standardbibliothek hinzugefügt sollte zu machen
asyncio
undunittest
Interaktion bequemer aus dem Kasten heraus .quelle
asyncio.get_event_loop()
und benutzteasyncio.new_event_loop()
asyncio.coroutine
die veraltet ist und in py3.10 entfernt wird: docs.python.org/3/library/…async_test
, vorgeschlagen von Marvin Killing, kann definitiv helfen - und auch direkt anrufenloop.run_until_complete()
Ich empfehle jedoch auch dringend, für jeden Test eine neue Ereignisschleife neu zu erstellen und die Schleife direkt an API-Aufrufe zu übergeben (zumindest
asyncio
akzeptiert sie selbstloop
nur Schlüsselwortparameter für jeden Aufruf, der sie benötigt).Mögen
class Test(unittest.TestCase): def setUp(self): self.loop = asyncio.new_event_loop() asyncio.set_event_loop(None) def test_xxx(self): @asyncio.coroutine def go(): reader, writer = yield from asyncio.open_connection( '127.0.0.1', 8888, loop=self.loop) yield from asyncio.sleep(0.01, loop=self.loop) self.loop.run_until_complete(go())
Dies isoliert Tests im Testfall und verhindert seltsame Fehler wie langjährige Coroutine, die in erstellt,
test_a
aber erst zurtest_b
Ausführungszeit beendet wurde.quelle
asyncio.set_event_loop(None)
und späterself.loop
explizit an übergeben,asyncio.open_connection()
anstatt es vonasyncio.set_event_loop(self.loop)
Anfang an richtig zu machen?asyncio.set_event_loop(None)
, spezifiziere ich direkt die Tatsache, dass die Bibliothek nicht auf die Existenz einer globalen Schleife weitergeleitet werden soll und sicher durch explizite Schleifenübergabe arbeiten soll. Es ist der Codestil für Asyncio-Tests selbst, ich verwende ihn auch in meinen Bibliotheken.asyncio.open_connection
nicht wahr? Laufen es produziertConnectionRefusedError: [Errno 61] Connect call failed ('127.0.0.1', 8888)
setUp
Methode auf der Adresse einrichten kann. Die konkrete Umsetzung hängt von Ihren Anforderungen ab.Da Python 3.8 unittest mit der für diesen Zweck entwickelten IsolatedAsyncioTestCase- Funktion geliefert wird .
from unittest import IsolatedAsyncioTestCase class Test(IsolatedAsyncioTestCase): async def test_functionality(self): result = await functionality() self.assertEqual(expected, result)
quelle
pytest-asyncio sieht vielversprechend aus:
@pytest.mark.asyncio async def test_some_asyncio_code(): res = await library.do_something() assert b'expected result' == res
quelle
unittest.TestCase
, das für mich sehr begrenzt ist. jacobbridges.github.io/post/unit-testing-with-asyncioGenau wie der
async_test
in https://stackoverflow.com/a/23036785/350195 erwähnte Wrapper ist hier eine aktualisierte Version für Python 3.5+def async_test(coro): def wrapper(*args, **kwargs): loop = asyncio.new_event_loop() return loop.run_until_complete(coro(*args, **kwargs)) return wrapper class TestSocketConnection(unittest.TestCase): def setUp(self): self.mock_server = MockServer("localhost", 1337) self.socket_connection = SocketConnection("localhost", 1337) @async_test async def test_sends_handshake_after_connect(self): await self.socket_connection.connect() self.assertTrue(self.mock_server.received_handshake())
quelle
nosetests
, möchten Sie möglicherweise den Dekorateur umbenennen, oder die Nase denkt, dass es sich tatsächlich auch um einen Test handelt, mit einer mysteriösen Meldung über dasasync_test
Fehlen eines erforderlichen Positionsarguments. I umbenanntasynctest
und eine zusätzliche Dekorateur hinzugefügt@nose.tools.istest
den Testfall autodiscoverable zu machenVerwenden Sie diese Klasse anstelle der
unittest.TestCase
Basisklasse:import asyncio import unittest class AioTestCase(unittest.TestCase): # noinspection PyPep8Naming def __init__(self, methodName='runTest', loop=None): self.loop = loop or asyncio.get_event_loop() self._function_cache = {} super(AioTestCase, self).__init__(methodName=methodName) def coroutine_function_decorator(self, func): def wrapper(*args, **kw): return self.loop.run_until_complete(func(*args, **kw)) return wrapper def __getattribute__(self, item): attr = object.__getattribute__(self, item) if asyncio.iscoroutinefunction(attr): if item not in self._function_cache: self._function_cache[item] = self.coroutine_function_decorator(attr) return self._function_cache[item] return attr class TestMyCase(AioTestCase): async def test_dispatch(self): self.assertEqual(1, 1)
EDIT 1:
Bitte beachten Sie die @ Nitay- Antwort zu verschachtelten Tests.
quelle
Sie können auch einen
aiounittest
ähnlichen Ansatz verwenden wie @Andrew Svetlov, @Marvin Killing antwortet und ihn in eine benutzerfreundlicheAsyncTestCase
Klasse einwickelt :import asyncio import aiounittest async def add(x, y): await asyncio.sleep(0.1) return x + y class MyTest(aiounittest.AsyncTestCase): async def test_async_add(self): ret = await add(5, 6) self.assertEqual(ret, 11) # or 3.4 way @asyncio.coroutine def test_sleep(self): ret = yield from add(5, 6) self.assertEqual(ret, 11) # some regular test code def test_something(self): self.assertTrue(true)
Wie Sie sehen, wird der asynchrone Fall von behandelt
AsyncTestCase
. Es unterstützt auch synchronen Test. Es besteht die Möglichkeit, eine benutzerdefinierte Ereignisschleife bereitzustellen, die einfach überschrieben wirdAsyncTestCase.get_event_loop
.Wenn Sie (aus irgendeinem Grund) die andere TestCase-Klasse bevorzugen (z. B.
unittest.TestCase
), können Sieasync_test
Decorator verwenden:import asyncio import unittest from aiounittest import async_test async def add(x, y): await asyncio.sleep(0.1) return x + y class MyTest(unittest.TestCase): @async_test async def test_async_add(self): ret = await add(5, 6) self.assertEqual(ret, 11)
quelle
Normalerweise definiere ich meine asynchronen Tests als Coroutinen und verwende einen Dekorator, um sie zu "synchronisieren":
import asyncio import unittest def sync(coro): def wrapper(*args, **kwargs): loop = asyncio.get_event_loop() loop.run_until_complete(coro(*args, **kwargs)) return wrapper class TestSocketConnection(unittest.TestCase): def setUp(self): self.mock_server = MockServer("localhost", 1337) self.socket_connection = SocketConnection("localhost", 1337) @sync async def test_sends_handshake_after_connect(self): await self.socket_connection.connect() self.assertTrue(self.mock_server.received_handshake())
quelle
Die Pylover-Antwort ist korrekt und sollte zu unittest IMO hinzugefügt werden.
Ich würde eine kleine Änderung hinzufügen, um verschachtelte asynchrone Tests zu unterstützen:
class TestCaseBase(unittest.TestCase): # noinspection PyPep8Naming def __init__(self, methodName='runTest', loop=None): self.loop = loop or asyncio.get_event_loop() self._function_cache = {} super(BasicRequests, self).__init__(methodName=methodName) def coroutine_function_decorator(self, func): def wrapper(*args, **kw): # Is the io loop is already running? (i.e. nested async tests) if self.loop.is_running(): t = func(*args, **kw) else: # Nope, we are the first t = self.loop.run_until_complete(func(*args, **kw)) return t return wrapper def __getattribute__(self, item): attr = object.__getattribute__(self, item) if asyncio.iscoroutinefunction(attr): if item not in self._function_cache: self._function_cache[item] = self.coroutine_function_decorator(attr) return self._function_cache[item] return attr
quelle
Wenn Sie neben der Antwort von pylover eine andere asynchrone Methode aus der Testklasse selbst verwenden möchten, funktioniert die folgende Implementierung besser:
import asyncio import unittest class AioTestCase(unittest.TestCase): # noinspection PyPep8Naming def __init__(self, methodName='runTest', loop=None): self.loop = loop or asyncio.get_event_loop() self._function_cache = {} super(AioTestCase, self).__init__(methodName=methodName) def coroutine_function_decorator(self, func): def wrapper(*args, **kw): return self.loop.run_until_complete(func(*args, **kw)) return wrapper def __getattribute__(self, item): attr = object.__getattribute__(self, item) if asyncio.iscoroutinefunction(attr) and item.startswith('test_'): if item not in self._function_cache: self._function_cache[item] = self.coroutine_function_decorator(attr) return self._function_cache[item] return attr class TestMyCase(AioTestCase): async def multiplier(self, n): await asyncio.sleep(1) # just to show the difference return n*2 async def test_dispatch(self): m = await self.multiplier(2) self.assertEqual(m, 4)
Die einzige Änderung war -
and item.startswith('test_')
in der__getattribute__
Methode.quelle