KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use start_client method in pytest-mozwebqa
Best Python code snippet using pytest-mozwebqa_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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run pytest-mozwebqa automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
