Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use start_client method in autotest

Best Python code snippet using autotest_python

test_client_server.py

Source:test_client_server.py

...39        self.loop.run_forever()40    def start_server(self, **kwds):41        server = serve(handler, 'localhost', 8642, **kwds)42        self.server = self.loop.run_until_complete(server)43    def start_client(self, path='', **kwds):44        client = connect('ws://localhost:8642/' + path, **kwds)45        self.client = self.loop.run_until_complete(client)46    def stop_client(self):47        try:48            self.loop.run_until_complete(49                asyncio.wait_for(self.client.worker_task, timeout=1))50        except asyncio.TimeoutError:                # pragma: no cover51            self.fail("Client failed to stop")52    def stop_server(self):53        self.server.close()54        try:55            self.loop.run_until_complete(56                asyncio.wait_for(self.server.wait_closed(), timeout=1))57        except asyncio.TimeoutError:                # pragma: no cover58            self.fail("Server failed to stop")59    def test_basic(self):60        self.start_server()61        self.start_client()62        self.loop.run_until_complete(self.client.send("Hello!"))63        reply = self.loop.run_until_complete(self.client.recv())64        self.assertEqual(reply, "Hello!")65        self.stop_client()66        self.stop_server()67    def test_server_close_while_client_connected(self):68        self.start_server()69        self.start_client()70        self.stop_server()71    def test_explicit_event_loop(self):72        self.start_server(loop=self.loop)73        self.start_client(loop=self.loop)74        self.loop.run_until_complete(self.client.send("Hello!"))75        reply = self.loop.run_until_complete(self.client.recv())76        self.assertEqual(reply, "Hello!")77        self.stop_client()78        self.stop_server()79    def test_protocol_attributes(self):80        self.start_server()81        self.start_client('attributes')82        expected_attrs = ('localhost', 8642, self.secure)83        client_attrs = (self.client.host, self.client.port, self.client.secure)84        self.assertEqual(client_attrs, expected_attrs)85        server_attrs = self.loop.run_until_complete(self.client.recv())86        self.assertEqual(server_attrs, repr(expected_attrs))87        self.stop_client()88        self.stop_server()89    def test_protocol_headers(self):90        self.start_server()91        self.start_client('headers')92        client_req = self.client.request_headers93        client_resp = self.client.response_headers94        self.assertEqual(client_req['User-Agent'], USER_AGENT)95        self.assertEqual(client_resp['Server'], USER_AGENT)96        server_req = self.loop.run_until_complete(self.client.recv())97        server_resp = self.loop.run_until_complete(self.client.recv())98        self.assertEqual(server_req, str(client_req))99        self.assertEqual(server_resp, str(client_resp))100        self.stop_client()101        self.stop_server()102    def test_protocol_raw_headers(self):103        self.start_server()104        self.start_client('raw_headers')105        client_req = self.client.raw_request_headers106        client_resp = self.client.raw_response_headers107        self.assertEqual(dict(client_req)['User-Agent'], USER_AGENT)108        self.assertEqual(dict(client_resp)['Server'], USER_AGENT)109        server_req = self.loop.run_until_complete(self.client.recv())110        server_resp = self.loop.run_until_complete(self.client.recv())111        self.assertEqual(server_req, repr(client_req))112        self.assertEqual(server_resp, repr(client_resp))113        self.stop_client()114        self.stop_server()115    def test_protocol_custom_request_headers_dict(self):116        self.start_server()117        self.start_client('raw_headers', extra_headers={'X-Spam': 'Eggs'})118        req_headers = self.loop.run_until_complete(self.client.recv())119        self.loop.run_until_complete(self.client.recv())120        self.assertIn("('X-Spam', 'Eggs')", req_headers)121        self.stop_client()122        self.stop_server()123    def test_protocol_custom_request_headers_list(self):124        self.start_server()125        self.start_client('raw_headers', extra_headers=[('X-Spam', 'Eggs')])126        req_headers = self.loop.run_until_complete(self.client.recv())127        self.loop.run_until_complete(self.client.recv())128        self.assertIn("('X-Spam', 'Eggs')", req_headers)129        self.stop_client()130        self.stop_server()131    def test_protocol_custom_response_headers_callable_dict(self):132        self.start_server(extra_headers=lambda p, r: {'X-Spam': 'Eggs'})133        self.start_client('raw_headers')134        self.loop.run_until_complete(self.client.recv())135        resp_headers = self.loop.run_until_complete(self.client.recv())136        self.assertIn("('X-Spam', 'Eggs')", resp_headers)137        self.stop_client()138        self.stop_server()139    def test_protocol_custom_response_headers_callable_list(self):140        self.start_server(extra_headers=lambda p, r: [('X-Spam', 'Eggs')])141        self.start_client('raw_headers')142        self.loop.run_until_complete(self.client.recv())143        resp_headers = self.loop.run_until_complete(self.client.recv())144        self.assertIn("('X-Spam', 'Eggs')", resp_headers)145        self.stop_client()146        self.stop_server()147    def test_protocol_custom_response_headers_dict(self):148        self.start_server(extra_headers={'X-Spam': 'Eggs'})149        self.start_client('raw_headers')150        self.loop.run_until_complete(self.client.recv())151        resp_headers = self.loop.run_until_complete(self.client.recv())152        self.assertIn("('X-Spam', 'Eggs')", resp_headers)153        self.stop_client()154        self.stop_server()155    def test_protocol_custom_response_headers_list(self):156        self.start_server(extra_headers=[('X-Spam', 'Eggs')])157        self.start_client('raw_headers')158        self.loop.run_until_complete(self.client.recv())159        resp_headers = self.loop.run_until_complete(self.client.recv())160        self.assertIn("('X-Spam', 'Eggs')", resp_headers)161        self.stop_client()162        self.stop_server()163    def test_no_subprotocol(self):164        self.start_server()165        self.start_client('subprotocol')166        server_subprotocol = self.loop.run_until_complete(self.client.recv())167        self.assertEqual(server_subprotocol, repr(None))168        self.assertEqual(self.client.subprotocol, None)169        self.stop_client()170        self.stop_server()171    def test_subprotocol_found(self):172        self.start_server(subprotocols=['superchat', 'chat'])173        self.start_client('subprotocol', subprotocols=['otherchat', 'chat'])174        server_subprotocol = self.loop.run_until_complete(self.client.recv())175        self.assertEqual(server_subprotocol, repr('chat'))176        self.assertEqual(self.client.subprotocol, 'chat')177        self.stop_client()178        self.stop_server()179    def test_subprotocol_not_found(self):180        self.start_server(subprotocols=['superchat'])181        self.start_client('subprotocol', subprotocols=['otherchat'])182        server_subprotocol = self.loop.run_until_complete(self.client.recv())183        self.assertEqual(server_subprotocol, repr(None))184        self.assertEqual(self.client.subprotocol, None)185        self.stop_client()186        self.stop_server()187    def test_subprotocol_not_offered(self):188        self.start_server()189        self.start_client('subprotocol', subprotocols=['otherchat', 'chat'])190        server_subprotocol = self.loop.run_until_complete(self.client.recv())191        self.assertEqual(server_subprotocol, repr(None))192        self.assertEqual(self.client.subprotocol, None)193        self.stop_client()194        self.stop_server()195    def test_subprotocol_not_requested(self):196        self.start_server(subprotocols=['superchat', 'chat'])197        self.start_client('subprotocol')198        server_subprotocol = self.loop.run_until_complete(self.client.recv())199        self.assertEqual(server_subprotocol, repr(None))200        self.assertEqual(self.client.subprotocol, None)201        self.stop_client()202        self.stop_server()203    @unittest.mock.patch.object(WebSocketServerProtocol, 'select_subprotocol')204    def test_subprotocol_error(self, _select_subprotocol):205        _select_subprotocol.return_value = 'superchat'206        self.start_server(subprotocols=['superchat'])207        with self.assertRaises(InvalidHandshake):208            self.start_client('subprotocol', subprotocols=['otherchat'])209        self.run_loop_once()210        self.stop_server()211    @unittest.mock.patch('websockets.server.read_request')212    def test_server_receives_malformed_request(self, _read_request):213        _read_request.side_effect = ValueError("read_request failed")214        self.start_server()215        with self.assertRaises(InvalidHandshake):216            self.start_client()217        self.stop_server()218    @unittest.mock.patch('websockets.client.read_response')219    def test_client_receives_malformed_response(self, _read_response):220        _read_response.side_effect = ValueError("read_response failed")221        self.start_server()222        with self.assertRaises(InvalidHandshake):223            self.start_client()224        self.run_loop_once()225        self.stop_server()226    @unittest.mock.patch('websockets.client.build_request')227    def test_client_sends_invalid_handshake_request(self, _build_request):228        def wrong_build_request(set_header):229            return '42'230        _build_request.side_effect = wrong_build_request231        self.start_server()232        with self.assertRaises(InvalidHandshake):233            self.start_client()234        self.stop_server()235    @unittest.mock.patch('websockets.server.build_response')236    def test_server_sends_invalid_handshake_response(self, _build_response):237        def wrong_build_response(set_header, key):238            return build_response(set_header, '42')239        _build_response.side_effect = wrong_build_response240        self.start_server()241        with self.assertRaises(InvalidHandshake):242            self.start_client()243        self.stop_server()244    @unittest.mock.patch('websockets.client.read_response')245    def test_server_does_not_switch_protocols(self, _read_response):246        @asyncio.coroutine247        def wrong_read_response(stream):248            code, headers = yield from read_response(stream)249            return 400, headers250        _read_response.side_effect = wrong_read_response251        self.start_server()252        with self.assertRaises(InvalidHandshake):253            self.start_client()254        self.run_loop_once()255        self.stop_server()256    @unittest.mock.patch('websockets.server.WebSocketServerProtocol.send')257    def test_server_handler_crashes(self, send):258        send.side_effect = ValueError("send failed")259        self.start_server()260        self.start_client()261        self.loop.run_until_complete(self.client.send("Hello!"))262        with self.assertRaises(ConnectionClosed):263            self.loop.run_until_complete(self.client.recv())264        self.stop_client()265        self.stop_server()266        # Connection ends with an unexpected error.267        self.assertEqual(self.client.close_code, 1011)268    @unittest.mock.patch('websockets.server.WebSocketServerProtocol.close')269    def test_server_close_crashes(self, close):270        close.side_effect = ValueError("close failed")271        self.start_server()272        self.start_client()273        self.loop.run_until_complete(self.client.send("Hello!"))274        reply = self.loop.run_until_complete(self.client.recv())275        self.assertEqual(reply, "Hello!")276        self.stop_client()277        self.stop_server()278        # Connection ends with an abnormal closure.279        self.assertEqual(self.client.close_code, 1006)280    @unittest.mock.patch.object(WebSocketClientProtocol, 'handshake')281    def test_client_closes_connection_before_handshake(self, handshake):282        self.start_server()283        self.start_client()284        # We have mocked the handshake() method to prevent the client from285        # performing the opening handshake. Force it to close the connection.286        self.loop.run_until_complete(self.client.close_connection(force=True))287        self.stop_client()288        # The server should stop properly anyway. It used to hang because the289        # worker handling the connection was waiting for the opening handshake.290        self.stop_server()291    @unittest.mock.patch('websockets.server.read_request')292    def test_server_shuts_down_during_opening_handshake(self, _read_request):293        _read_request.side_effect = asyncio.CancelledError294        self.start_server()295        self.server.closing = True296        with self.assertRaises(InvalidHandshake) as raised:297            self.start_client()298        self.stop_server()299        # Opening handshake fails with 503 Service Unavailable300        self.assertEqual(str(raised.exception), "Bad status code: 503")301    def test_server_shuts_down_during_connection_handling(self):302        self.start_server()303        self.start_client()304        self.server.close()305        with self.assertRaises(ConnectionClosed):306            self.loop.run_until_complete(self.client.recv())307        self.stop_client()308        self.stop_server()309        # Websocket connection terminates with 1001 Going Away.310        self.assertEqual(self.client.close_code, 1001)311@unittest.skipUnless(os.path.exists(testcert), "test certificate is missing")312class SSLClientServerTests(ClientServerTests):313    secure = True314    @property315    def server_context(self):316        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)317        ssl_context.load_cert_chain(testcert)318        return ssl_context319    @property320    def client_context(self):321        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)322        ssl_context.load_verify_locations(testcert)323        ssl_context.verify_mode = ssl.CERT_REQUIRED324        return ssl_context325    def start_server(self, *args, **kwds):326        kwds['ssl'] = self.server_context327        server = serve(handler, 'localhost', 8642, **kwds)328        self.server = self.loop.run_until_complete(server)329    def start_client(self, path='', **kwds):330        kwds['ssl'] = self.client_context331        client = connect('wss://localhost:8642/' + path, **kwds)332        self.client = self.loop.run_until_complete(client)333    def test_ws_uri_is_rejected(self):334        self.start_server()335        client = connect('ws://localhost:8642/', ssl=self.client_context)336        with self.assertRaises(ValueError):337            self.loop.run_until_complete(client)338        self.stop_server()339class ClientServerOriginTests(unittest.TestCase):340    def setUp(self):341        self.loop = asyncio.new_event_loop()342        asyncio.set_event_loop(self.loop)343    def tearDown(self):...

test_service_scheduler.py

Source:test_service_scheduler.py

...37    server.start()38    yield39    server.stop(None)40@pytest.fixture()41def start_client():42    channel = grpc.insecure_channel(43        f"{host}:{listening_port}", options=insecure_options)44    stub = scheduler_pb2_grpc.SchedulerStub(channel)45    return stub46def yield_post_request():47    requests = [48        commu_pb2.PostRequest(key='test~test_channel_1~1', value=bytes(1)),49        commu_pb2.PostRequest(key='test~test_channel_1~1', value=bytes(2)),50        commu_pb2.PostRequest(key='test~test_channel_1~1', value=bytes(3))51    ]52    for r in requests:53        yield r54class TestSchedulerService():55    def test_post(self, start_client, mocker):...

pydevd_hooks.py

Source:pydevd_hooks.py

...12    client, _ = server.accept()13    pydevd = daemon.start(server)14    daemon.set_connection(client)15    return pydevd16def start_client(daemon, host, port):17    """Return a socket to an existing "remote" pydevd-handling daemon.18    The daemon supports the pydevd client wire protocol, sending19    requests and handling responses (and events).20    This is a replacement for _pydevd_bundle.pydevd_comm.start_client.21    """22    client = create_client()23    client.connect((host, port))24    pydevd = daemon.start()25    daemon.set_connection(client)26    return pydevd27def install(pydevd, address,28            start_server=start_server, start_client=start_client,29            **kwargs):30    """Configure pydevd to use our wrapper.31    This is a bit of a hack to allow us to run our VSC debug adapter32    in the same process as pydevd.  Note that, as with most hacks,33    this is somewhat fragile (since the monkeypatching sites may34    change).35    """36    addr = Address.from_raw(address)37    daemon = Daemon(**kwargs)38    _start_server = (lambda p: start_server(daemon, addr.host, p))39    _start_server.orig = start_server40    _start_client = (lambda h, p: start_client(daemon, h, p))41    _start_client.orig = start_client42    # These are the functions pydevd invokes to get a socket to the client.43    pydevd_comm.start_server = _start_server44    pydevd_comm.start_client = _start_client45    # Ensure that pydevd is using our functions.46    pydevd.start_server = _start_server47    pydevd.start_client = _start_client48    __main__ = sys.modules['__main__']49    if __main__ is not pydevd and __main__.__file__ == pydevd.__file__:50        __main__.start_server = _start_server51        __main__.start_client = _start_client...

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.