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 include done flag.

Browse source
This commit is contained in:
Jakub Skiepko 2014-08-21 15:10:05 +02:00
parent cd5665813d
commit 9628654db4
5 changed files with 33 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

14
routeros_api/api.py
View file

@ -23,8 +23,8 @@ class RouterOsApi(object):
def login(self, login, password):
response = self.get_binary_resource('/').call(
'login', include_done=True)
token = binascii.unhexlify(response[0]['ret'])
'login')
token = binascii.unhexlify(response.done_message['ret'])
hasher = hashlib.md5()
hasher.update(b'\x00')
hasher.update(password.encode())
@ -79,18 +79,16 @@ class RouterOsBinaryResource(object):
return self.call_async('remove', kwargs)
def call(self, command, arguments=None, queries=None,
additional_queries=(), include_done=False):
additional_queries=()):
return self.communicator.call(
self.path, command, arguments=arguments, queries=queries,
additional_queries=additional_queries,
include_done=include_done).get()
additional_queries=additional_queries).get()
def call_async(self, command, arguments=None, queries=None,
additional_queries=(), include_done=False):
additional_queries=()):
return self.communicator.call(
self.path, command, arguments=arguments, queries=queries,
additional_queries=additional_queries,
include_done=include_done)
additional_queries=additional_queries)
def __repr__(self):
return type(self).__name__ + '({path})'.format(path=self.path)

33
routeros_api/api_communicator/base.py
View file

@ -10,13 +10,13 @@ class ApiCommunicatorBase(object):
self.response_buffor = {}
def send(self, path, command, arguments=None, queries=None,
additional_queries=(), include_done=False):
additional_queries=()):
tag = self._get_next_tag()
command = self.get_command(path, command, arguments, queries, tag=tag,
additional_queries=additional_queries)
self.send_command(command)
self.response_buffor[tag] = AsynchronousResponse(
command, include_done)
self.response_buffor[tag] = AsynchronousResponseCollector(
command)
return tag
def get_command(self, path, command, arguments=None, queries=None,
@ -57,11 +57,13 @@ class ApiCommunicatorBase(object):
raise exceptions.FatalRouterOsApiError(
"Unknown tag %s", response.tag)
asynchronous_response = self.response_buffor[response.tag]
if response.type in asynchronous_response.get_meaningfull_responses():
if response.type == b're':
attributes = response.attributes
asynchronous_response.attributes.append(attributes)
if response.type == b'done':
asynchronous_response.done = True
attributes = response.attributes
asynchronous_response.attributes.done_message = attributes
elif response.type == b'trap':
asynchronous_response.error = response.attributes[b'message']
elif response.type == b'fatal':
@ -78,17 +80,20 @@ class ApiCommunicatorBase(object):
return response
class AsynchronousResponse(object):
def __init__(self, command, include_done):
self.attributes = []
class AsynchronousResponseCollector(object):
def __init__(self, command):
self.attributes = AsynchronousResponse()
self.done = False
self.error = None
self.command = command
self.include_done = include_done
def get_meaningfull_responses(self):
if self.include_done:
save_responses = [b're', b'done']
else:
save_responses = [b're']
return save_responses
class AsynchronousResponse(list):
def __init__(self, *args, **kwargs):
super(AsynchronousResponse, self).__init__(*args, **kwargs)
self.done_message = {}
def map(self, function):
result = type(self)(function(item) for item in self)
result.done_message = function(self.done_message)
return result

8
routeros_api/api_communicator/encoding_decorator.py
View file

@ -3,14 +3,13 @@ class EncodingApiCommunicator(object):
self.inner = inner
def call(self, path, command, arguments=None, queries=None,
additional_queries=(), include_done=False):
additional_queries=()):
path = path.encode()
command = command.encode()
arguments = self.transform_dictionary(arguments or {})
queries = self.transform_dictionary(queries or {})
promise = self.inner.call(
path, command, arguments, queries, additional_queries,
include_done)
path, command, arguments, queries, additional_queries)
return self.decorate_promise(promise)
def transform_dictionary(self, dictionary):
@ -24,14 +23,13 @@ class EncodingApiCommunicator(object):
return EncodedPromiseDecorator(promise)
class EncodedPromiseDecorator(object):
def __init__(self, inner):
self.inner = inner
def get(self):
response = self.inner.get()
return [self.transform_row(row) for row in response]
return response.map(self.transform_row)
def transform_row(self, row):
return dict(self.transform_item(item) for item in row.items())

7
routeros_api/api_communicator/key_cleaner_decorator.py
View file

@ -3,17 +3,16 @@ class KeyCleanerApiCommunicator(object):
self.inner = inner
def send(self, path, command, arguments=None, queries=None,
additional_queries=(), include_done=False):
additional_queries=()):
encoded_arguments = encode_dictionary(arguments or {})
encoded_queries = encode_dictionary(queries or {})
return self.inner.send(
path, command, arguments=encoded_arguments,
queries=encoded_queries, additional_queries=additional_queries,
include_done=include_done)
queries=encoded_queries, additional_queries=additional_queries)
def receive(self, tag):
answers = self.inner.receive(tag)
return [decode_dictionary(answer) for answer in answers]
return answers.map(decode_dictionary)
def encode_dictionary(dictionary):

4
tests/test_api_communicator.py
View file

@ -14,8 +14,8 @@ class TestCommunicator(TestCase):
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'], b'some-hex')
response = communicator.call('/', 'login').get()
self.assertEqual(response.done_message['ret'], b'some-hex')
def test_normal_call(self):
base = mock.Mock()