test_args.test_extra_kwargs_with_class() - Code Metrics - Inspection of "Update API reference" - expert-m/aiohttp-rpc - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 1d9f68...a2ade1 )

by Michael

created 2020-05-30 18:46 UTC

test_args.test_extra_kwargs_with_class() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	16
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	12
dl	0
loc	16
rs	9.8
c	0
b	0
f	0
cc	2
nop	1

import functools
import os
import sys

import pytest

import aiohttp_rpc
from aiohttp_rpc import errors


sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from tests import utils


@pytest.mark.asyncio
async def test_args(aiohttp_client):
    def method(a=1):
        return [1, 2, a]

    rpc_server = aiohttp_rpc.JsonRpcServer()
    rpc_server.add_method(method)

    assert await rpc_server.call('method') == [1, 2, 1]
    assert await rpc_server.call('method', args=[1]) == [1, 2, 1]

    client = await utils.make_client(aiohttp_client, rpc_server)

    async with aiohttp_rpc.JsonRpcClient('/rpc', session=client) as rpc:
        assert await rpc.call('method') == [1, 2, 1]
        assert await rpc.call('method', 1) == [1, 2, 1]


@pytest.mark.asyncio
async def test_kwargs(aiohttp_client):
    def method(a=1, *, b=2):
        return [1, a, b]

    rpc_server = aiohttp_rpc.JsonRpcServer()
    rpc_server.add_method(method)

    with pytest.raises(errors.InvalidParams):
        await rpc_server.call('method', args=[1, 2])

    assert await rpc_server.call('method', kwargs={'a': 1, 'b': 2}) == [1, 1, 2]
    assert await rpc_server.call('method', args=[2], kwargs={'b': 2}) == [1, 2, 2]

    client = await utils.make_client(aiohttp_client, rpc_server)

    async with aiohttp_rpc.JsonRpcClient('/rpc', session=client) as rpc:
        assert await rpc.call('method', a=1, b=2) == [1, 1, 2]

        with pytest.raises(errors.InvalidParams):
            await rpc.call('method', 2, b=2)


@pytest.mark.asyncio
async def test_varargs(aiohttp_client):
    def method(a=1, *args):
        return [a, *args]

    rpc_server = aiohttp_rpc.JsonRpcServer()
    rpc_server.add_method(method)

    assert await rpc_server.call('method') == [1]
    assert await rpc_server.call('method', args=[2]) == [2]
    assert await rpc_server.call('method', args=[2, 3]) == [2, 3]

    client = await utils.make_client(aiohttp_client, rpc_server)

    async with aiohttp_rpc.JsonRpcClient('/rpc', session=client) as rpc:
        assert await rpc.call('method') == [1]
        assert await rpc.call('method', 2) == [2]
        assert await rpc.call('method', 2, 3) == [2, 3]


@pytest.mark.asyncio
async def test_varkw(aiohttp_client):
    def method(a=1, **kwargs):
        return [a, kwargs]

    rpc_server = aiohttp_rpc.JsonRpcServer()
    rpc_server.add_method(method)

    with pytest.raises(errors.InvalidParams):
        await rpc_server.call('method', args=[1, 2])

    assert await rpc_server.call('method', kwargs={'a': 1, 'b': 2}) == [1, {'b': 2}]

    client = await utils.make_client(aiohttp_client, rpc_server)

    async with aiohttp_rpc.JsonRpcClient('/rpc', session=client) as rpc:
        with pytest.raises(errors.InvalidParams):
            await rpc.call('method', 1, 2)

        assert await rpc.call('method', a=1, b=2) == [1, {'b': 2}]


@pytest.mark.asyncio
async def test_extra_kwargs(aiohttp_client):
    def method(rpc_request):
        return rpc_request.__class__.__name__

    def method_2(*, rpc_request):
        return rpc_request.__class__.__name__

    rpc_server = aiohttp_rpc.JsonRpcServer(middlewares=(aiohttp_rpc.middlewares.extra_args_middleware,))
    rpc_server.add_method(method)
    rpc_server.add_method(method_2)

    assert await rpc_server.call('method', extra_args={'rpc_request': 123}), 123
    assert await rpc_server.call('method_2', extra_args={'rpc_request': 123}), 123

    client = await utils.make_client(aiohttp_client, rpc_server)

    async with aiohttp_rpc.JsonRpcClient('/rpc', session=client) as rpc:
        assert await rpc.call('method') == 'JsonRpcRequest'
        assert await rpc.call('method_2') == 'JsonRpcRequest'


@pytest.mark.asyncio
async def test_extra_kwargs_with_class(aiohttp_client):
    class TestClass:
        def __init__(self, rpc_request):
            self.rpc_request = rpc_request

        def __str__(self):
            return self.rpc_request.__class__.__name__

    rpc_server = aiohttp_rpc.JsonRpcServer(middlewares=(aiohttp_rpc.middlewares.extra_args_middleware,))
    rpc_server.add_method(aiohttp_rpc.JsonRpcMethod('', TestClass, prepare_result=str))

    client = await utils.make_client(aiohttp_client, rpc_server)

    async with aiohttp_rpc.JsonRpcClient('/rpc', session=client) as rpc:
        assert await rpc.call('TestClass') == 'JsonRpcRequest'


@pytest.mark.asyncio
async def test_extra_kwargs_with_wrapper(aiohttp_client):
    def test_decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)

        return wrapper

    @test_decorator
    def method(rpc_request):
        return rpc_request.__class__.__name__

    @test_decorator
    def method_2():
        return True

    rpc_server = aiohttp_rpc.JsonRpcServer(middlewares=(aiohttp_rpc.middlewares.extra_args_middleware,))
    rpc_server.add_methods((method, method_2,))

    client = await utils.make_client(aiohttp_client, rpc_server)

    async with aiohttp_rpc.JsonRpcClient('/rpc', session=client) as rpc:
        assert await rpc.call('method') == 'JsonRpcRequest'
        assert await rpc.call('method_2') is True


@pytest.mark.asyncio
async def test_builtin_funcs(aiohttp_client):
    rpc_server = aiohttp_rpc.JsonRpcServer(middlewares=(aiohttp_rpc.middlewares.extra_args_middleware,))
    rpc_server.add_method(sum)
    rpc_server.add_method(aiohttp_rpc.JsonRpcMethod('', zip, prepare_result=list))

    client = await utils.make_client(aiohttp_client, rpc_server)

    async with aiohttp_rpc.JsonRpcClient('/rpc', session=client) as rpc:
        assert await rpc.sum([1, 2, 3]) == 6
        assert await rpc.zip(['a', 'b'], [1, 2]) == [['a', 1], ['b', 2]]


1			import functools
2			import os
3			import sys
4
5			import pytest
6
7			import aiohttp_rpc
8			from aiohttp_rpc import errors
9
10
11			sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
12			from tests import utils
13
14
15			@pytest.mark.asyncio
16			async def test_args(aiohttp_client):
17			def method(a=1):
18			return [1, 2, a]
19
20			rpc_server = aiohttp_rpc.JsonRpcServer()
21			rpc_server.add_method(method)
22
23			assert await rpc_server.call('method') == [1, 2, 1]
24			assert await rpc_server.call('method', args=[1]) == [1, 2, 1]
25
26			client = await utils.make_client(aiohttp_client, rpc_server)
27
28			async with aiohttp_rpc.JsonRpcClient('/rpc', session=client) as rpc:
29			assert await rpc.call('method') == [1, 2, 1]
30			assert await rpc.call('method', 1) == [1, 2, 1]
31
32
33			@pytest.mark.asyncio
34			async def test_kwargs(aiohttp_client):
35			def method(a=1, *, b=2):
36			return [1, a, b]
37
38			rpc_server = aiohttp_rpc.JsonRpcServer()
39			rpc_server.add_method(method)
40
41			with pytest.raises(errors.InvalidParams):
42			await rpc_server.call('method', args=[1, 2])
43
44			assert await rpc_server.call('method', kwargs={'a': 1, 'b': 2}) == [1, 1, 2]
45			assert await rpc_server.call('method', args=[2], kwargs={'b': 2}) == [1, 2, 2]
46
47			client = await utils.make_client(aiohttp_client, rpc_server)
48
49			async with aiohttp_rpc.JsonRpcClient('/rpc', session=client) as rpc:
50			assert await rpc.call('method', a=1, b=2) == [1, 1, 2]
51
52			with pytest.raises(errors.InvalidParams):
53			await rpc.call('method', 2, b=2)
54
55
56			@pytest.mark.asyncio
57			async def test_varargs(aiohttp_client):
58			def method(a=1, *args):
59			return [a, *args]
60
61			rpc_server = aiohttp_rpc.JsonRpcServer()
62			rpc_server.add_method(method)
63
64			assert await rpc_server.call('method') == [1]
65			assert await rpc_server.call('method', args=[2]) == [2]
66			assert await rpc_server.call('method', args=[2, 3]) == [2, 3]
67
68			client = await utils.make_client(aiohttp_client, rpc_server)
69
70			async with aiohttp_rpc.JsonRpcClient('/rpc', session=client) as rpc:
71			assert await rpc.call('method') == [1]
72			assert await rpc.call('method', 2) == [2]
73			assert await rpc.call('method', 2, 3) == [2, 3]
74
75
76			@pytest.mark.asyncio
77			async def test_varkw(aiohttp_client):
78			def method(a=1, **kwargs):
79			return [a, kwargs]
80
81			rpc_server = aiohttp_rpc.JsonRpcServer()
82			rpc_server.add_method(method)
83
84			with pytest.raises(errors.InvalidParams):
85			await rpc_server.call('method', args=[1, 2])
86
87			assert await rpc_server.call('method', kwargs={'a': 1, 'b': 2}) == [1, {'b': 2}]
88
89			client = await utils.make_client(aiohttp_client, rpc_server)
90
91			async with aiohttp_rpc.JsonRpcClient('/rpc', session=client) as rpc:
92			with pytest.raises(errors.InvalidParams):
93			await rpc.call('method', 1, 2)
94
95			assert await rpc.call('method', a=1, b=2) == [1, {'b': 2}]
96
97
98			@pytest.mark.asyncio
99			async def test_extra_kwargs(aiohttp_client):
100			def method(rpc_request):
101			return rpc_request.__class__.__name__
102
103			def method_2(*, rpc_request):
104			return rpc_request.__class__.__name__
105
106			rpc_server = aiohttp_rpc.JsonRpcServer(middlewares=(aiohttp_rpc.middlewares.extra_args_middleware,))
107			rpc_server.add_method(method)
108			rpc_server.add_method(method_2)
109
110			assert await rpc_server.call('method', extra_args={'rpc_request': 123}), 123
111			assert await rpc_server.call('method_2', extra_args={'rpc_request': 123}), 123
112
113			client = await utils.make_client(aiohttp_client, rpc_server)
114
115			async with aiohttp_rpc.JsonRpcClient('/rpc', session=client) as rpc:
116			assert await rpc.call('method') == 'JsonRpcRequest'
117			assert await rpc.call('method_2') == 'JsonRpcRequest'
118
119
120			@pytest.mark.asyncio
121			async def test_extra_kwargs_with_class(aiohttp_client):
122			class TestClass:
123			def __init__(self, rpc_request):
124			self.rpc_request = rpc_request
125
126			def __str__(self):
127			return self.rpc_request.__class__.__name__
128
129			rpc_server = aiohttp_rpc.JsonRpcServer(middlewares=(aiohttp_rpc.middlewares.extra_args_middleware,))
130			rpc_server.add_method(aiohttp_rpc.JsonRpcMethod('', TestClass, prepare_result=str))
131
132			client = await utils.make_client(aiohttp_client, rpc_server)
133
134			async with aiohttp_rpc.JsonRpcClient('/rpc', session=client) as rpc:
135			assert await rpc.call('TestClass') == 'JsonRpcRequest'
136
137
138			@pytest.mark.asyncio
139			async def test_extra_kwargs_with_wrapper(aiohttp_client):
140			def test_decorator(func):
141			@functools.wraps(func)
142			def wrapper(args, *kwargs):
143			return func(args, *kwargs)
144
145			return wrapper
146
147			@test_decorator
148			def method(rpc_request):
149			return rpc_request.__class__.__name__
150
151			@test_decorator
152			def method_2():
153			return True
154
155			rpc_server = aiohttp_rpc.JsonRpcServer(middlewares=(aiohttp_rpc.middlewares.extra_args_middleware,))
156			rpc_server.add_methods((method, method_2,))
157
158			client = await utils.make_client(aiohttp_client, rpc_server)
159
160			async with aiohttp_rpc.JsonRpcClient('/rpc', session=client) as rpc:
161			assert await rpc.call('method') == 'JsonRpcRequest'
162			assert await rpc.call('method_2') is True
163
164
165			@pytest.mark.asyncio
166			async def test_builtin_funcs(aiohttp_client):
167			rpc_server = aiohttp_rpc.JsonRpcServer(middlewares=(aiohttp_rpc.middlewares.extra_args_middleware,))
168			rpc_server.add_method(sum)
169			rpc_server.add_method(aiohttp_rpc.JsonRpcMethod('', zip, prepare_result=list))
170
171			client = await utils.make_client(aiohttp_client, rpc_server)
172
173			async with aiohttp_rpc.JsonRpcClient('/rpc', session=client) as rpc:
174			assert await rpc.sum([1, 2, 3]) == 6
175			assert await rpc.zip(['a', 'b'], [1, 2]) == [['a', 1], ['b', 2]]
176

expert-m / aiohttp-rpc

Push — master ( 1d9f68...a2ade1 )

test_args.test_extra_kwargs_with_class() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like