4 changed files with 770 additions and 617 deletions
@ -1,30 +0,0 @@
|
||||
import pytest |
||||
from pyircbot.modules.Calc import Calc |
||||
from pyircbot.pyircbot import ModuleLoader |
||||
|
||||
|
||||
class FakeBaseBot(ModuleLoader): |
||||
|
||||
" IRC methods " |
||||
def act_PRIVMSG(self, towho, message): |
||||
"""Use the `/msg` command |
||||
|
||||
:param towho: the target #channel or user's name |
||||
:type towho: str |
||||
:param message: the message to send |
||||
:type message: str""" |
||||
# self.sendRaw("PRIVMSG %s :%s" % (towho, message)) |
||||
print("act_PRIVMSG(towho={}, message={})".format(towho, message)) |
||||
|
||||
|
||||
@pytest.fixture |
||||
def fakebot(): |
||||
bot = FakeBaseBot() |
||||
bot.botconfig = {"bot": {"datadir": "./examples/data/"}} |
||||
bot.loadmodule("SQLite") |
||||
bot.loadmodule("Calc") |
||||
return bot |
||||
|
||||
|
||||
def test_foo(fakebot): |
||||
print(fakebot) |
@ -0,0 +1,221 @@
|
||||
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) |
Loading…
Reference in new issue