Dynamic Configuration
In some situation, it can be useful to change the behaviour of the auxiliary in-use dynamically. For example, switching for a brand new channel or simply change an attribute value.
Thanks to the common auxiliary interface, users can easily change their auxiliary
configuration by simply stop it (call of delete_instance public method) ,access
it different public attributes, and then just restarts the auxiliary (call of
the public method delete_instance
)
Warning
if you are using the original auxiliary instance don’t forget to switch back to its initial configuration for the next test cases.
Find below a complete example where during the test, the current pcan connector is replaced by a simple CCLoopback:
import logging
import time
import pykiso
# as usual import your auxiliairies
from pykiso.auxiliaries import aux1, aux2, proxy_aux
from pykiso.lib.connectors.cc_raw_loopback import CCLoopback
@pykiso.define_test_parameters(
suite_id=2,
case_id=3,
aux_list=[aux1, aux2],
)
class TestCaseOverride(pykiso.BasicTest):
"""In this test case we will simply use 2 communication auxiliaries
bounded with a proxy one. The first communication auxiliary will be
used for sending and the other one for the reception
"""
def setUp(self):
"""If a fixture is not use just override it like below."""
logging.info(
f"--------------- SETUP: {self.test_suite_id}, {self.test_case_id} ---------------"
)
# start auxiliary one and two because I need it
aux1.start()
aux2.start()
# start the proxy auxiliary in order to open the connector
proxy_aux.start()
def test_run(self):
"""Just send some raw bytes using aux1 and log first 100
received messages using aux2.
"""
logging.info(
f"--------------- RUN: {self.test_suite_id}, {self.test_case_id} ---------------"
)
logging.info(f">> Send and receive message using the connected pcan <<")
logging.info(f"send 300 messages using aux1/aux2")
# just send some requests
self._send_messages(300)
# log the first 100 received messages, with the aux2
self._receive_message(100)
logging.info(
f">> Change proxy_aux channel dynamically and continue to send/receive <<"
)
logging.info(f"Stop current running auxiliaries")
self._stop_auxes()
# save current channel used by the proxy
self.pcan_channel = proxy_aux.channel
# change proxy attached channel to CCLoopback
proxy_aux.channel = CCLoopback()
logging.info(f"Restart all auxiliaries")
self._start_auxes()
logging.info(f">> Send and receive message using the connected CCLoopback <<")
logging.info(f"send 30 messages using aux1/aux2")
# just send some requests
self._send_messages(10)
# log the first 10 received messages, with the aux2
self._receive_message(10)
logging.info(
f">> Switch back to the pcan channel and continue to send/receive <<"
)
logging.info(f"Stop current running auxiliaries")
self._stop_auxes()
# switch back with pcan connector
proxy_aux.channel = self.pcan_channel
logging.info(f"Restart all auxiliaries")
self._start_auxes()
logging.info(f">> Send and receive message using the connected initial pcan <<")
logging.info(f"send 30 messages using aux1/aux2")
# just send some requests
self._send_messages(10)
# log the first 10 received messages, with the aux2
self._receive_message(10)
def _stop_auxes(self) -> None:
"""Stop all auxiliaries currently in use."""
# always stop the proxy auxiliary at the end
aux1.delete_instance()
aux2.delete_instance()
proxy_aux.delete_instance()
def _start_auxes(self) -> None:
"""Start all configured auxiliaries."""
# always start the proxy auxiliary at the end
aux1.create_instance()
aux2.create_instance()
proxy_aux.create_instance()
def _send_messages(self, nb_msg: int) -> None:
"""Send n messages a defined number of times.
:param nb_msg: number of messages pack to send
"""
for _ in range(nb_msg):
# send random messages using aux1
aux1.send_message(b"\x01\x02\x03")
aux2.send_message(b"\x04\x05\x06")
aux1.send_message(b"\x07\x08\x09")
def _receive_message(self, nb_msg: int) -> None:
"""Get messages from the reception queue.
:param nb_msg: number of messages to dequeue
"""
for _ in range(nb_msg):
logging.info(f"received message: {aux2.receive_message()}")
def tearDown(self):
"""If a fixture is not use just override it like below."""
logging.info(
f"--------------- TEARDOWN: {self.test_suite_id}, {self.test_case_id} ---------------"
)
Warning
this feature allows to change the complete auxiliary configuration, so depending on which parameters are changed the auxiliary execution could lead to unexpected behaviors.