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

Remove support for ASCII resources.

Browse source
This commit is contained in:
Jakub Skiepko 2014-08-21 13:32:04 +02:00
parent 9890e22cd9
commit cd5665813d
3 changed files with 35 additions and 65 deletions
Download patch file Download diff file Expand all files Collapse all files

16
routeros_api/api.py
View file

@ -34,20 +34,19 @@ class RouterOsApi(object):
'login', {'name': login.encode(), 'response': hashed})
def get_resource(self, path):
return RouterOsResource(self.communicator, path)
raise NotImplementedError
def get_binary_resource(self, path):
return RouterOsResource(self.communicator, path, binary=True)
return RouterOsBinaryResource(self.communicator, path)
def close(self):
self.socket.close()
class RouterOsResource(object):
def __init__(self, communicator, path, binary=False):
class RouterOsBinaryResource(object):
def __init__(self, communicator, path):
self.communicator = communicator
self.path = clean_path(path)
self.binary = binary
def get(self, **kwargs):
return self.call('print', {}, kwargs)
@ -83,19 +82,18 @@ class RouterOsResource(object):
additional_queries=(), include_done=False):
return self.communicator.call(
self.path, command, arguments=arguments, queries=queries,
additional_queries=additional_queries, binary=self.binary,
additional_queries=additional_queries,
include_done=include_done).get()
def call_async(self, command, arguments=None, queries=None,
additional_queries=(), include_done=False):
return self.communicator.call(
self.path, command, arguments=arguments, queries=queries,
additional_queries=additional_queries, binary=self.binary,
additional_queries=additional_queries,
include_done=include_done)
def __repr__(self):
return 'RouterOsResource({path}, {binary})'.format(path=self.path,
binary=self.binary)
return type(self).__name__ + '({path})'.format(path=self.path)
def clean_path(path):

50
routeros_api/api_communicator/encoding_decorator.py
View file

@ -3,45 +3,39 @@ class EncodingApiCommunicator(object):
self.inner = inner
def call(self, path, command, arguments=None, queries=None,
additional_queries=(), include_done=False, binary=False):
queries = queries or {}
additional_queries=(), include_done=False):
path = path.encode()
command = command.encode()
arguments = self.encode_dictionary(arguments or {}, binary)
queries = self.encode_dictionary(queries or {}, binary)
arguments = self.transform_dictionary(arguments or {})
queries = self.transform_dictionary(queries or {})
promise = self.inner.call(
path, command, arguments, queries, additional_queries,
include_done)
return EncodedPromiseDecorator(promise, binary)
return self.decorate_promise(promise)
def transform_dictionary(self, dictionary):
return dict(self.transform_item(item) for item in dictionary.items())
def transform_item(self, item):
key, value = item
return (key.encode(), value)
def decorate_promise(self, promise):
return EncodedPromiseDecorator(promise)
def encode_dictionary(self, dictionary, binary_values):
encoded_arguments = {}
for key, value in dictionary.items():
if binary_values:
encoded_arguments[key.encode()] = value
else:
encoded_arguments[key.encode()] = value.encode()
return encoded_arguments
class EncodedPromiseDecorator(object):
def __init__(self, inner, binary_values):
def __init__(self, inner):
self.inner = inner
self.binary_values = binary_values
def get(self):
response = self.inner.get()
decoded_response = []
for row in response:
decoded_row = self.decode_dictionary(row)
decoded_response.append(decoded_row)
return decoded_response
return [self.transform_row(row) for row in response]
def decode_dictionary(self, dictionary):
decoded_dictionary = {}
for key, value in dictionary.items():
if self.binary_values:
decoded_dictionary[key.decode()] = value
else:
decoded_dictionary[key.decode()] = value.decode()
return decoded_dictionary
def transform_row(self, row):
return dict(self.transform_item(item) for item in row.items())
def transform_item(self, item):
key, value = item
return (key.decode(), value)

34
tests/test_api_communicator.py
View file

@ -8,14 +8,14 @@ except ImportError:
from routeros_api import exceptions
from routeros_api import api_communicator
class TestResponseSentence(TestCase):
class TestCommunicator(TestCase):
def test_login_call(self):
base = mock.Mock()
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).get()
self.assertEqual(response[0]['ret'], 'some-hex')
self.assertEqual(response[0]['ret'], b'some-hex')
def test_normal_call(self):
base = mock.Mock()
@ -23,7 +23,7 @@ class TestResponseSentence(TestCase):
[b'!done', b'.tag=1']]
communicator = api_communicator.ApiCommunicator(base)
response = communicator.call('/interface/', 'print').get()
self.assertEqual(response, [{'x': 'y'}])
self.assertEqual(response, [{'x': b'y'}])
def test_mixed_calls(self):
base = mock.Mock()
@ -35,17 +35,8 @@ class TestResponseSentence(TestCase):
promise = communicator.call('/interface/', 'print')
response2 = communicator.call('/interface/', 'print').get()
response1 = promise.get()
self.assertEqual(response1, [{'x1': 'y1'}])
self.assertEqual(response2, [{'x2': 'y2'}])
def test_binary_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('/file/', 'print', binary=True)
response = promise.get()
self.assertEqual(response, [{'x': b'y'}])
self.assertEqual(response1, [{'x1': b'y1'}])
self.assertEqual(response2, [{'x2': b'y2'}])
def test_error_call(self):
base = mock.Mock()
@ -69,7 +60,7 @@ class TestResponseSentence(TestCase):
base = mock.Mock()
base.receive_sentence.return_value = [b'!done', b'.tag=1']
communicator = api_communicator.ApiCommunicator(base)
communicator.call('/interface/', 'set', {'x': 'y'})
communicator.call('/interface/', 'set', {'x': b'y'})
base.send_sentence.assert_called_once_with(
[b'/interface/set', b'=x=y', b'.tag=1'])
@ -84,16 +75,3 @@ class TestResponseSentence(TestCase):
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'.tag=2'],
[b'!done', b'.tag=1'],
[b'!done', b'.tag=2']]
communicator = api_communicator.ApiCommunicator(base)
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'}])