PyTestによる非同期コードのテスト（翻訳）

コースの資料を準備するとき、私たちはあなたと共有したい興味深い記事に定期的に出くわします！



投稿者Stefan Scherfke 「テスト（非同期）コルーチンをpytestで」







PyTestはPythonでのテストに最適なパッケージであり、長い間私のお気に入りのパッケージの1つでした。 これにより、テストの記述が非常に容易になり、失敗したテストについて報告する機会が十分にあります。



ただし、バージョン2.7の時点では、（asyncio）ルーチンのテストの効果は低くなります。 したがって、次の方法でテストしないでください。

# tests/test_coros.py import asyncio def test_coro(): loop = asyncio.get_event_loop() @asyncio.coroutine def do_test(): yield from asyncio.sleep(0.1, loop=loop) assert 0 # onoes! loop.run_until_complete(do_test())
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

この方法には多くの欠点と過剰があります。 興味深い行は、yield fromステートメントとassertステートメントを含む行のみです。



各テストケースには、テストの成功に関係なく正しく完了する独自のイベントループが必要です。



上記のように歩留まりを適用することは不可能です。pytestは、テストが新しいテストケースを返すと判断します。



したがって、実際のテストを含む別のサブルーチンを作成する必要があり、それを実行するためにイベントループが起動されます。

このようにすれば、テストはよりきれいになります。

 # tests/test_coros.py @asyncio.coroutine def test_coro(loop): yield from asyncio.sleep(0.1, loop=loop) assert 0
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

Pytestには柔軟なプラグインシステムがあり、実装が可能です。 しかし、残念なことに、必要なフックのほとんどは文書化されていないか、まったく文書化されていないため、それらを実行する方法を見つけることには問題があります。



ローカルのポップアッププラグインは、「実際の」外部プラグインを作成するよりも少し簡単だからこそ作成されます。 Pytestは、各テストディレクトリでconftest.pyというファイルを見つけ、そのディレクトリに実装されているフィクスチャとフックをこのディレクトリのすべてのテストに適用します。



各テストケースの新しいイベントループを作成し、テストの終了時に正しく閉じるフィクスチャを作成することから始めましょう。

 # tests/conftest.py import asyncio import pytest @pytest.yield_fixture def loop(): #  loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) yield loop #  loop.close() # tests/test_coros.py def test_coro(loop): @asyncio.coroutine def do_test(): yield from asyncio.sleep(0.1, loop=loop) assert 0 # onoes! loop.run_until_complete(do_test())
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

各テストの前に、pytestは最初のyieldステートメントまでループフィクスチャを実行します。 yieldが返すものは、テストケースのループ引数（つまり、ループ）として渡されます。 テストが完了すると（成功したかどうかに関係なく）、pytestはループフィクスチャを完了し、正しく閉じます。 同様に、各テスト後にソケットを作成して閉じるテストフィクスチャを記述することができます（ソケットフィクスチャは、この例と同じようにループフィクスチャに依存する可能性があります。クールですね。）



しかし、終わりはまだ遠い。 pytestにテストルーチンの実行方法を教える必要があります。 これを行うには、asyncioルーチンのアセンブル方法（テストジェネレーターとしてではなく、通常のテスト関数としてビルドする必要があります）およびそれらの実行方法（loop.run_until_complete（）を使用）を変更する必要があります。

 # tests/conftest.py def pytest_pycollect_makeitem(collector, name, obj): """ asyncio    ,   .” if collector.funcnamefilter(name) and asyncio.iscoroutinefunction(obj): #       #    ,         # ,    "pytest.mark.parametrize()". return list(collector._genfunctions(name, obj)) # else: #   None,    pytest'   “obj” def pytest_pyfunc_call(pyfuncitem): """ ``pyfuncitem.obj`` -  asyncio ,         """ testfunction = pyfuncitem.obj if not asyncio.iscoroutinefunction(testfunction): #  None,    . Pytest   #      return #       : funcargs = pyfuncitem.funcargs #     argnames = pyfuncitem._fixtureinfo.argnames #     testargs = {arg: funcargs[arg] for arg in argnames} #  -    (   !) coro = testfunction(**testargs) #         loop = testargs['loop'] if loop in testargs else asyncio.get_event_loop() loop.run_until_complete(coro) return True #  pytest',    
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

このプラグインは、pytestバージョン2.4以降で動作します。 バージョン2.6および2.7でパフォーマンスをテストしました。



そして、すべてがうまくいきますが、Stack Overflowでこのソリューションが公開されてすぐに、PyTest-Asyncioプラグインが登場しましたが、Stefanはまったく動揺していませんでしたが、このプラグインの詳細な分析を行いました

高度な非同期コードテスト

最初の記事で、pytestが品質テストの作成にどのように役立つかを示しました。 フィクスチャを使用すると、テストケースごとにクリーンなイベントループを作成できます。プラグインシステムのおかげで、実際には非同期のコルーチンであるテスト関数を作成できます。



しかし、この資料の作業が進行中に、Tin Tvrtkowitzはpytest-asyncioプラグインを作成しました。



要するに、これを行うことができます：

 import asyncio import time import pytest @pytest.mark.asyncio async def test_coro(event_loop): before = time.monotonic() await asyncio.sleep(0.1, loop=event_loop) after = time.monotonic() assert after - before >= 0.1
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

代わりに：

 import asyncio import time def test_coro(): loop = asyncio.new_event_loop() try: asyncio.set_event_loop(loop) before = time.monotonic() loop.run_until_complete(asyncio.sleep(0.1, loop=loop)) after = time.monotonic() assert after - before >= 0.1 finally: loop.close()
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

pytest-asyncioを使用すると、テストが明らかに改善されます（これはプラグインの機能の制限ではありません！）。



私がアイオマに取り組んだとき、私は満たすのが簡単ではない追加の要件に遭遇しました。



aiomas自体について少し説明します。 asyncioトランスポートに3層の抽象化を追加します。

• チャネルレイヤーを使用すると、「要求/応答」スキームに従ってJSONまたはMsgPack形式でメッセージを送信できます。 さまざまなタイプのトランスポートで動作するカスタムプロトコルを使用します：TCPソケット、Unixドメインソケット、カスタムトランスポート-ローカルキュー。
• RPC層は、チャネル層上にリモートプロシージャコールシステムを作成します。
• エージェント層（マルチエージェントシステム用）には、さらにネットワーク関連の詳細が含まれており、ネットワーク接続を介して他のクラスのメソッドを呼び出すクラスを作成できます。

チャネルレイヤーの動作の最も簡単な例：

 import aiomas async def handle_client(channel): """ """ req = await channel.recv() print(req.content) await req.reply('cya') await channel.close() async def client(): """ :      """ channel = await aiomas.channel.open_connection(('localhost', 5555)) rep = await channel.send('ohai') print(rep) await channel.close() server = aiomas.run(aiomas.channel.start_server( ('localhost', 5555), handle_client)) aiomas.run(client()) server.close() aiomas.run(server.wait_closed())
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

テスト要件

各テストにきれいなイベントループが必要です。



これは、pytest-asyncioにあるevent_loopフィクスチャを使用して実行できます。

各テストは、可能なすべてのトランスポート（TCPソケット、Unixドメインソケットなど）で実行する必要があります。



理論的には、これはpytest.mark.parametrize（）デコレータを使用して解決できます（ただし、今回のケースでは、これが後で明らかになる方法はありません）。



各テストにはクライアントコルーチンが必要です。 理想的には、テスト自体。



pytest-asyncioのpytest.mark.asyncioデコレーターがこのタスクを処理します。



各テストには、クライアント接続用のカスタムコールバックを備えたサーバーが必要です。 テストの最後に、テスト結果に関係なくサーバーの電源を切る必要があります。



コルーチンはこの問題を解決できるようですが、各サーバーはクライアント接続を管理するために特定のコールバックを必要とします。 これは問題の解決を複雑にします。 テストの1つが失敗しても、「アドレスは既に使用されています」というエラーが表示されるとは思わない pytest-asyncioの治具used_tcp_portは、急いで助けになります。



loop.run_until_complete（）を常に使用したくありません。



そして、pytest.mark.asyncioデコレーターは問題を解決します。



解決すべき課題をまとめましょう。各テストには2つのフィクスチャ（1つはイベントループ用、もう1つはアドレスタイプ用）が必要ですが、それらを1つにまとめたいと思います。 サーバーを構成するにはフィクスチャを作成する必要がありますが、その方法は？

最初のアプローチ

ループとアドレスタイプをフィクスチャでラップできます。 ctx（テストコンテキストの略）と呼びましょう。 フィクスチャパラメータのおかげで、アドレスのタイプごとに個別のアドレスを簡単に作成できます。

 import tempfile import py import pytest class Context: def __init__(self, loop, addr): self.loop = loop self.addr = addr @pytest.fixture(params=['tcp', 'unix']) def ctx(request, event_loop, unused_tcp_port, short_tmpdir): """   TCP     Unix.""" addr_type = request.param if addr_type == 'tcp': addr = ('127.0.0.1', unused_tcp_port) elif addr_type == 'unix': addr = short_tmpdir.join('sock').strpath else: raise RuntimeError('Unknown addr type: %s' % addr_type) ctx = Context(event_loop, addr) return ctx @pytest.yield_fixture() def short_tmpdir(): """      Unix. ,   pytest' tmpdir,     """ with tempfile.TemporaryDirectory() as tdir: yield py.path.local(tdir)
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

これにより、次のようなテストを作成できます。

 import aiomas @pytest.mark.asyncio async def test_channel(ctx): results = [] async def handle_client(channel): req = await channel.recv() results.append(req.content) await req.reply('cya') await channel.close() server = await aiomas.channel.start_server(ctx.addr, handle_client) try: channel = await aiomas.channel.open_connection(ctx.addr) rep = await channel.send('ohai') results.append(rep) await channel.close() finally: server.close() await server.wait_closed() assert results == ['ohai', 'cya']
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

すでにかなりうまく機能しており、ctxフィクスチャを使用するすべてのテストは、アドレスのタイプごとに1回実行されます。



ただし、2つの問題が残っています。

1. Fixtureは、未使用のTCPポート+一時ディレクトリを常に必要とします-必要なのは2つのうち1つだけです。
2. サーバーのセットアップ（およびサーバーのクローズ）には、すべてのテストで同一になる一定量のコードが含まれているため、フィクスチャーに含める必要があります。 ただし、各サーバーにはテスト依存のコールバックが必要であるため、直接機能しません（これは、サーバーを作成する行で確認できますserver = await ...）。 しかし、サーバーの備品がなければ、彼の解体はありません...

これらの問題を解決する方法を見てみましょう。

第二のアプローチ

最初の問題は、フィクスチャが受け取るリクエストオブジェクトに属するgetfuncargvalue（）メソッドを使用して解決されます。 このメソッドを使用すると、その関数を手動で呼び出すことができます。

 @pytest.fixture(params=['tcp', 'unix']) def ctx(request, event_loop): """   TCP     Unix.""" addr_type = request.param if addr_type == 'tcp': port = request.getfuncargvalue('unused_tcp_port') addr = ('127.0.0.1', port) elif addr_type == 'unix': tmpdir = request.getfuncargvalue('short_tmpdir') addr = tmpdir.join('sock').strpath else: raise RuntimeError('Unknown addr type: %s' % addr_type) ctx = Context(event_loop, addr) return ctx
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

2番目の問題を解決するために、各テストに渡されるContextクラスを拡張できます。 テストから直接呼び出すことができるContext.start_server（client_handler）メソッドを追加します。 また、ctxフィクスチャに最終的な分解を追加します。これにより、終了後にサーバーが閉じられます。 とりわけ、ショートカット用のいくつかの関数を作成できます。

 import asyncio import tempfile import py import pytest class Context: def __init__(self, loop, addr): self.loop = loop self.addr = addr self.server = None async def connect(self, **kwargs): """    "self.addr".""" return (await aiomas.channel.open_connection( self.addr, loop=self.loop, **kwargs)) async def start_server(self, handle_client, **kwargs): """    *handle_client*,  "self.addr".""" self.server = await aiomas.channel.start_server( self.addr, handle_client, loop=self.loop, **kwargs) async def start_server_and_connect(self, handle_client, server_kwargs=None, client_kwargs=None): """ :: await ctx.start_server(...) channel = await ctx.connect()" """ if server_kwargs is None: server_kwargs = {} if client_kwargs is None: client_kwargs = {} await self.start_server(handle_client, **server_kwargs) return (await self.connect(**client_kwargs)) async def close_server(self): """ .""" if self.server is not None: server, self.server = self.server, None server.close() await server.wait_closed() @pytest.yield_fixture(params=['tcp', 'unix']) def ctx(request, event_loop): """   TCP     Unix.""" addr_type = request.param if addr_type == 'tcp': port = request.getfuncargvalue('unused_tcp_port') addr = ('127.0.0.1', port) elif addr_type == 'unix': tmpdir = request.getfuncargvalue('short_tmpdir') addr = tmpdir.join('sock').strpath else: raise RuntimeError('Unknown addr type: %s' % addr_type) ctx = Context(event_loop, addr) yield ctx #        : aiomas.run(ctx.close_server()) aiomas.run(asyncio.gather(*asyncio.Task.all_tasks(event_loop), return_exceptions=True))
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

このような追加機能により、テストケースは大幅に短くなり、読みやすくなり、信頼性が向上します。

 import aiomas @pytest.mark.asyncio async def test_channel(ctx): results = [] async def handle_client(channel): req = await channel.recv() results.append(req.content) await req.reply('cya') await channel.close() channel = await ctx.start_server_and_connect(handle_client) rep = await channel.send('ohai') results.append(rep) await channel.close() assert results == ['ohai', 'cya']
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ctxフィクスチャ（およびContextクラス）は、もちろん、これまでに書いた中で最短ではありませんが、約200行の定型コードのテストを取り除くのに役立ちました。

終わり

エレガントなソリューションと信頼できるコード！