Awesome IRC bot
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

221 lines
7.2 KiB

import os
import pytest
from pyircbot import jsonrpc
from threading import Thread
from random import randint
from socket import SHUT_RDWR
from time import sleep
# Sample server methods
def sample(value):
return value
class _sample(object):
def sample(self, value):
return value
def client(port, v=2):
return jsonrpc.ServerProxy((jsonrpc.JsonRpc20 if v == 2 else jsonrpc.JsonRpc10)(),
jsonrpc.TransportTcpIp(addr=("127.0.0.1", port), timeout=2.0))
# Fixures for each server version provide a (server_instance, port) tuple.
# Each have the method "sample", which returns the value passed
# Each have a class instance registered as "obj", which the method "sample" as well
@pytest.fixture
def j1testserver():
port = randint(40000, 60000)
server = jsonrpc.Server(jsonrpc.JsonRpc10(),
jsonrpc.TransportTcpIp(addr=("127.0.0.1", port)))
server.register_function(sample)
server.register_instance(_sample(), name="obj")
Thread(target=server.serve, daemon=True).start()
sleep(0.1) # Give the serve() time to set up the serversocket
yield (server, port)
server._Server__transport.s.shutdown(SHUT_RDWR)
@pytest.fixture
def j2testserver():
port = randint(40000, 60000)
server = jsonrpc.Server(jsonrpc.JsonRpc20(),
jsonrpc.TransportTcpIp(addr=("127.0.0.1", port)))
server.register_function(sample)
server.register_instance(_sample(), name="obj")
Thread(target=server.serve, daemon=True).start()
sleep(0.1) # Give the serve() time to set up the serversocket
yield (server, port)
server._Server__transport.s.shutdown(SHUT_RDWR)
# Basic functionality
def test_1_basic(j1testserver):
str(jsonrpc.RPCFault(-32700, "foo", "bar"))
server, port = j1testserver
str(client(port, v=1))
ret = client(port, v=1).sample("foobar")
assert ret == "foobar"
def test_2_basic(j2testserver):
server, port = j2testserver
str(client(port))
ret = client(port).sample("foobar")
assert ret == "foobar"
def test_1_instance(j1testserver):
server, port = j1testserver
ret = client(port, v=1).obj.sample("foobar")
assert ret == "foobar"
def test_2_instance(j2testserver):
server, port = j2testserver
ret = client(port).obj.sample("foobar")
assert ret == "foobar"
# Missing methods raise clean error
def test_1_notfound(j1testserver):
server, port = j1testserver
with pytest.raises(jsonrpc.RPCMethodNotFound):
client(port, v=1).idontexist("f")
with pytest.raises(jsonrpc.RPCMethodNotFound):
client(port, v=1).neither.idontexist("f")
def test_2_notfound(j2testserver):
server, port = j2testserver
with pytest.raises(jsonrpc.RPCMethodNotFound):
client(port).idontexist("f")
with pytest.raises(jsonrpc.RPCMethodNotFound):
client(port).neither.idontexist("f")
# Underscore methods are blocked
def test_1_underscore():
with pytest.raises(AttributeError):
client(-1)._notallowed()
def test_2_underscore():
with pytest.raises(AttributeError):
client(-1)._notallowed()
# Response parsing hardness
def _test_1_protocol_parse_base(method):
with pytest.raises(jsonrpc.RPCParseError): # Not json
method("")
with pytest.raises(jsonrpc.RPCInvalidRPC): # Not a dict
method("[]")
with pytest.raises(jsonrpc.RPCInvalidRPC): # Missing 'id'
method("{}")
with pytest.raises(jsonrpc.RPCInvalidRPC): # not 3 fields
method('{"id": 0, "baz": 0}')
def _test_2_protocol_parse_base(method):
with pytest.raises(jsonrpc.RPCParseError): # Not json
method("")
with pytest.raises(jsonrpc.RPCInvalidRPC): # Not a dict
method("[]")
with pytest.raises(jsonrpc.RPCInvalidRPC): # missing jsonrpc
method('{}')
with pytest.raises(jsonrpc.RPCInvalidRPC): # jsonrpc must be str
method('{"jsonrpc": 1}')
with pytest.raises(jsonrpc.RPCInvalidRPC): # jsonrpc must be "2.0"
method('{"jsonrpc": "2.1"}')
def test_1_invalid_response():
j = jsonrpc.JsonRpc10()
_test_1_protocol_parse_base(j.loads_response)
with pytest.raises(jsonrpc.RPCInvalidRPC): # can't have result and error
j.loads_response('{"id": 0, "result": 1, "error": 0}')
def test_2_invalid_response():
j = jsonrpc.JsonRpc20()
_test_2_protocol_parse_base(j.loads_response)
with pytest.raises(jsonrpc.RPCInvalidRPC): # Missing 'id'
j.loads_response('{"jsonrpc": "2.0"}')
with pytest.raises(jsonrpc.RPCInvalidRPC): # not 4 fields
j.loads_response('{"id": 0, "jsonrpc": "2.0", "bar": 1}')
with pytest.raises(jsonrpc.RPCInvalidRPC): # can't have result and error
j.loads_response('{"id": 0, "jsonrpc": "2.0", "result": 1, "error": 0}')
# Request parsing hardness
def test_1_invalid_request():
j = jsonrpc.JsonRpc10()
_test_1_protocol_parse_base(j.loads_request)
with pytest.raises(jsonrpc.RPCInvalidRPC): # missing method
j.loads_request('{"id": 0}')
with pytest.raises(jsonrpc.RPCInvalidRPC): # method must be str
j.loads_request('{"id": 0, "method": -1}')
with pytest.raises(jsonrpc.RPCInvalidRPC): # params is bad type
j.loads_request('{"id": 0, "method": "foo", "params": -1}')
with pytest.raises(jsonrpc.RPCInvalidRPC): # wrong number of fields
j.loads_request('{"ba": 0, "method": "foo", "asdf": 1, "foobar": 2}')
j.loads_request('{"id": 0, "method": "foo", "params": []}')
j.loads_request('{"method": "foo", "params": []}')
# Request parsing hardness
def test_2_invalid_request():
j = jsonrpc.JsonRpc20()
_test_2_protocol_parse_base(j.loads_request)
with pytest.raises(jsonrpc.RPCInvalidRPC): # missing method
j.loads_request('{"id": 0, "jsonrpc": "2.0"}')
with pytest.raises(jsonrpc.RPCInvalidRPC): # method must be str
j.loads_request('{"id": 0, "jsonrpc": "2.0", "method": 1}')
with pytest.raises(jsonrpc.RPCInvalidRPC): # params is bad type
j.loads_request('{"id": 0, "jsonrpc": "2.0", "method": "foo", "params": -1}')
with pytest.raises(jsonrpc.RPCInvalidRPC): # wrong number of fields
j.loads_request('{"id": 0, "jsonrpc": "2.0", "method": "foo", "asdf": 1, "foobar": 2}')
j.loads_request('{"id": 0, "jsonrpc": "2.0", "method": "foo", "params": []}')
j.loads_request('{"jsonrpc": "2.0", "method": "foo", "params": []}')
def test_1_dumps_reqest():
j = jsonrpc.JsonRpc20()
with pytest.raises(TypeError):
j.dumps_request(-1)
with pytest.raises(TypeError):
j.dumps_request("foo", params=-1)
j.dumps_request("foo")
def test_2_dumps_reqest():
j = jsonrpc.JsonRpc20()
with pytest.raises(TypeError):
j.dumps_request(-1)
with pytest.raises(TypeError):
j.dumps_request("foo", params=-1)
j.dumps_request("foo", params=[])
j.dumps_request("foo")
# Misc stuff
def test_logging(tmpdir):
msg = "test log message"
jsonrpc.log_dummy(msg)
jsonrpc.log_stdout(msg)
logpath = os.path.join(tmpdir, "test.log")
logger = jsonrpc.log_file(logpath)
logger(msg)
assert os.path.exists(logpath)
logpath = os.path.join(tmpdir, "test2.log")
logger2 = jsonrpc.log_filedate(os.path.join(tmpdir, "test2.log"))
logger2(msg)
assert os.path.exists(logpath)