How to use test_disconnect method in avocado

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__":...

Automation Testing Tutorials

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.