Build a concreate transport layer based on pytestconfig. This class can be passed to new process and then build new transport layer (concreate transport layer cannot be passed to new process because of thread lock, so we have to pass it this way).
| 42 | |
| 43 | |
| 44 | class TransportFactory: |
| 45 | """ |
| 46 | Build a concreate transport layer based on pytestconfig. |
| 47 | |
| 48 | This class can be passed to new process and then build new transport layer (concreate transport layer cannot be |
| 49 | passed to new process because of thread lock, so we have to pass it this way). |
| 50 | """ |
| 51 | |
| 52 | def __init__(self, config: Config): |
| 53 | self.port = int(cast(int, config.getoption("port"))) |
| 54 | self.host = config.getoption("host") |
| 55 | self.serial = config.getoption("serial") |
| 56 | self.baud = int(cast(int, config.getoption("baud"))) |
| 57 | self.arbitrator = None |
| 58 | |
| 59 | def build_transport(self, is_server: bool, |
| 60 | create_arbitrator: bool = False) -> TCPTransport | SerialTransport | TransportArbitrator: |
| 61 | if self.serial: |
| 62 | transport = SerialTransport(self.serial, self.baud) |
| 63 | else: |
| 64 | transport = TCPTransport(self.host, self.port, is_server) |
| 65 | |
| 66 | if create_arbitrator: |
| 67 | transport = TransportArbitrator(transport, BasicCodec()) |
| 68 | |
| 69 | return transport |
| 70 | |
| 71 | |
| 72 | class CommonTestServiceHandler(test_unit_test_common.interface.ICommon): |
no outgoing calls