Make an auxiliary copy
Warning
feature under development and testing. The copycat ability is not available for the multiprocessing/robot auxiliaries
In some situation, it is useful to create a copy of the in-use auxiliary. This very special usecase is covered by methods create_copy and destroy_copy.
Those methods are part of the AuxiliaryCommon interface. Due to this fact only threaded and multiprocessing auxiliaries are “copy capable”.
By invoking the create_copy method, the “original” auxiliary will be automatically suspended and a brand new one will be created. The only difference between both is: the configuration.
The introduction of this copy capability allows the user to create a copycat with a different configuration in order to have for specific test-cases, temporarily, the auxiliary to be defined differently than usually. at a very specific testing time (during only one test case, setup, teardown…). Thanks to the usage of python set, users only have to specify the parameters they want to change. This means, create_copy method will instanciate a brand new auxiliary based on the input yaml configuration file and the user’s desired changes.
Warning
create_copy only allows named parameters so don’t forget to name it!!
To get the “original” auxiliary back, users have only to invoke the method destroy_copy . This simply stops the current copied auxiliary and resumes the original one.
Warning
this feature allows to change the complete auxiliary configuration, so depending on which parameters are changed the copied auxiliary could lead to unexpected behavior.
Please find below a complete example, where a original communication auxiliary is copied and the copy is copied at it’s turn.
import logging
import pykiso
# as usual import your auxiliairies
from pykiso.auxiliaries import com_aux
from pykiso.lib.connectors.cc_raw_loopback import CCLoopback
@pykiso.define_test_parameters(
suite_id=3,
case_id=4,
aux_list=[com_aux],
)
class TestCaseOverride(pykiso.BasicTest):
"""In this test case we will simply use the two available methods
for a communication auxiliary (send_message and receive_message)
"""
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} ---------------"
)
# just create a communication auxiliary with a bran new channel
# in order to send odd bytes. Always use named arguments!
com_aux.start()
self.com_aux_odd = com_aux.create_copy(com=CCLoopback())
# just create a communication auxiliary based on the given
# parameters present in the yaml config (com_aux.yaml) to send
# even bytes
self.com_aux_even = self.com_aux_odd.create_copy()
# start com_aux_even because original configuration has
# auto_start flag set to False
self.com_aux_even.start()
def test_run(self):
"""Thanks to the usage of dev cc_raw_loopback, let's try to send
a message and receive it.
"""
logging.info(
f"--------------- RUN: {self.test_suite_id}, {self.test_case_id} ---------------"
)
# first send the even bytes and stop the copy in order to
# automatically resume the original one (self.com_aux_odd)
self.com_aux_even.send_message(b"\x02\x04\x06")
response = self.com_aux_even.receive_message()
self.assertEqual(response, b"\x02\x04\x06")
# destroy com_aux_even and start working with com_aux_odd aux
# copy
self.com_aux_odd.destroy_copy()
# Then send the odd bytes and stop the copy in order to
# automatically resume the original one (com_aux)
self.com_aux_odd.send_message(b"\x01\x03\x05")
response = self.com_aux_odd.receive_message()
self.assertEqual(response, b"\x01\x03\x05")
# destroy com_aux_odd and start working with com_aux
com_aux.destroy_copy()
# Just use the original communication auxiliary to send and
# receive some bytes
com_aux.send_message(b"\x01\x02\x03\x04\x05\x06")
response = com_aux.receive_message()
logging.info(f"received message: {response}")
self.assertEqual(response, b"\x01\x02\x03\x04\x05\x06")
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} ---------------"
)
In case of a proxy setup please find below a concrete example :
import logging
import time
import pykiso
# as usual import your auxiliairies
from pykiso.auxiliaries import aux1, aux2, proxy_aux
@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."""
# 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 10 received
messages using aux2.
"""
logging.info(
f"--------------- RUN: {self.test_suite_id}, {self.test_case_id} ---------------"
)
# send random messages using aux1
aux1.send_message(b"\x01\x02\x03")
aux1.send_message(b"\x04\x05\x06")
aux1.send_message(b"\x07\x08\x09")
# Just create a copy of aux one and two
self.aux1_copy = aux1.create_copy()
self.aux2_copy = aux2.create_copy()
# create a copy of proxy_aux with the brand new aux1 and aux2
# copy
self.proxy_copy = proxy_aux.create_copy(
aux_list=[self.aux1_copy, self.aux2_copy]
)
# all original auxiliaries have the auto_start falg to False,
# so copy too. Just start them, always finish with the proxy
# auxiliary
self.aux1_copy.start()
self.aux2_copy.start()
self.proxy_copy.start()
# send some random messages from aux1 and aux2 copies
self.aux1_copy.send_message(b"\x10\x11\x12")
self.aux1_copy.send_message(b"\x13\x14\x15")
self.aux2_copy.send_message(b"\x16\x17\x18")
# just wait a little bit to be sure everything is sent
time.sleep(1)
# destroy the all the copies
proxy_aux.destroy_copy()
aux2.destroy_copy()
aux1.destroy_copy()
# log the first 10 received messages, with the "original" aux2
for _ in range(10):
logging.info(f"received message: {aux2.receive_message()}")
def tearDown(self):
"""If a fixture is not use just override it like below."""
pass