Best Python code snippet using avocado_python
_client_tests.py
Source:_client_tests.py
...67 "Client random was not recovered after deserialization:\n\t" \68 + f"original: {original_key_bytes}\n\t" \69 + f"received: {received_key_bytes}"7071 def test_disconnect():72 assert False, "Should not have closed"7374 def test_upstream_private_packet(*args):75 assert False, "Should not have sent private packet to client"7677 def test_upstream_broadcast(*args):78 assert False, "Should not have sent broadcast packet to client"7980 handshake.on_packet_send.add_handler(test_packet)81 handshake.on_disconnect.add_handler(test_disconnect)8283 handshake.on_private_packet_received.add_handler(test_upstream_private_packet)84 handshake.on_broadcast_received.add_handler(test_upstream_broadcast)8586 handshake.start_connection()8788 assert call_flag, "Client Authenticate Not Sent"8990def test_receive_server_hello():91 ca_private_key = rsa.generate_private_key(92 65537,93 2048)94 ca_public_key = ca_private_key.public_key()9596 username = "bob"97 signature = ca_private_key.sign(98 (username + '\n').encode('utf-8'),99 padding.PKCS1v15(),100 hashes.SHA256())101102 handshake = ClientStateMachine()103 handshake._generate_parameters()104 handshake._main_state = "WAITING_FOR_HELLO"105106 #handshake will not sent client_authenticate otherwise107 handshake.set_username(username, signature)108109 server_random = os.urandom(packets.handshake_module.RANDOM_LENGTH)110 server_ephemeral_private_key = X25519PrivateKey.generate()111 server_ephemeral_public_key = server_ephemeral_private_key.public_key()112 server_persistent_private_key = rsa.generate_private_key(113 65537,114 2048)115 server_persistent_public_key = server_persistent_private_key.public_key()116117 premaster_key = server_ephemeral_private_key.exchange(118 handshake.client_ephemeral_public_key)119120 master_key = (HKDF(121 algorithm = hashes.SHA256(),122 length = handshake_module.MASTER_KEY_LENGTH,123 salt = handshake.client_random + server_random,124 info = None)125 .derive(premaster_key))126127 client_send_key_bytes = master_key[0:packets.crypt.CHACHA20_KEY_LENGTH]128 server_send_key_bytes = master_key[packets.crypt.CHACHA20_KEY_LENGTH:]129130 server_hello = ServerHello(131 server_ephemeral_public_key,132 server_persistent_public_key,133 server_random,134 handshake.client_random)135136 serialized_server_hello = server_hello.sign_and_serialize(137 server_persistent_private_key)138139 call_flag = False140141 def test_authenticate(message_tag,serialized_message):142 nonlocal call_flag143 call_flag = True144145 assert message_tag == PACKETS["CLIENT_AUTHENTICATE"],\146 f"Expected CLIENT_AUTHENTICATE, found {INV_PACKETS.get(type,'INVALID')}"147148 #Ensure values from sender_hello are incorporated149 assert handshake.server_random == server_random150 assert (151 _format_x25519_public(152 handshake.server_ephemeral_public_key)153 == _format_x25519_public(154 server_ephemeral_public_key))155156 #Ensure keys are derived correctly157 assert handshake._client_send_key_bytes == client_send_key_bytes,\158 (159 f"Expected: {client_send_key_bytes}\n"160 +f"Actual: {handshake._client_send_key_bytes}"161 )162 assert handshake._server_send_key_bytes == server_send_key_bytes163164 #ensure CLIENT_AUTHENTICATE deserializes correctly165166 client_authenticate = ClientAuthenticate.decrypt_deserialize_and_validate(167 serialized_message,168 ChaCha20Poly1305(client_send_key_bytes),169 ca_public_key)170171 assert client_authenticate.username == username172 assert client_authenticate.signature == signature173174 def test_disconnect():175 assert False, "Should not have closed"176177 def test_upstream_private_packet(*args):178 assert False, "Should not have sent private packet to client"179180 def test_upstream_broadcast(*args):181 assert False, "Should not have sent broadcast packet to client"182183 handshake.on_packet_send.add_handler(test_authenticate)184 handshake.on_disconnect.add_handler(test_disconnect)185186 handshake.on_private_packet_received.add_handler(test_upstream_private_packet)187 handshake.on_broadcast_received.add_handler(test_upstream_broadcast)188189 handshake.handle_packet(190 PACKETS["SERVER_HELLO"],191 serialized_server_hello)192193 assert call_flag, "Client Authenticate Not Sent"194195def test_receive_server_envelope():196 server_send_key = ChaCha20Poly1305(ChaCha20Poly1305.generate_key())197198 handshake = ClientStateMachine()199 handshake.server_send_key = server_send_key200 handshake._main_state = "RECEIVE_ONLY"201202 expected_sender = "alice"203 expected_message = "hello".encode('utf-8')204205 server_envelope = ServerEnvelope(206 expected_sender,207 expected_message)208209 serialized_message = server_envelope.encrypt_and_serialize(210 server_send_key)211212 call_flag = False213214 def test_client(actual_sender,actual_message):215 nonlocal call_flag216 call_flag = True217218 assert expected_sender == actual_sender,\219 (220 f"expected: {expected_sender}\n"221 +f"actual: {actual_sender}"222 )223 assert expected_message == actual_message,\224 (225 f"expected: {expected_message}\n"226 +f"actual: {actual_message}"227 )228229 def test_disconnect():230 assert False, "Should not have closed"231232 def test_upstream_broadcast(*args):233 assert False, "Should not have sent broadcast packet to client"234235 handshake.on_private_packet_received.add_handler(236 test_client)237 handshake.on_disconnect.add_handler(238 test_disconnect)239240 handshake.on_broadcast_received.add_handler(241 test_upstream_broadcast)242243 handshake.handle_packet(244 PACKETS["SERVER_ENVELOPE"],245 serialized_message)246247 assert call_flag248249def test_send_private():250 client_send_key = ChaCha20Poly1305(ChaCha20Poly1305.generate_key())251252 handshake = ClientStateMachine()253 handshake.client_send_key = client_send_key254 handshake._main_state = "READY"255256 expected_recipient = "alice"257 expected_packet = "hello".encode('utf-8')258259 call_flag = False260261 def test_server_envelope(message_tag,serialized_message):262 nonlocal call_flag263 call_flag = True264265 assert message_tag == PACKETS["SERVER_ENVELOPE"],\266 (267 "Expected: SERVER_ENVELOPE\n"268 +f"actual: {INV_PACKETS.get(message_tag,'INVALID')}"269 )270271 server_envelope = ServerEnvelope.decrypt_and_deserialize(272 serialized_message,273 handshake.client_send_key274 )275276 actual_recipient = server_envelope.username277 actual_packet = server_envelope.private_packet278279 assert expected_recipient == actual_recipient,\280 (281 f"expected: {expected_recipient}\n"282 +f"actual: {actual_recipient}"283 )284 assert expected_packet == actual_packet,\285 (286 f"expected: {expected_packet}\n"287 +f"actual: {actual_packet}"288 )289290 def test_disconnect():291 assert False, "Should not have closed"292293 def test_private_packet(*args):294 assert False, "Should not have sent private packet to client"295296 def test_broadcast(*args):297 assert False, "Should not have sent broadcast packet to client"298299 handshake.on_packet_send.add_handler(300 test_server_envelope)301 handshake.on_disconnect.add_handler(302 test_disconnect)303304 handshake.on_private_packet_received.add_handler(305 test_private_packet)306 handshake.on_broadcast_received.add_handler(307 test_broadcast)308309 handshake.send_private_packet(310 expected_recipient,311 expected_packet)312313 assert call_flag314315def test_client_handshake_receive_server_broadcast():316 server_send_key = ChaCha20Poly1305(ChaCha20Poly1305.generate_key())317318 handshake = ClientStateMachine()319 handshake.server_send_key = server_send_key320 handshake._main_state = "RECEIVE_ONLY"321322 expected_sender = "alice"323 expected_message = "hello"324325 broadcast = DistributeBroadcast(326 expected_sender,327 expected_message)328329 serialized_message = broadcast.encrypt_and_serialize(330 server_send_key)331332 call_flag = False333334 def test_broadcast(actual_sender,actual_message):335 nonlocal call_flag336 call_flag = True337338 assert expected_sender == actual_sender,\339 (340 f"expected: {expected_sender}\n"341 +f"actual: {actual_sender}"342 )343 assert expected_message == actual_message,\344 (345 f"expected: {expected_message}\n"346 +f"actual: {actual_message}"347 )348349 def test_disconnect():350 assert False, "Should not have closed"351352 def test_private_packet(*args):353 assert False, "Should not have sent private packet to client"354355 handshake.on_packet_send.add_handler(356 test_broadcast)357 handshake.on_disconnect.add_handler(358 test_disconnect)359360 handshake.on_broadcast_received.add_handler(361 test_broadcast)362363 handshake.on_private_packet_received.add_handler(364 test_private_packet)365366 handshake.handle_packet(367 PACKETS["DISTRIBUTE_SERVER_BROADCAST"],368 serialized_message)369370 assert call_flag371372def test_client_handshake_send_broadcast():373 server_send_key = ChaCha20Poly1305(ChaCha20Poly1305.generate_key())374375 handshake = ClientStateMachine()376 handshake.client_send_key = server_send_key377 handshake._main_state = "READY"378379 expected_message = "hello"380381 call_flag = False382383 def test_client(message_tag,serialized_message):384 nonlocal call_flag385 call_flag = True386387 assert message_tag == PACKETS["SEND_SERVER_BROADCAST"],\388 (389 "Expected: DISTRIBUTE_SERVER_BROADCAST\n"390 +f"actual: {INV_PACKETS.get(message_tag,'INVALID')}"391 )392393 server_broadcast = SendBroadcast.decrypt_and_deserialize(394 serialized_message,395 handshake.client_send_key396 )397398 actual_message = server_broadcast.message399400 assert expected_message == actual_message,\401 (402 f"expected: {expected_message}\n"403 +f"actual: {actual_message}"404 )405406 def test_disconnect():407 assert False, "Should not have closed"408409 def test_upstream_private_packet(*args):410 assert False, "Should not have sent private packet to client"411412 def test_upstream_broadcast(*args):413 assert False, "Should not have sent broadcast packet to client"414415 handshake.on_packet_send.add_handler(416 test_client)417 handshake.on_disconnect.add_handler(418 test_disconnect)419420 handshake.on_broadcast_received.add_handler(
...
server_disconnect.py
Source:server_disconnect.py
...42 cfg.name, cfg.sid, cfg.user, cfg.japd,43 os_type=getattr(machine, cfg.serv_os)(), host=cfg.host,44 port=cfg.port, defaults_file=cfg.cfg_file)45 self.svr.connect()46 def test_disconnect(self):47 """Function: test_disconnect48 Description: Test disconnect method.49 Arguments:50 """51 self.svr.disconnect()52 self.assertFalse(self.svr.conn.is_connected())53if __name__ == "__main__":...
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!