bthlabs-jsonrpc/packages/bthlabs-jsonrpc-aiohttp/tests/views/test_JSONRPCView.py

# -*- coding: utf-8 -*-
# type: ignore
from __future__ import annotations

from unittest import mock

from aiohttp.test_utils import TestClient
from aiohttp import web
from bthlabs_jsonrpc_core import codecs, exceptions
import pytest

from bthlabs_jsonrpc_aiohttp import views
from bthlabs_jsonrpc_aiohttp.executor import AioHttpExecutor


@pytest.fixture
def fake_aiohttp_executor() -> mock.Mock:
    return mock.Mock(spec=AioHttpExecutor)


def test_init():
    # When
    result = views.JSONRPCView()

    # Then
    assert result.namespace is None
    assert result.codec == codecs.JSONCodec


def test_init_with_namespace():
    # When
    result = views.JSONRPCView(namespace='testing')

    # Then
    assert result.namespace == 'testing'


def test_init_with_codec(fake_custom_codec: mock.Mock):
    # When
    result = views.JSONRPCView(codec=fake_custom_codec)

    # Then
    assert result.codec == fake_custom_codec


async def test_get_executor(fake_request: mock.Mock):
    # Given
    view = views.JSONRPCView()

    # When
    result = await view.get_executor(fake_request)

    # Then
    assert isinstance(result, views.AioHttpExecutor) is True


async def test_get_executor_dependency_calls(fake_aiohttp_executor: mock.Mock,
                                             fake_custom_codec: mock.Mock,
                                             fake_request: mock.Mock):
    # Given
    with mock.patch.object(views, 'AioHttpExecutor') as mock_aiohttp_executor:
        with mock.patch.object(views.JSONRPCView, 'get_codec') as mock_get_codec:
            mock_aiohttp_executor.return_value = fake_aiohttp_executor
            mock_get_codec.return_value = fake_custom_codec

            view = views.JSONRPCView()

            # When
            result = await view.get_executor(fake_request)

            # Then
            assert result == fake_aiohttp_executor

            mock_get_codec.assert_awaited_once_with(fake_request)

            mock_aiohttp_executor.assert_called_once_with(
                fake_request,
                view.can_call,
                namespace=view.namespace,
                codec=fake_custom_codec,
            )


async def test_can_call(fake_request: mock.Mock):
    # Given
    view = views.JSONRPCView()

    # When
    result = await view.can_call(fake_request, 'test', [], {})

    # Then
    assert result is True


async def test_view(aiohttp_client: TestClient):
    # 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: TestClient):
    # 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: TestClient):
    # 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


async def test_view_async_codec_encode(async_json_codec: mock.Mock,
                                       aiohttp_client: TestClient):
    # Given
    mock_codec = mock.Mock(return_value=async_json_codec)

    view = views.JSONRPCView(codec=mock_codec)

    app = web.Application()
    app.router.add_post('/', view)

    client = await aiohttp_client(app)

    call = {
        'jsonrpc': '2.0',
        'id': 'test',
        'method': 'system.list_methods',
    }

    # When
    response = await client.post('/', json=call)

    # Then
    assert response.status == 200

    data = await response.json()
    expected_result_data = {
        'jsonrpc': '2.0',
        'id': 'test',
        'result': ['system.list_methods'],
    }
    assert data == expected_result_data