Skip to content

M
Mock Turtle
Commit 1d099591 authored by Federico Vaga's avatar Federico Vaga
Browse files
Options

test: check also statistics for HMQ 

Signed-off-by: Federico Vaga's avatarFederico Vaga <federico.vaga@cern.ch>
parent bd8a89e9
Hide whitespace changes
Inline Side-by-side
Showing with 25 additions and 3 deletions
tests/test_hmq.py
View file @ 1d099591
......@@ -56,6 +56,11 @@ class TestHmq(object):
def test_async_send(self, trtl_cpu, trtl_msg, trtl_binary_hmq_async_send):
"""It sends the test messages on all available HMQ.
It validates:
- messages are correctly sent
- statistic counter increase coherently
The test is successful when we read "OK" from the serial console
"""
trtl_cpu.load_application_file(trtl_binary_hmq_async_send)
......@@ -67,7 +72,10 @@ class TestHmq(object):
for hmq in trtl_cpu.hmq:
for msg in trtl_msg:
sb = hmq.get_stats()
hmq.send_msg(msg)
sa = hmq.get_stats()
assert sb["message_sent"] + 1 == sa["message_sent"]
time.sleep(0.5)
......@@ -75,15 +83,26 @@ class TestHmq(object):
assert self.confirm == ser.read(ser.in_waiting).decode()
def test_async_recv(self, trtl_cpu, trtl_binary_hmq_async_recv):
"""It receivess messages on all available HMQ.
The test is successful when we recevie all the expected messages
"""It receives messages on all available HMQ.
It validates:
- messages are correctly received
- statistic counter increase coherently
The test is succesful if the received message content is the expected
one
"""
sb = {}
for hmq in trtl_cpu.hmq:
sb[hmq] = hmq.get_stats()
trtl_cpu.load_application_file(trtl_binary_hmq_async_recv)
trtl_cpu.enable()
time.sleep(0.1)
for hmq in trtl_cpu.hmq:
for n in range(trtl_cpu.trtl_dev.rom.hmq[trtl_cpu.idx_cpu][hmq.idx_hmq].entries):
tot = trtl_cpu.trtl_dev.rom.hmq[trtl_cpu.idx_cpu][hmq.idx_hmq].entries
for n in range(tot):
msg = hmq.recv_msg()
payload = list(msg.payload)
for i, val in enumerate(payload):
......@@ -91,6 +110,9 @@ class TestHmq(object):
break
assert i == val
sa = hmq.get_stats()
assert sb[hmq]["message_received"] + tot == sa["message_received"]
def test_sync(self, trtl_cpu, trtl_msg, trtl_binary_hmq_sync):
"""It sends the test messages on all available HMQ.
The test is successful when what we read back is the same we sent
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment