如何确保结果是由异步而不是池提供的

Question

需要测试的函数（意味着我看不到代码，我只能导入它们）：

文件async_data.py

import asyncio
import socket
import aiohttp


async def get_json(client, uid):
    json_url = 'https://jsonplaceholder.typicode.com/todos/{uid}'.format(uid=uid)
    resp = await client.request('GET', json_url)
    data = await resp.json()
    return data


async def main_async(range_max):
    conn = aiohttp.TCPConnector(family=socket.AF_INET, verify_ssl=True)
    async with aiohttp.ClientSession(trust_env=True, connector=conn) as client:
        tasks = [get_json(client, x) for x in range(range_max)]
        data = await asyncio.gather(*tasks, return_exceptions=True)
        return data

第二个（同步模式或使用池中的相同任务）sync_data.py

import json
import urllib.request
from multiprocessing import Pool


def get_json_url(uid):
    json_url = 'https://jsonplaceholder.typicode.com/todos/{uid}'.format(uid=uid)
    jsondata = {}
    try:
        with urllib.request.urlopen(json_url) as url:
            jsondata = json.loads(url.read().decode())
    except urllib.error.HTTPError:
        pass
    return jsondata


def main_sync(range_max):
    return [get_json_url(uid) for uid in range(range_max)]


def main_pool(range_max):
    with Pool() as pool:
        result = pool.map(get_json_url, range(range_max))
    return result

主块，这里的函数main_async，main_sync，main_pool看起来像在黑盒子里，运行测试：

import time
import asyncio
from async_data import main_async
from sync_data import main_sync, main_pool

def main():
    total_cnt = 200
    # async block
    async_start = time.clock()
    loop = asyncio.get_event_loop()
    try:
        async_data = loop.run_until_complete(main_async(total_cnt))
    finally:
        loop.close()
    async_time = time.clock() - async_start
    # pool block
    pool_start = time.clock()
    pool_data = main_pool(total_cnt)
    pool_time = time.clock() - pool_start
    # sync block
    sync_start = time.clock()
    sync_data = main_sync(total_cnt)
    sync_time = time.clock() - sync_start
    # assert data
    sorted_async = sorted([x.get('id', -1) for x in async_data])
    sorted_pool = sorted([x.get('id', -1) for x in pool_data])
    sorted_sync = sorted([x.get('id', -1) for x in sync_data])
    assert sorted_async == sorted_pool
    assert sorted_async == sorted_sync
    assert sync_time > async_time
    assert sync_time > pool_time
    # AND here i want to be ensure that the result was given by async not pool

if __name__ == '__main__':
    main()

测试数据是否由async或sync方法接收的简单方法是检查执行时间。但是，如果代码使用pool或async，您可以测试哪种方式？

Answer 1

您可以尝试一些模拟测试：

import multiprocessing.pool
from unittest.mock import patch

...

with patch(
    'multiprocessing.pool.ApplyResult.get',
    autospec=True,
    wraps=multiprocessing.pool.ApplyResult.get
) as patched:
    async_start = time.clock()
    loop = asyncio.get_event_loop()
    try:
        async_data = loop.run_until_complete(main_async(total_cnt))
    finally:
        loop.close()
    async_time = time.clock() - async_start
    patched.assert_not_called()

    ...

    pool_start = time.clock()
    pool_data = main_pool(total_cnt)
    pool_time = time.clock() - pool_start
    patched.assert_called()

pool.ApplyResult.get是在从pool.map返回值之前调用的方法（以及apply，join，所以如果你不确定第二个测试模块使用多处理的确切方法，你可以坚持使用pool.ApplyResult 。得到）。

然后是unittest.mock.patch对象：它是用于测试的工具，它的目的是在标准库或第三方库中替换某些方法或对象。通常，它会阻止调用修补方法，只返回一些模仿原始方法工作的预定义值。

但您可以使用wraps参数以不同的方式使用。如果将原始方法传递给此参数，则将在进程中调用原始方法。仍然，pool.ApplyResult.get将包含修补的对象而不是原始的get方法。但是当修补的对象处理呼叫时，会调用原始的get。因此，您既可以获得该方法的结果，也可以获得unittest库提供的一些额外统计信息，例如assert_called。