Как обеспечить, чтобы результат был предоставлен асинхронно, а не пулом - PullRequest
Новая
       21

Как обеспечить, чтобы результат был предоставлен асинхронно, а не пулом

0 голосов
/ 02 апреля 2019

функций необходимо протестировать (значит, я не вижу код, я могу только импортировать их):

файл 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?

1 Ответ

2 голосов
/ 02 апреля 2019

Вы можете попробовать издеваться над своими тестами: 

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), поэтому, если вы не уверены, какой именно метод из многопроцессорной обработки использует второй протестированный модуль, вы можете придерживаться пула.ApplyResult.get).

Затем объект unittest.mock.patch: это инструмент, используемый при тестировании, и его цель - заменить какой-либо метод или объект либо в стандартной библиотеке, либо в сторонних библиотеках. Обычно он предотвращает вызов исправленного метода и просто возвращает какое-то предопределенное значение, имитирующее работу исходного метода.

Но вы можете использовать по-другому, с параметром wraps. Если вы передадите исходный метод этому параметру, исходный метод будет вызван в процессе. Тем не менее, pool.ApplyResult.get будет содержать исправленный объект вместо оригинального get метода. Но оригинальный get вызывается, когда исправленный объект обрабатывает вызов. Таким образом, вы можете получить как результат этого метода, так и некоторую дополнительную статистику, предоставляемую библиотекой unittest, например assert_called.

...