mirror/galaxy-integrations-python-api
1
0
Fork 0
mirror of https://github.com/gogcom/galaxy-integrations-python-api.git synced 2026-05-24 06:26:33 -04:00
Code Issues Packages Projects Releases Wiki Activity

Use MagicMocks with async_return_value and pytest.mark.asyncio

Browse Source
This commit is contained in:
Romuald Bierbasz
2019-08-02 12:28:13 +02:00
parent b14595bef5
commit 3bcc674518
17 changed files with 505 additions and 475 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
tests/test_persistent_cache.py
View File

@@ -1,23 +1,28 @@
import asyncio
import json
import pytest
from galaxy.unittest.mock import async_return_value
from tests import create_message, get_messages
def assert_rpc_response(write, response_id, result=None):
assert json.loads(write.call_args[0][0]) == {
"jsonrpc": "2.0",
"id": str(response_id),
"result": result
}
assert get_messages(write) == [
{
"jsonrpc": "2.0",
"id": str(response_id),
"result": result
}
]
def assert_rpc_request(write, method, params=None):
assert json.loads(write.call_args[0][0]) == {
"jsonrpc": "2.0",
"method": method,
"params": {"data": params}
}
assert get_messages(write) == [
{
"jsonrpc": "2.0",
"method": method,
"params": {"data": params}
}
]
@pytest.fixture
@@ -28,7 +33,8 @@ def cache_data():
}
def test_initialize_cache(plugin, read, write, cache_data):
@pytest.mark.asyncio
async def test_initialize_cache(plugin, read, write, cache_data):
request_id = 3
request = {
"jsonrpc": "2.0",
@@ -36,36 +42,32 @@ def test_initialize_cache(plugin, read, write, cache_data):
"method": "initialize_cache",
"params": {"data": cache_data}
}
read.side_effect = [json.dumps(request).encode() + b"\n"]
read.side_effect = [async_return_value(create_message(request)), async_return_value(b"")]
assert {} == plugin.persistent_cache
asyncio.run(plugin.run())
await plugin.run()
plugin.handshake_complete.assert_called_once_with()
assert cache_data == plugin.persistent_cache
assert_rpc_response(write, response_id=request_id)
def test_set_cache(plugin, write, cache_data):
async def runner():
assert {} == plugin.persistent_cache
@pytest.mark.asyncio
async def test_set_cache(plugin, write, cache_data):
assert {} == plugin.persistent_cache
plugin.persistent_cache.update(cache_data)
plugin.push_cache()
plugin.persistent_cache.update(cache_data)
plugin.push_cache()
assert_rpc_request(write, "push_cache", cache_data)
assert cache_data == plugin.persistent_cache
asyncio.run(runner())
assert_rpc_request(write, "push_cache", cache_data)
assert cache_data == plugin.persistent_cache
def test_clear_cache(plugin, write, cache_data):
async def runner():
plugin._persistent_cache = cache_data
@pytest.mark.asyncio
async def test_clear_cache(plugin, write, cache_data):
plugin._persistent_cache = cache_data
plugin.persistent_cache.clear()
plugin.push_cache()
plugin.persistent_cache.clear()
plugin.push_cache()
assert_rpc_request(write, "push_cache", {})
assert {} == plugin.persistent_cache
asyncio.run(runner())
assert_rpc_request(write, "push_cache", {})
assert {} == plugin.persistent_cache