如何确保结果是由异步而不是池给出的

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 返回值之前调用的方法（以及从应用，加入，所以如果你不确定第二个测试模块使用的多处理的确切方法，你可以坚持 pool.ApplyResult.get)。

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

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