405 lines
11 KiB
Python
405 lines
11 KiB
Python
from pyln.client import Plugin
|
|
from pyln.client.plugin import Request, Millisatoshi
|
|
import itertools
|
|
import pytest # type: ignore
|
|
|
|
|
|
def test_simple_methods():
|
|
"""Test the dispatch of methods, with a variety of bindings.
|
|
"""
|
|
call_list = []
|
|
p = Plugin(autopatch=False)
|
|
|
|
@p.method("test1")
|
|
def test1(name):
|
|
"""Has a single positional argument."""
|
|
assert name == 'World'
|
|
call_list.append(test1)
|
|
request = p._parse_request({
|
|
'id': 1,
|
|
'jsonrpc': '2.0',
|
|
'method': 'test1',
|
|
'params': {'name': 'World'}
|
|
})
|
|
p._dispatch_request(request)
|
|
assert call_list == [test1]
|
|
|
|
@p.method("test2")
|
|
def test2(name, plugin):
|
|
"""Also asks for the plugin instance. """
|
|
assert plugin == p
|
|
call_list.append(test2)
|
|
request = p._parse_request({
|
|
'id': 1,
|
|
'jsonrpc': '2.0',
|
|
'method': 'test2',
|
|
'params': {'name': 'World'}
|
|
})
|
|
p._dispatch_request(request)
|
|
assert call_list == [test1, test2]
|
|
|
|
@p.method("test3")
|
|
def test3(name, request):
|
|
"""Also asks for the request instance. """
|
|
assert request is not None
|
|
call_list.append(test3)
|
|
request = p._parse_request({
|
|
'id': 1,
|
|
'jsonrpc': '2.0',
|
|
'method': 'test3',
|
|
'params': {'name': 'World'}
|
|
})
|
|
p._dispatch_request(request)
|
|
assert call_list == [test1, test2, test3]
|
|
|
|
@p.method("test4")
|
|
def test4(name):
|
|
"""Try the positional arguments."""
|
|
assert name == 'World'
|
|
call_list.append(test4)
|
|
request = p._parse_request({
|
|
'id': 1,
|
|
'jsonrpc': '2.0',
|
|
'method': 'test4',
|
|
'params': ['World']
|
|
})
|
|
p._dispatch_request(request)
|
|
assert call_list == [test1, test2, test3, test4]
|
|
|
|
@p.method("test5")
|
|
def test5(name, request, plugin):
|
|
"""Try the positional arguments, mixing in the request and plugin."""
|
|
assert name == 'World'
|
|
assert request is not None
|
|
assert p == plugin
|
|
call_list.append(test5)
|
|
request = p._parse_request({
|
|
'id': 1,
|
|
'jsonrpc': '2.0',
|
|
'method': 'test5',
|
|
'params': ['World']
|
|
})
|
|
p._dispatch_request(request)
|
|
assert call_list == [test1, test2, test3, test4, test5]
|
|
|
|
answers = []
|
|
|
|
@p.method("test6")
|
|
def test6(name, answer=42):
|
|
"""This method has a default value for one of its params"""
|
|
assert name == 'World'
|
|
answers.append(answer)
|
|
call_list.append(test6)
|
|
|
|
# Both calls should work (with and without the default param
|
|
request = p._parse_request({
|
|
'id': 1,
|
|
'jsonrpc': '2.0',
|
|
'method': 'test6',
|
|
'params': ['World']
|
|
})
|
|
p._dispatch_request(request)
|
|
assert call_list == [test1, test2, test3, test4, test5, test6]
|
|
assert answers == [42]
|
|
|
|
request = p._parse_request({
|
|
'id': 1,
|
|
'jsonrpc': '2.0',
|
|
'method': 'test6',
|
|
'params': ['World', 31337]
|
|
})
|
|
p._dispatch_request(request)
|
|
assert call_list == [test1, test2, test3, test4, test5, test6, test6]
|
|
assert answers == [42, 31337]
|
|
|
|
|
|
def test_methods_errors():
|
|
"""A bunch of tests that should fail calling the methods."""
|
|
call_list = []
|
|
p = Plugin(autopatch=False)
|
|
|
|
# Fails because we haven't added the method yet
|
|
request = p._parse_request({
|
|
'id': 1,
|
|
'jsonrpc': '2.0',
|
|
'method': 'test1',
|
|
'params': {}
|
|
})
|
|
with pytest.raises(ValueError):
|
|
p._dispatch_request(request)
|
|
assert call_list == []
|
|
|
|
@p.method("test1")
|
|
def test1(name):
|
|
call_list.append(test1)
|
|
|
|
# Attempting to add it twice should fail
|
|
with pytest.raises(ValueError):
|
|
p.add_method("test1", test1)
|
|
|
|
# Fails because it is missing the 'name' argument
|
|
request = p._parse_request({'id': 1, 'jsonrpc': '2.0', 'method': 'test1', 'params': {}})
|
|
with pytest.raises(TypeError):
|
|
p._exec_func(test1, request)
|
|
assert call_list == []
|
|
|
|
# The same with positional arguments
|
|
request = p._parse_request({'id': 1, 'jsonrpc': '2.0', 'method': 'test1', 'params': []})
|
|
with pytest.raises(TypeError):
|
|
p._exec_func(test1, request)
|
|
assert call_list == []
|
|
|
|
# Fails because we have a non-matching argument
|
|
request = p._parse_request({
|
|
'id': 1,
|
|
'jsonrpc': '2.0',
|
|
'method': 'test1',
|
|
'params': {'name': 'World', 'extra': 1}
|
|
})
|
|
with pytest.raises(TypeError):
|
|
p._exec_func(test1, request)
|
|
assert call_list == []
|
|
|
|
request = p._parse_request({
|
|
'id': 1,
|
|
'jsonrpc': '2.0',
|
|
'method': 'test1',
|
|
'params': ['World', 1]
|
|
})
|
|
|
|
with pytest.raises(TypeError):
|
|
p._exec_func(test1, request)
|
|
assert call_list == []
|
|
|
|
|
|
def test_positional_inject():
|
|
p = Plugin()
|
|
rdict = Request(
|
|
plugin=p,
|
|
req_id=1,
|
|
method='func',
|
|
params={'a': 1, 'b': 2, 'kwa': 3, 'kwb': 4}
|
|
)
|
|
rarr = Request(
|
|
plugin=p,
|
|
req_id=1,
|
|
method='func',
|
|
params=[1, 2, 3, 4],
|
|
)
|
|
|
|
def pre_args(plugin, a, b, kwa=3, kwb=4):
|
|
assert (plugin, a, b, kwa, kwb) == (p, 1, 2, 3, 4)
|
|
|
|
def in_args(a, plugin, b, kwa=3, kwb=4):
|
|
assert (plugin, a, b, kwa, kwb) == (p, 1, 2, 3, 4)
|
|
|
|
def post_args(a, b, plugin, kwa=3, kwb=4):
|
|
assert (plugin, a, b, kwa, kwb) == (p, 1, 2, 3, 4)
|
|
|
|
def post_kwargs(a, b, kwa=3, kwb=4, plugin=None):
|
|
assert (plugin, a, b, kwa, kwb) == (p, 1, 2, 3, 4)
|
|
|
|
def in_multi_args(a, request, plugin, b, kwa=3, kwb=4):
|
|
assert request in [rarr, rdict]
|
|
assert (plugin, a, b, kwa, kwb) == (p, 1, 2, 3, 4)
|
|
|
|
def in_multi_mix_args(a, plugin, b, request=None, kwa=3, kwb=4):
|
|
assert request in [rarr, rdict]
|
|
assert (plugin, a, b, kwa, kwb) == (p, 1, 2, 3, 4)
|
|
|
|
def extra_def_arg(a, b, c, d, e=42):
|
|
""" Also uses a different name for kwa and kwb
|
|
"""
|
|
assert (a, b, c, d, e) == (1, 2, 3, 4, 42)
|
|
|
|
def count(plugin, count, request):
|
|
assert count == 42 and plugin == p
|
|
|
|
funcs = [pre_args, in_args, post_args, post_kwargs, in_multi_args]
|
|
|
|
for func, request in itertools.product(funcs, [rdict, rarr]):
|
|
p._exec_func(func, request)
|
|
|
|
p._exec_func(extra_def_arg, rarr)
|
|
|
|
p._exec_func(count, Request(
|
|
plugin=p,
|
|
req_id=1,
|
|
method='func',
|
|
params=[42],
|
|
))
|
|
|
|
# This should fail since it is missing one positional argument
|
|
with pytest.raises(TypeError):
|
|
p._exec_func(count, Request(
|
|
plugin=p,
|
|
req_id=1,
|
|
method='func',
|
|
params=[])
|
|
)
|
|
|
|
|
|
def test_bind_pos():
|
|
p = Plugin(autopatch=False)
|
|
|
|
req = object()
|
|
params = ['World']
|
|
|
|
def test1(name):
|
|
assert name == 'World'
|
|
bound = p._bind_pos(test1, params, req)
|
|
test1(*bound.args, **bound.kwargs)
|
|
|
|
def test2(name, plugin):
|
|
assert name == 'World'
|
|
assert plugin == p
|
|
bound = p._bind_pos(test2, params, req)
|
|
test2(*bound.args, **bound.kwargs)
|
|
|
|
def test3(plugin, name):
|
|
assert name == 'World'
|
|
assert plugin == p
|
|
bound = p._bind_pos(test3, params, req)
|
|
test3(*bound.args, **bound.kwargs)
|
|
|
|
def test4(plugin, name, request):
|
|
assert name == 'World'
|
|
assert plugin == p
|
|
assert request == req
|
|
bound = p._bind_pos(test4, params, req)
|
|
test4(*bound.args, **bound.kwargs)
|
|
|
|
def test5(request, name, plugin):
|
|
assert name == 'World'
|
|
assert plugin == p
|
|
assert request == req
|
|
bound = p._bind_pos(test5, params, req)
|
|
test5(*bound.args, **bound.kwargs)
|
|
|
|
def test6(request, name, plugin, answer=42):
|
|
assert name == 'World'
|
|
assert plugin == p
|
|
assert request == req
|
|
assert answer == 42
|
|
bound = p._bind_pos(test6, params, req)
|
|
test6(*bound.args, **bound.kwargs)
|
|
|
|
# Now mix in a catch-all parameter that needs to be assigned
|
|
def test6(request, name, plugin, *args, **kwargs):
|
|
assert name == 'World'
|
|
assert plugin == p
|
|
assert request == req
|
|
assert args == (42,)
|
|
assert kwargs == {}
|
|
bound = p._bind_pos(test6, params + [42], req)
|
|
test6(*bound.args, **bound.kwargs)
|
|
|
|
|
|
def test_bind_kwargs():
|
|
p = Plugin(autopatch=False)
|
|
|
|
req = object()
|
|
params = {'name': 'World'}
|
|
|
|
def test1(name):
|
|
assert name == 'World'
|
|
bound = p._bind_kwargs(test1, params, req)
|
|
test1(*bound.args, **bound.kwargs)
|
|
|
|
def test2(name, plugin):
|
|
assert name == 'World'
|
|
assert plugin == p
|
|
bound = p._bind_kwargs(test2, params, req)
|
|
test2(*bound.args, **bound.kwargs)
|
|
|
|
def test3(plugin, name):
|
|
assert name == 'World'
|
|
assert plugin == p
|
|
bound = p._bind_kwargs(test3, params, req)
|
|
test3(*bound.args, **bound.kwargs)
|
|
|
|
def test4(plugin, name, request):
|
|
assert name == 'World'
|
|
assert plugin == p
|
|
assert request == req
|
|
bound = p._bind_kwargs(test4, params, req)
|
|
test4(*bound.args, **bound.kwargs)
|
|
|
|
def test5(request, name, plugin):
|
|
assert name == 'World'
|
|
assert plugin == p
|
|
assert request == req
|
|
bound = p._bind_kwargs(test5, params, req)
|
|
test5(*bound.args, **bound.kwargs)
|
|
|
|
def test6(request, name, plugin, answer=42):
|
|
assert name == 'World'
|
|
assert plugin == p
|
|
assert request == req
|
|
assert answer == 42
|
|
bound = p._bind_kwargs(test6, params, req)
|
|
test6(*bound.args, **bound.kwargs)
|
|
|
|
# Now mix in a catch-all parameter that needs to be assigned
|
|
def test6(request, name, plugin, *args, **kwargs):
|
|
assert name == 'World'
|
|
assert plugin == p
|
|
assert request == req
|
|
assert args == ()
|
|
assert kwargs == {'answer': 42}
|
|
bound = p._bind_kwargs(test6, {'name': 'World', 'answer': 42}, req)
|
|
test6(*bound.args, **bound.kwargs)
|
|
|
|
|
|
def test_argument_coercion():
|
|
p = Plugin(autopatch=False)
|
|
|
|
def test1(msat: Millisatoshi):
|
|
assert isinstance(msat, Millisatoshi)
|
|
|
|
ba = p._bind_kwargs(test1, {"msat": "100msat"}, None)
|
|
test1(*ba.args)
|
|
|
|
ba = p._bind_pos(test1, ["100msat"], None)
|
|
test1(*ba.args, **ba.kwargs)
|
|
|
|
|
|
def test_duplicate_result():
|
|
p = Plugin(autopatch=False)
|
|
|
|
def test1(request):
|
|
request.set_result(1) # MARKER1
|
|
request.set_result(1)
|
|
|
|
req = Request(p, req_id=1, method="test1", params=[])
|
|
ba = p._bind_kwargs(test1, {}, req)
|
|
with pytest.raises(ValueError, match=r'current state is RequestState\.FINISHED(.*\n.*)*MARKER1'):
|
|
test1(*ba.args)
|
|
|
|
def test2(request):
|
|
request.set_exception(1) # MARKER2
|
|
request.set_exception(1)
|
|
|
|
req = Request(p, req_id=2, method="test2", params=[])
|
|
ba = p._bind_kwargs(test2, {}, req)
|
|
with pytest.raises(ValueError, match=r'current state is RequestState\.FAILED(.*\n*.*)*MARKER2'):
|
|
test2(*ba.args)
|
|
|
|
def test3(request):
|
|
request.set_exception(1) # MARKER3
|
|
request.set_result(1)
|
|
|
|
req = Request(p, req_id=3, method="test3", params=[])
|
|
ba = p._bind_kwargs(test3, {}, req)
|
|
with pytest.raises(ValueError, match=r'current state is RequestState\.FAILED(.*\n*.*)*MARKER3'):
|
|
test3(*ba.args)
|
|
|
|
def test4(request):
|
|
request.set_result(1) # MARKER4
|
|
request.set_exception(1)
|
|
|
|
req = Request(p, req_id=4, method="test4", params=[])
|
|
ba = p._bind_kwargs(test4, {}, req)
|
|
with pytest.raises(ValueError, match=r'current state is RequestState\.FINISHED(.*\n*.*)*MARKER4'):
|
|
test4(*ba.args)
|