Skip to content

Commit

Permalink
Fix formatting and linting
Browse files Browse the repository at this point in the history
  • Loading branch information
luqasz committed Jul 15, 2024
1 parent 0408bb5 commit 990cc32
Show file tree
Hide file tree
Showing 5 changed files with 25 additions and 306 deletions.
24 changes: 12 additions & 12 deletions librouteros/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,13 +17,13 @@
from librouteros.api import Api

DEFAULTS = {
'timeout': 10,
'port': 8728,
'saddr': '',
'subclass': Api,
'encoding': 'ASCII',
'ssl_wrapper': lambda sock: sock,
'login_method': plain,
"timeout": 10,
"port": 8728,
"saddr": "",
"subclass": Api,
"encoding": "ASCII",
"ssl_wrapper": lambda sock: sock,
"login_method": plain,
}


Expand All @@ -44,18 +44,18 @@ def connect(host: str, username: str, password: str, **kwargs) -> typing.Type[Ap
"""
arguments = ChainMap(kwargs, DEFAULTS)
transport = create_transport(host, **arguments)
protocol = ApiProtocol(transport=transport, encoding=arguments['encoding'])
api = arguments['subclass'](protocol=protocol)
protocol = ApiProtocol(transport=transport, encoding=arguments["encoding"])
api = arguments["subclass"](protocol=protocol)

try:
arguments['login_method'](api=api, username=username, password=password)
arguments["login_method"](api=api, username=username, password=password)
return api
except (ConnectionClosed, FatalError):
transport.close()
raise


def create_transport(host: str, **kwargs) -> SocketTransport:
sock = create_connection((host, kwargs['port']), kwargs['timeout'], (kwargs['saddr'], 0))
sock = kwargs['ssl_wrapper'](sock)
sock = create_connection((host, kwargs["port"]), kwargs["timeout"], (kwargs["saddr"], 0))
sock = kwargs["ssl_wrapper"](sock)
return SocketTransport(sock=sock)
11 changes: 7 additions & 4 deletions librouteros/protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,17 +26,17 @@ def parse_word(word: str) -> typing.Tuple[str, typing.Any]:
mapping = {"yes": True, "true": True, "no": False, "false": False}
_, key, value = word.split("=", 2)
try:
value = int(value) # type: ignore
value = int(value) # type: ignore
except ValueError:
value = mapping.get(value, value) # type: ignore
value = mapping.get(value, value) # type: ignore
return (key, value)


def cast_to_api(value: typing.Any) -> str:
"""Cast python equivalent to API."""
mapping = {True: "yes", False: "no"}
# this is necesary because 1 == True, 0 == False
if type(value) == int: # noqa: E721
if type(value) == int: # noqa: E721
return str(value)
return mapping.get(value, str(value))

Expand All @@ -50,6 +50,7 @@ def compose_word(key: str, value: typing.Any) -> str:


class Encoder:

def encodeSentence(self, *words: str) -> bytes:
"""
Encode given sentence in API format.
Expand All @@ -70,7 +71,7 @@ def encodeWord(self, word: str) -> bytes:
:returns: Encoded word.
"""
# pylint: disable=no-member
encoded_word = word.encode(encoding=self.encoding, errors="strict") # type: ignore
encoded_word = word.encode(encoding=self.encoding, errors="strict") # type: ignore
return Encoder.encodeLength(len(encoded_word)) + encoded_word

@staticmethod
Expand Down Expand Up @@ -100,6 +101,7 @@ def encodeLength(length: int) -> bytes:


class Decoder:

@staticmethod
def determineLength(length: bytes) -> int:
"""
Expand Down Expand Up @@ -154,6 +156,7 @@ def decodeLength(length: bytes) -> int:


class ApiProtocol(Encoder, Decoder):

def __init__(self, transport: SocketTransport, encoding: str):
self.transport = transport
self.encoding = encoding
Expand Down
Loading

0 comments on commit 990cc32

Please sign in to comment.