mirror.MikroTik/socialwifi.RouterOS-api
1
0
Fork 0
mirror of https://github.com/socialwifi/RouterOS-api.git synced 2025-08-31 15:19:25 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Fix tests.

Browse source
This commit is contained in:
Jakub Skiepko 2014-07-03 12:06:21 +02:00
parent 9303ea684c
commit 576563366b
2 changed files with 31 additions and 34 deletions
Download patch file Download diff file Expand all files Collapse all files

6
routeros_api/api_communicator/key_cleaner_decorator.py
View file

@ -17,7 +17,8 @@ class KeyCleanerApiCommunicator(object):
def encode_dictionary(dictionary):
return {encode_key(key): value for key, value in dictionary.items()}
return dict([(encode_key(key), value) for key, value in
dictionary.items()])
def encode_key(key):
@ -28,7 +29,8 @@ def encode_key(key):
return key
def decode_dictionary(dictionary):
return {decode_key(key): value for key, value in dictionary.items()}
return dict([(decode_key(key), value) for key, value in
dictionary.items()])
def decode_key(key):

59
tests/test_api_communicator.py
View file

@ -11,36 +11,29 @@ from routeros_api import api_communicator
class TestResponseSentence(TestCase):
def test_login_call(self):
base = mock.Mock()
base.receive_sentence.return_value = [b'!done', b'=ret=some-hex']
base.receive_sentence.return_value = [b'!done', b'=ret=some-hex',
b'.tag=1']
communicator = api_communicator.ApiCommunicator(base)
response = communicator.call('/', 'login', include_done=True)
response = communicator.call('/', 'login', include_done=True).get()
self.assertEqual(response[0]['ret'], 'some-hex')
def test_normal_call(self):
base = mock.Mock()
base.receive_sentence.side_effect = [[b'!re', b'=x=y'],[b'!done']]
communicator = api_communicator.ApiCommunicator(base)
response = communicator.call('/interface/', 'print')
self.assertEqual(response, [{'x': 'y'}])
def test_async_call(self):
base = mock.Mock()
base.receive_sentence.side_effect = [[b'!re', b'=x=y', b'.tag=1'],
[b'!done', b'.tag=1']]
communicator = api_communicator.ApiCommunicator(base)
promise = communicator.call_async('/interface/', 'print')
response = promise.get()
response = communicator.call('/interface/', 'print').get()
self.assertEqual(response, [{'x': 'y'}])
def test_mixed_calls(self):
base = mock.Mock()
base.receive_sentence.side_effect = [[b'!re', b'=x1=y1', b'.tag=1'],
[b'!re', b'=x2=y2',],
[b'!re', b'=x2=y2', b'.tag=2'],
[b'!done', b'.tag=1'],
[b'!done']]
[b'!done', b'.tag=2']]
communicator = api_communicator.ApiCommunicator(base)
promise = communicator.call_async('/interface/', 'print')
response2 = communicator.call('/interface/', 'print')
promise = communicator.call('/interface/', 'print')
response2 = communicator.call('/interface/', 'print').get()
response1 = promise.get()
self.assertEqual(response1, [{'x1': 'y1'}])
self.assertEqual(response2, [{'x2': 'y2'}])
@ -50,55 +43,57 @@ class TestResponseSentence(TestCase):
base.receive_sentence.side_effect = [[b'!re', b'=x=y', b'.tag=1'],
[b'!done', b'.tag=1']]
communicator = api_communicator.ApiCommunicator(base)
promise = communicator.call_async('/file/', 'print', binary=True)
promise = communicator.call('/file/', 'print', binary=True)
response = promise.get()
self.assertEqual(response, [{'x': b'y'}])
def test_error_call(self):
base = mock.Mock()
base.receive_sentence.side_effect = [[b'!trap', b'=message=y'],
[b'!done']]
base.receive_sentence.side_effect = [[b'!trap', b'=message=y',
b'.tag=1'],
[b'!done', b'.tag=1']]
communicator = api_communicator.ApiCommunicator(base)
promise = communicator.call('/file/', 'print')
self.assertRaises(exceptions.RouterOsApiCommunicationError,
communicator.call, '/file/', 'print')
promise.get)
def test_query_call(self):
base = mock.Mock()
base.receive_sentence.return_value = [b'!done']
base.receive_sentence.return_value = [b'!done', b'.tag=1']
communicator = api_communicator.ApiCommunicator(base)
communicator.call('/interface/', 'print', queries={'x': 'y'})
communicator.call('/interface/', 'print', queries={'x': 'y'}).get()
base.send_sentence.assert_called_once_with(
[b'/interface/print', b'?x=y'])
[b'/interface/print', b'?x=y', b'.tag=1'])
def test_set_call(self):
base = mock.Mock()
base.receive_sentence.return_value = [b'!done']
base.receive_sentence.return_value = [b'!done', b'.tag=1']
communicator = api_communicator.ApiCommunicator(base)
communicator.call('/interface/', 'set', {'x': 'y'})
base.send_sentence.assert_called_once_with(
[b'/interface/set', b'=x=y'])
[b'/interface/set', b'=x=y', b'.tag=1'])
def test_async_error_raises_when_synchronizing(self):
base = mock.Mock()
base.receive_sentence.side_effect = [
[b'!trap', b'=message=m', b'.tag=1'],
[b'!done', b'.tag=1'],
[b'!done']]
[b'!done', b'.tag=2'],
[b'!done', b'.tag=1']]
communicator = api_communicator.ApiCommunicator(base)
promise = communicator.call_async('/interface/', 'print')
communicator.call('/interface/', 'print')
promise = communicator.call('/interface/', 'print')
communicator.call('/interface/', 'print').get()
self.assertRaises(exceptions.RouterOsApiCommunicationError,
promise.get)
def test_mixed_binary_and_ascii_calls(self):
base = mock.Mock()
base.receive_sentence.side_effect = [[b'!re', b'=x1=y1', b'.tag=1'],
[b'!re', b'=x2=y2',],
[b'!re', b'=x2=y2', b'.tag=2'],
[b'!done', b'.tag=1'],
[b'!done']]
[b'!done', b'.tag=2']]
communicator = api_communicator.ApiCommunicator(base)
promise = communicator.call_async('/interface/', 'print', binary=True)
response2 = communicator.call('/interface/', 'print')
promise = communicator.call('/interface/', 'print', binary=True)
response2 = communicator.call('/interface/', 'print').get()
response1 = promise.get()
self.assertEqual(response1, [{'x1': b'y1'}])
self.assertEqual(response2, [{'x2': 'y2'}])