149 lines
3.2 KiB
Python
149 lines
3.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
from unittest import mock
|
|
|
|
from aiohttp import web
|
|
from bthlabs_jsonrpc_core import exceptions
|
|
|
|
from bthlabs_jsonrpc_aiohttp import views
|
|
|
|
|
|
def test_init():
|
|
# When
|
|
result = views.JSONRPCView()
|
|
|
|
# Then
|
|
assert result.namespace is None
|
|
|
|
|
|
def test_init_with_namespace():
|
|
# When
|
|
result = views.JSONRPCView(namespace='testing')
|
|
|
|
# Then
|
|
assert result.namespace == 'testing'
|
|
|
|
|
|
async def test_can_call(fake_request):
|
|
# Given
|
|
view = views.JSONRPCView()
|
|
|
|
# When
|
|
result = await view.can_call(fake_request, 'test', [], {})
|
|
|
|
# Then
|
|
assert result is True
|
|
|
|
|
|
async def test_view(aiohttp_client):
|
|
# Given
|
|
view = views.JSONRPCView()
|
|
|
|
app = web.Application()
|
|
app.router.add_post('/', view)
|
|
|
|
client = await aiohttp_client(app)
|
|
|
|
batch = [
|
|
{
|
|
'jsonrpc': '2.0',
|
|
'id': 'call_1',
|
|
'method': 'system.list_methods',
|
|
'params': ['spam'],
|
|
},
|
|
{
|
|
'jsonrpc': '2.0',
|
|
'id': 'call_2',
|
|
'method': 'idontexist',
|
|
},
|
|
{
|
|
'jsonrpc': '2.0',
|
|
'method': 'system.list_methods',
|
|
'params': {'spam': True},
|
|
},
|
|
]
|
|
|
|
# When
|
|
response = await client.post('/', json=batch)
|
|
|
|
# Then
|
|
assert response.status == 200
|
|
|
|
data = await response.json()
|
|
expected_result_data = [
|
|
{
|
|
'jsonrpc': '2.0',
|
|
'id': 'call_1',
|
|
'result': ['system.list_methods'],
|
|
},
|
|
{
|
|
'jsonrpc': '2.0',
|
|
'id': 'call_2',
|
|
'error': {
|
|
'code': exceptions.JSONRPCMethodNotFoundError.ERROR_CODE,
|
|
'message': exceptions.JSONRPCMethodNotFoundError.ERROR_MESSAGE,
|
|
},
|
|
},
|
|
]
|
|
assert data == expected_result_data
|
|
|
|
|
|
async def test_view_empty_response(aiohttp_client):
|
|
# Given
|
|
view = views.JSONRPCView()
|
|
|
|
app = web.Application()
|
|
app.router.add_post('/', view)
|
|
|
|
client = await aiohttp_client(app)
|
|
|
|
call = {
|
|
'jsonrpc': '2.0',
|
|
'method': 'system.list_methods',
|
|
}
|
|
|
|
# When
|
|
response = await client.post('/', json=call)
|
|
|
|
# Then
|
|
assert response.status == 200
|
|
|
|
data = await response.content.read()
|
|
assert data == b''
|
|
|
|
|
|
async def test_view_permission_denied(aiohttp_client):
|
|
# Given
|
|
view = views.JSONRPCView()
|
|
|
|
app = web.Application()
|
|
app.router.add_post('/', view)
|
|
|
|
client = await aiohttp_client(app)
|
|
|
|
call = {
|
|
'jsonrpc': '2.0',
|
|
'id': 'call_1',
|
|
'method': 'system.list_methods',
|
|
}
|
|
|
|
with mock.patch.object(view, 'can_call') as mock_can_call:
|
|
mock_can_call.return_value = False
|
|
|
|
# When
|
|
response = await client.post('/', json=call)
|
|
|
|
# Then
|
|
assert response.status == 200
|
|
|
|
data = await response.json()
|
|
expected_result_data = {
|
|
'jsonrpc': '2.0',
|
|
'id': 'call_1',
|
|
'error': {
|
|
'code': exceptions.JSONRPCAccessDeniedError.ERROR_CODE,
|
|
'message': exceptions.JSONRPCAccessDeniedError.ERROR_MESSAGE,
|
|
'data': 'can_call',
|
|
},
|
|
}
|
|
assert data == expected_result_data
|