Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_id_on_errored_request(self):
# Test that parsing a ping request works
client = HttpClient()
client.send_and_consume(u'websocket.receive',
text='{"id":52, "jsonrpc":"2.0", "method":"ping", "params":["test"]}')
msg = client.receive()
self.assertEqual(msg['id'], 52)
def test_connection(self):
# Test connection
client = HttpClient()
client.send_and_consume(u'websocket.connect')
self.assertEquals(client.receive(), None)
def test_error_on_rpc_call_with_data(self):
@MyJsonRpcWebsocketConsumerTest.rpc_method()
def ping_with_error_data():
raise Exception("test_data", True)
# Test that parsing a ping request works
client = HttpClient()
client.send_and_consume(u'websocket.receive',
text='{"id":1, "jsonrpc":"2.0", "method":"ping_with_error_data", "params":{}}')
msg = client.receive()
self.assertEqual(msg['id'], 1)
self.assertEqual(msg['error']['code'], JsonRpcConsumerTest.GENERIC_APPLICATION_ERROR)
self.assertEqual(msg['error']['data'], ['test_data', True])
def test_custom_json_encoder(self):
some_date = datetime.utcnow()
@MyJsonRpcWebsocketConsumerTest.rpc_method()
def test_method():
return {
'date': some_date
}
client = HttpClient()
try:
client.send_and_consume(u'websocket.receive',
text='{"id":1, "jsonrpc":"2.0", "method":"test_method", "params":{}}')
self.fail('Looks like test does not work')
except TypeError:
pass
@DjangoJsonRpcWebsocketConsumerTest.rpc_method()
def test_method1():
return {
'date': some_date
}
client.send_and_consume(u'websocket.receive',
text='{"id":1, "jsonrpc":"2.0", "method":"test_method1", "params":{}}',
path='/django/')
def test_inbound_notifications(self):
@MyJsonRpcWebsocketConsumerTest.rpc_notification()
def notif1(params, **kwargs):
self.assertEqual(params, {"payload": True})
@MyJsonRpcWebsocketConsumerTest.rpc_notification('notif.notif2')
def notif2(params, **kwargs):
self.assertEqual(params, {"payload": 12345})
client = HttpClient()
# we send a notification to the server
client.send_and_consume(u'websocket.receive',
text='{"jsonrpc":"2.0", "method":"notif1", "params":[{"payload": true}]}')
msg = client.receive()
self.assertEqual(msg, None)
# we test with method rewriting
client.send_and_consume(u'websocket.receive',
text='{"jsonrpc":"2.0", "method":"notif.notif2", "params":[{"payload": 12345}]}')
self.assertEqual(msg, None)
def test_response_are_well_formatted(self):
# Answer should always json-rpc2
client = HttpClient()
client.send_and_consume(u'websocket.receive', {'value': 'my_value'})
response = client.receive()
self.assertEqual(response['error'],
{u'code': JsonRpcConsumerTest.INVALID_REQUEST,
u'message': JsonRpcConsumerTest.errors[JsonRpcConsumerTest.INVALID_REQUEST]})
self.assertEqual(response['jsonrpc'], '2.0')
if 'id' in response:
self.assertEqual(response['id'], None)
def test_parsing_with_good_request(self):
# Test that parsing a ping request works
client = HttpClient()
client.send_and_consume(u'websocket.receive',
text='{"id":1, "jsonrpc":"2.0", "method":"ping", "params":[false]}')
msg = client.receive()
self.assertEquals(msg['result'], "pong")
def test_method(self):
@MyJsonRpcWebsocketConsumer.rpc_method()
def ping2():
return "pong2"
client = HttpClient()
client.send_and_consume(u'websocket.receive', text='{"id":1, "jsonrpc":"2.0", "method":"ping2", "params":{}}')
msg = client.receive()
self.assertEqual(msg['result'], "pong2")
def test_inadequate_request(self):
client = HttpClient()
client.send_and_consume(u'websocket.receive', {'value': 'my_value'})
self.assertEqual(client.receive()['error'], {u'code': JsonRpcWebsocketConsumer.INVALID_REQUEST,
u'message': JsonRpcWebsocketConsumer.errors[JsonRpcWebsocketConsumer.INVALID_REQUEST]})
client.send_and_consume(u'websocket.receive', text='{"value": "my_value"}')
self.assertEqual(client.receive()['error'], {u'code': JsonRpcWebsocketConsumer.INVALID_REQUEST,
u'message': JsonRpcWebsocketConsumer.errors[JsonRpcWebsocketConsumer.INVALID_REQUEST]})
client.send_and_consume(u'websocket.receive', text='sqwdw')
self.assertEqual(client.receive()['error'], {u'code': JsonRpcWebsocketConsumer.PARSE_ERROR,
u'message': JsonRpcWebsocketConsumer.errors[JsonRpcWebsocketConsumer.PARSE_ERROR]})
client.send_and_consume(u'websocket.receive', {})
self.assertEqual(client.receive()['error'], {u'code': JsonRpcWebsocketConsumer.INVALID_REQUEST,
u'message': JsonRpcWebsocketConsumer.errors[JsonRpcWebsocketConsumer.INVALID_REQUEST]})
def test_parsing_with_good_request_wrong_params(self):
@JsonRpcWebsocketConsumer.rpc_method()
def ping2():
return "pong2"
# Test that parsing a ping request works
client = HttpClient()
client.send_and_consume(u'websocket.receive',
text='{"id":1, "jsonrpc":"2.0", "method":"ping2", "params":["test"]}')
msg = client.receive()
self.assertIn(msg['error']['message'],
[u'ping2() takes 0 positional arguments but 1 was given',
u'ping2() takes no arguments (1 given)'])