Skip to content
Snippets Groups Projects
Commit 11f27101 authored by KERDREUX Jerome's avatar KERDREUX Jerome
Browse files

No more type hints error

Ouff ..
parent 5ccfdb4c
Branches
No related tags found
1 merge request!1First try of type hints
......@@ -174,13 +174,15 @@ class TestMessageFactory(unittest.TestCase):
mf = test_factory()
# too young
msg = test_message()
msg.timestamp[0] = msg.timestamp[0] + 60*5
target = (msg.timestamp[0] + 60*5, msg.timestamp[1])
msg.timestamp = target
data = mf.encode_msg(msg)
with self.assertRaises(MessageParserError):
mf.decode_msg(data)
# too old
msg.timestamp = messages.build_timestamp()
msg.timestamp[0] = msg.timestamp[0] - 60*5
target = (msg.timestamp[0] - 60*5, msg.timestamp[1])
msg.timestamp = target
data = mf.encode_msg(msg)
with self.assertRaises(MessageParserError):
mf.decode_msg(data)
......
......@@ -65,11 +65,11 @@ class Message(object):
def __init__(self):
self.version = config.STACK_VERSION # message API version
self.__targets = [] # target property
self.timestamp = None # message timestamp
self.source = None # message source
self.dev_type = None # message dev_type
self.msg_type = None # message type
self.action = None # message action
self.timestamp: tuple = () # message timestamp
self.source: Optional[UUID] = None # message source
self.dev_type: Optional[str] = None # message dev_type
self.msg_type: Optional[MessageType] = None # message type
self.action: Optional[str] = None # message action
self.body = {} # message body
@property
......@@ -105,7 +105,7 @@ class Message(object):
tmp = tmp + "- %-12s %s\n" % (k, v)
# tmp = tmp.strip()
r.append(['body', tmp])
print(tabulate(r, headers=['Fied', 'Value'], tablefmt='psql'))
print(tabulate(r, headers=['Field', 'Value'], tablefmt='psql'))
def __repr__(self) -> str:
return f"<xaal.Message {id(self):x} {self.source} {self.dev_type} {self.msg_type} {self.action}>"
......@@ -161,7 +161,7 @@ class MessageFactory(object):
# key encode / decode message built from passphrase
self.cipher_key = cipher_key
def encode_msg(self, msg: Message):
def encode_msg(self, msg: Message) -> bytes:
"""Apply security layer and return encode MSG in CBOR
:param msg: xAAL msg instance
:type msg: Message
......@@ -179,6 +179,10 @@ class MessageFactory(object):
# Format payload & ciphering
buf = []
if not msg.source:
raise MessageError("No source address in message")
if not msg.msg_type:
raise MessageError("No msg_type in message")
buf.append(msg.source.bytes)
buf.append(msg.dev_type)
buf.append(msg.msg_type.value)
......@@ -218,7 +222,7 @@ class MessageFactory(object):
msg_time = data_rx[1]
targets = cbor.loads(data_rx[3])
msg.targets = [UUID(bytes=t) for t in targets]
msg.timestamp = [data_rx[1], data_rx[2]]
msg.timestamp = (data_rx[1], data_rx[2])
except IndexError:
raise MessageParserError("Bad Message, wrong fields")
......@@ -336,7 +340,7 @@ def build_nonce(data: tuple) -> bytes:
return nonce
def build_timestamp() -> list:
def build_timestamp() -> tuple:
"""Return array [seconds since epoch, microseconds since last seconds] Time = UTC+0000"""
epoch = datetime.datetime.fromtimestamp(0, datetime.UTC)
timestamp = datetime.datetime.now(datetime.UTC) - epoch
......@@ -346,6 +350,6 @@ def build_timestamp() -> list:
# for better performance, I choose to use this trick to fix the change in size for Py3.
# only test once.
if sys.version_info.major == 2:
_packtimestamp = lambda t1, t2: [long(t1), int(t2)]
_packtimestamp = lambda t1, t2: (long(t1), int(t2))
else:
_packtimestamp = lambda t1, t2: [int(t1), int(t2)]
_packtimestamp = lambda t1, t2: (int(t1), int(t2))
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please to comment