KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _rendezvous_test method in autotest

Best Python code snippet using autotest_python

base_barrier_unittest.py

Source:base_barrier_unittest.py

...69            sender.close()70            receiver.close()71    def test_rendezvous_basic(self):72        # Basic rendezvous testing73        self._rendezvous_test(60, port=11920)74    def test_rendezvous_timeout(self):75        # The rendezvous should time out here and throw a76        # BarrierError since we are specifying a timeout of 077        self.assertRaises(error.BarrierError,78                          self._rendezvous_test, 0, port=11921)79    def test_rendezvous_abort_ok(self):80        # Test with abort flag set to not abort.81        self._rendezvous_test(60, port=11920,82                              test_abort=True, abort=False)83    def test_rendezvous_abort(self):84        # The rendezvous should abort here and throw a85        # BarrierError since we are asking to abort86        self.assertRaises(error.BarrierError,87                          self._rendezvous_test, 0, port=11921,88                          test_abort=True, abort=True)89    def test_rendezvous_servers_basic(self):90        # The rendezvous should time out here and throw a91        # BarrierError since we are specifying a timeout of 092        self._rendezvous_test(60, port=11921,93                              rendezvous_servers=True)94    def test_rendezvous_servers_timeout(self):95        # The rendezvous should time out here and throw a96        # BarrierError since we are specifying a timeout of 097        self.assertRaises(error.BarrierError,98                          self._rendezvous_test, 0, port=11922,99                          rendezvous_servers=True)100    def test_rendezvous_servers_abort_ok(self):101        # Test with abort flag set to not abort.102        self._rendezvous_test(60, port=11920, rendezvous_servers=True,103                              test_abort=True, abort=False)104    def test_rendezvous_servers_abort(self):105        # The rendezvous should abort here and throw a106        # BarrierError since we are asking to abort107        self.assertRaises(error.BarrierError,108                          self._rendezvous_test, 0, port=11922,109                          rendezvous_servers=True,110                          test_abort=True, abort=True)111    def _rendezvous_test(self, timeout, port=11922,112                         rendezvous_servers=False, test_abort=False,113                         abort=False, listen_server=None):114        if listen_server:115            port = None116        def _rdv(addr):117            b1 = barrier.barrier(addr, "test_meeting", timeout, port,118                                 listen_server=listen_server)119            if not rendezvous_servers:120                if test_abort:121                    b1.rendezvous('127.0.0.1#0', '127.0.0.1#1', abort=abort)122                else:123                    b1.rendezvous('127.0.0.1#0', '127.0.0.1#1')124            else:125                if test_abort:126                    b1.rendezvous_servers('127.0.0.1#0', '127.0.0.1#1',127                                          abort=abort)128                else:129                    b1.rendezvous_servers('127.0.0.1#0', '127.0.0.1#1')130        def _thread_rdv(addr):131            # We need to ignore the exception on one side.132            try:133                _rdv(addr)134            except error.BarrierError:135                pass136        client = threading.Thread(target=_thread_rdv,137                                  args=('127.0.0.1#0',))138        client.start()139        _rdv('127.0.0.1#1')140        client.join()141    def test_reusing_listen_server(self):142        """143        Test that reusing the same listen server object works.144        """145        server = barrier.listen_server()146        self._rendezvous_test(10, listen_server=server)147        self._rendezvous_test(10, listen_server=server)148        self._rendezvous_test(10, listen_server=server)149if __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.