(self, requesthandle, seqhdr, response, msgtype=ua.MessageType.SecureMessage)
| 41 | self._connection.set_policy_factories(policies) |
| 42 | |
| 43 | def send_response(self, requesthandle, seqhdr, response, msgtype=ua.MessageType.SecureMessage): |
| 44 | with self._socketlock: |
| 45 | response.ResponseHeader.RequestHandle = requesthandle |
| 46 | data = self._connection.message_to_binary( |
| 47 | struct_to_binary(response), message_type=msgtype, request_id=seqhdr.RequestId) |
| 48 | |
| 49 | self.socket.write(data) |
| 50 | |
| 51 | def open_secure_channel(self, algohdr, seqhdr, body): |
| 52 | request = struct_from_binary(ua.OpenSecureChannelRequest, body) |
no test coverage detected