KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use expect_websocket method in Playwright Python
Best Python code snippet using playwright-python
realtimechess_test.py
Source:realtimechess_test.py 
...33 async def disable_time(self):34 await self.call("setdebug")35 async def enable_time(self):36 await self.call("setdebug", {"debug": 0})37 async def expect_websocket(self):38 return json.loads((await self.ws.receive()).data)39 async def get_state(self):40 return GameState(await self.call("getstate"))41 async def join_game(self, host):42 await self.ws.close()43 data = await self.request("/?g=" + host.game)44 self.game = host.game45 self.ws = await self.client.ws_connect("/websocket?g=" + self.game)46 return data47 async def losses(self):48 map = json.loads(await self.request("/getplayer"))49 return int(map["losses"])50 async def move(self, from_pos, to_pos):51 await self.call("move", {"from": from_pos, "to": to_pos})52 async def move_websocket(self, from_pos, to_pos):53 await self.ws.send_str("/move?g={}&from={}&to={}".format(54 self.game, from_pos, to_pos))55 async def new_game(self):56 # Access implementation detail to restore cookies.57 self.client.session._cookie_jar = self.cookie_jar58 response = await self.client.request("GET", "/")59 response.raise_for_status()60 self.game = response.url.query.get("g")61 async def rating(self):62 map = json.loads(await self.request("/getplayer"))63 return int(map["rating"])64 async def request(self, path, data=None, method="GET"):65 # Access implementation detail to restore cookies.66 self.client.session._cookie_jar = self.cookie_jar67 response = await self.client.request(method, path, data=data)68 response.raise_for_status()69 return await response.text()70 async def wins(self):71 map = json.loads(await self.request("/getplayer"))72 return int(map["wins"])73class GameState:74 def __init__(self, json_string):75 self.data = json.loads(json_string)76 self.data["state"] = int(self.data["state"])77 def board(self):78 pieces = []79 for i in range(32):80 pieces.append(self.data["p" + str(i)])81 return board.Board(pieces)82 def moving(self, piece):83 pass84 def ready(self, color):85 assert color == constants.WHITE or color == constants.BLACK86 if color == constants.WHITE:87 return self.data["userXReady"]88 else:89 return self.data["userOReady"]90 def game_state(self):91 return self.data["state"]92 def __str__(self):93 return str(self.data)94def _wrap_in_loop(f):95 """Decorator that runs the member function in self.loop if needed."""96 @functools.wraps(f)97 def wrapper(self, *args, **kwargs):98 result = f(self, *args, **kwargs)99 if inspect.iscoroutine(result):100 result = self.loop.run_until_complete(result)101 return result102 return wrapper103def async_test(cls):104 """Class decorator that allows test methods to be async."""105 attributes = dir(cls)106 for attribute in attributes:107 if attribute.startswith("test"):108 f = getattr(cls, attribute)109 setattr(cls, attribute, _wrap_in_loop(f))110 return cls111@async_test112class GameTest(AioHTTPTestCase):113 async def get_application(self):114 return realtimechess.make_app(True)115 async def setUpAsync(self):116 self.user1 = User(self.client, "user1")117 await self.user1.connect()118 self.user2 = User(self.client, "user2")119 await self.user2.connect()120 self.assertEqual(constants.STATE_START,121 (await self.user1.get_state()).game_state())122 await self.user2.join_game(self.user1)123 self.assertEqual(constants.STATE_START,124 (await self.user1.get_state()).game_state())125 await self.user1.call("ready", {"ready": 1})126 self.assertEqual(constants.STATE_START,127 (await self.user2.get_state()).game_state())128 await self.user2.call("ready", {"ready": 1})129 await self.user1.expect_websocket()130 await self.user2.expect_websocket()131 self.assertEqual(constants.STATE_PLAY,132 (await self.user1.get_state()).game_state())133 async def tearDownAsync(self):134 pass135 def test_async_test(self):136 # Test that non-coroutines also work.137 pass138 async def test_redirect_to_loginpage(self):139 # Access implementation detail to clear cookies.140 self.client.session._cookie_jar = aiohttp.CookieJar(unsafe=True)141 response = await self.client.request("GET", "/")142 assert response.status == 200143 text = await response.text()144 self.assertIn("Choose your name", text)145 self.assertIn("loginpage", str(response.url))146 async def test_redirect_to_loginpage_with_game(self):147 # Access implementation detail to clear cookies.148 self.client.session._cookie_jar = aiohttp.CookieJar(unsafe=True)149 response = await self.client.request("GET", "/?g=deadbeef")150 assert response.status == 200151 text = await response.text()152 self.assertIn("Choose your name", text)153 self.assertIn("deadbeef", str(response.url))154 self.assertIn("loginpage", str(response.url))155 async def test_state(self):156 self.assertTrue((await self.user1.get_state()).board().has_piece("A2"))157 self.assertEqual(board.PAWN,158 (await159 self.user1.get_state()).board().piece("A2").type)160 self.assertEqual(board.WHITE,161 (await162 self.user1.get_state()).board().piece("A2").color)163 async def test_opened(self):164 await self.user1.call("opened")165 await self.user1.expect_websocket()166 await self.user2.expect_websocket()167 async def test_move(self):168 self.assertTrue((await self.user1.get_state()).board().has_piece("A2"))169 await self.user1.move("A2", "A3")170 await self.user1.expect_websocket()171 await self.user2.expect_websocket()172 self.assertFalse((await173 self.user1.get_state()).board().has_piece("A2"))174 await self.user1.disable_time()175 self.assertTrue((await self.user1.get_state()).board().has_piece("A3"))176 async def test_move_websocket(self):177 self.assertTrue((await self.user1.get_state()).board().has_piece("A2"))178 await self.user1.move_websocket("A2", "A3")179 await self.user1.expect_websocket()180 await self.user2.expect_websocket()181 state = await self.user1.get_state()182 self.assertFalse(state.board().has_piece("A2"))183 await self.user1.disable_time()184 state = await self.user1.get_state()185 self.assertTrue((await self.user1.get_state()).board().has_piece("A3"))186 async def test_move_websocket_error(self):187 await self.user1.move_websocket("A2", "A1")188 await self.user1.move_websocket("A2", "")189 await self.user1.move_websocket("Q21", "A121")190 async def test_move_no_destination(self):191 # Will not give an error.192 await self.user1.ws.send_str("/move")193 with self.assertRaises(aiohttp.ClientResponseError) as cm:194 await self.user1.call("move", {"from": "A2"})195 self.assertEqual(400, cm.exception.code)196 async def test_move_other_players(self):197 with self.assertRaises(aiohttp.ClientResponseError) as cm:198 await self.user1.move("A7", "A6")199 self.assertEqual(403, cm.exception.code)200 async def test_move_from_invalid_position(self):201 with self.assertRaises(aiohttp.ClientResponseError) as cm:202 await self.user1.move("Q9", "A6")203 self.assertEqual(400, cm.exception.code)204 async def test_move_to_invalid_position(self):205 with self.assertRaises(aiohttp.ClientResponseError) as cm:206 await self.user1.move("A2", "P0")207 self.assertEqual(400, cm.exception.code)208 async def test_move_from_empty(self):209 with self.assertRaises(aiohttp.ClientResponseError) as cm:210 await self.user1.move("A3", "A4")211 self.assertEqual(404, cm.exception.code)212 async def test_move_moving(self):213 await self.user1.move("A2", "A3")214 with self.assertRaises(aiohttp.ClientResponseError) as cm:215 await self.user1.move("A3", "A4")216 # There is no piece in A3 (yet).217 self.assertEqual(404, cm.exception.code)218 self.assertFalse((await219 self.user1.get_state()).board().has_piece("A4"))220 async def test_capture(self):221 await self.user1.disable_time()222 await self.user1.move("E2", "E3")223 await self.user1.move("D1", "G4") #QG4224 self.assertEqual(board.QUEEN,225 (await226 self.user1.get_state()).board().piece("G4").type)227 await self.user2.move("D7", "D6")228 await self.user2.move("C8", "G4") #BG4229 self.assertEqual(board.BISHOP,230 (await231 self.user1.get_state()).board().piece("G4").type)232 await self.user2.move("G4", "F3") #BF3233 self.assertIs(None, (await self.user1.get_state()).board().piece("A3"))234 async def test_capture2(self):235 await self.user1.disable_time()236 await self.user1.move("B2", "B3")237 await self.user1.move("C1", "B2")238 self.assertEqual(constants.PAWN,239 (await240 self.user1.get_state()).board().piece("C7").type)241 await self.user1.move("B2", "G7")242 self.assertEqual(constants.BISHOP,243 (await244 self.user1.get_state()).board().piece("G7").type)245 self.assertEqual(constants.ROOK,246 (await247 self.user1.get_state()).board().piece("H8").type)248 await self.user1.move("G7", "H8")249 self.assertEqual(constants.BISHOP,250 (await251 self.user1.get_state()).board().piece("H8").type)252 async def test_promotion_white(self):253 await self.user1.disable_time()254 await self.user1.move("B2", "B4")255 await self.user1.move("B4", "B5")256 await self.user1.move("B5", "B6")257 await self.user1.move("B6", "C7")258 self.assertEqual(constants.KNIGHT,259 (await260 self.user1.get_state()).board().piece("B8").type)261 await self.user1.move("C7", "B8")262 self.assertEqual(constants.QUEEN,263 (await264 self.user1.get_state()).board().piece("B8").type)265 async def test_promotion_black(self):266 await self.user1.disable_time()267 await self.user2.move("H7", "H5")268 await self.user2.move("H5", "H4")269 await self.user2.move("H4", "H3")270 await self.user2.move("H3", "G2")271 self.assertEqual(constants.ROOK,272 (await273 self.user1.get_state()).board().piece("H1").type)274 self.assertEqual(constants.WHITE,275 (await276 self.user1.get_state()).board().piece("H1").color)277 await self.user2.move("G2", "H1")278 self.assertEqual(constants.QUEEN,279 (await280 self.user1.get_state()).board().piece("H1").type)281 self.assertEqual(constants.BLACK,282 (await283 self.user1.get_state()).board().piece("H1").color)284 async def test_websocket_ping(self):285 # Invalid command should be ignored.286 await self.user1.ws.send_str("/invalid")287 await self.user1.move("A2", "A3")288 await self.user1.expect_websocket()289 await self.user1.ws.send_str("/ping")290 await self.user1.expect_websocket()291 async def test_websocket_no_game(self):292 with self.assertRaises(aiohttp.ClientResponseError) as cm:293 await self.user1.client.ws_connect("/websocket")294 self.assertEqual(404, cm.exception.code)295 async def test_websocket_invalid_game(self):296 with self.assertRaises(aiohttp.ClientResponseError) as cm:297 await self.user1.client.ws_connect("/websocket?g=deadbeef")298 self.assertEqual(404, cm.exception.code)299 async def test_full_games(self):300 self.assertEqual(1000, await self.user1.rating())301 self.assertEqual(1000, await self.user2.rating())302 self.assertEqual(0, await self.user1.wins())303 self.assertEqual(0, await self.user1.losses())304 self.assertEqual(0, await self.user2.wins())...
conftest.py
Source:conftest.py
...164 nb.write_text(json.dumps(notebook_content))165 url_base = f"http://localhost:{port}"166 name = nb.name167 url = f"{url_base}/notebooks/{name}"168 with page.expect_websocket(kernel_ready):169 page.goto(url)170 # Blank cells do not increment execution_count171 expected_count = sum(1 for cell in cells if cell["source"])172 run_menu = "#celllink"173 run_all_button = "text=Run All"174 # This will raise a TimeoutError when the alert is presented, since the175 # `wait_for_selector` will never happen. Raising our own error here doesn't176 # work due to https://github.com/microsoft/playwright-python/issues/1017177 def close_on_dialog(dialog: Dialog) -> None:178 if dialog.message.startswith("WARNING:"):179 page.close()180 else:181 dialog.dismiss()182 page.on("dialog", close_on_dialog)183 page.click(run_menu)184 page.click(run_all_button)185 page.wait_for_selector(f"text=[{expected_count}]:", strict=True)186 with page.expect_response(lambda resp: is_saved(resp, name, port)) as resp:187 page.click('button[title="Save and Checkpoint"]')188 kernel_menu = "#kernellink"189 kernel_shutdown_button = "text=Shutdown"190 real_shutdown_button = 'button:has-text("Shutdown")'191 page.click(kernel_menu)192 page.click(kernel_shutdown_button)193 with page.expect_response(lambda resp: is_closing(resp, port)) as resp:194 page.click(real_shutdown_button)195 resp.value.finished()196 return json.loads(nb.read_text())197@pytest.fixture(scope="function")198def lab(199 jupyter_lab: t.Tuple[BrowserContext, Path, int],200) -> t.Callable:201 """Provide function-scoped fixture leveraging longer lived fixtures."""202 context, tmp, port = jupyter_lab203 path = tmp / f"notebook-{uuid4()}.ipynb"204 return lambda json_in: _lab(205 json_in,206 context=context,207 port=port,208 nb=path,209 )210def _lab(211 json_in: t.Dict[str, t.Any],212 context: BrowserContext,213 port: int,214 nb: Path,215) -> t.Dict[str, t.Any]:216 page = context.new_page()217 notebook_content = _base_notebook()218 cells = t.cast(t.List, notebook_content["cells"])219 cells.extend(json_in)220 notebook_content["cells"] = [make_cell(cell) for cell in cells]221 name = nb.name222 nb.write_text(json.dumps(notebook_content))223 url_base = f"http://localhost:{port}"224 url = f"{url_base}/lab/tree/{name}"225 with page.expect_websocket(kernel_ready):226 page.goto(url)227 # Blank cells do not increment execution_count228 expected_count = sum(1 for cell in cells if cell["source"])229 run_menu = "text=Run"230 run_all_button = 'ul[role="menu"] >> text=Run All Cells'231 page.click(run_menu)232 page.click(run_all_button)233 page.wait_for_selector(f"text=[{expected_count}]:", strict=True)234 with page.expect_response(lambda resp: is_saved(resp, name, port)) as resp:235 page.click("text=File")236 page.click('ul[role="menu"] >> text=Save Notebook')237 kernel_menu = "text=Kernel"238 shutdown_button = "text=Shut Down Kernel"239 page.click(kernel_menu)...
test_html_rendering.py
Source:test_html_rendering.py
...41 gui._set_frame(inspect.currentframe())42 gui.add_page("page1", Html(html_content))43 helpers.run_e2e(gui)44 page.goto("/page1")45 page.expect_websocket()46 page.wait_for_selector("#text1")47 assert (48 page.evaluate('window.getComputedStyle(document.querySelector("#text1"), null).getPropertyValue("color")')49 == "rgb(0, 128, 0)"50 )51 assert (52 page.evaluate('window.getComputedStyle(document.querySelector("#text2"), null).getPropertyValue("color")')53 == "rgb(0, 0, 255)"54 )55@pytest.mark.teste2e56def test_html_render_bind_assets(page: "Page", gui: Gui, helpers):57 gui._set_frame(inspect.currentframe())58 gui.add_pages(pages=f"{Path(Path(__file__).parent.resolve())}{os.path.sep}test-assets")59 helpers.run_e2e(gui)60 assert ".taipy-text" in urlopen("http://127.0.0.1:5000/test-assets/style/style.css").read().decode("utf-8")61 page.goto("/test-assets/page1")62 page.expect_websocket()63 page.wait_for_selector("#text1")64 retry = 065 while (66 retry < 1067 and page.evaluate('window.getComputedStyle(document.querySelector("#text1"), null).getPropertyValue("color")')68 != "rgb(0, 128, 0)"69 ):70 retry += 171 time.sleep(0.1)72 assert (73 page.evaluate('window.getComputedStyle(document.querySelector("#text1"), null).getPropertyValue("color")')74 == "rgb(0, 128, 0)"75 )76 assert (77 page.evaluate('window.getComputedStyle(document.querySelector("#text2"), null).getPropertyValue("color")')78 == "rgb(0, 0, 255)"79 )80@pytest.mark.teste2e81def test_html_render_path_mapping(page: "Page", gui: Gui, helpers):82 gui._server = _Server(83 gui,84 path_mapping={"style": f"{Path(Path(__file__).parent.resolve())}{os.path.sep}test-assets{os.path.sep}style"},85 )86 gui.add_page("page1", Html(f"{Path(Path(__file__).parent.resolve())}{os.path.sep}page1.html"))87 helpers.run_e2e(gui)88 assert ".taipy-text" in urlopen("http://127.0.0.1:5000/style/style.css").read().decode("utf-8")89 page.goto("/page1")90 page.expect_websocket()91 page.wait_for_selector("#text1")92 retry = 093 while (94 retry < 1095 and page.evaluate('window.getComputedStyle(document.querySelector("#text1"), null).getPropertyValue("color")')96 != "rgb(0, 128, 0)"97 ):98 retry += 199 time.sleep(0.1)100 assert (101 page.evaluate('window.getComputedStyle(document.querySelector("#text1"), null).getPropertyValue("color")')102 == "rgb(0, 128, 0)"103 )104 assert (...
test_slider_action.py
Source:test_slider_action.py
...25 gui._set_frame(inspect.currentframe())26 gui.add_page(name="test", page=page_md)27 helpers.run_e2e(gui)28 page.goto("/test")29 page.expect_websocket()30 page.wait_for_selector("#text1")31 text1 = page.query_selector("#text1")32 assert text1.inner_text() == "10"33 page.wait_for_selector("#slider1")34 page.fill("#slider1 input", "20")35 function_evaluated = True36 try:37 page.wait_for_function("document.querySelector('#text1').innerText !== '10'")38 except Exception as e:39 function_evaluated = False40 logging.getLogger().debug(f"Function evaluation timeout.\n{e}")41 if function_evaluated:42 text1_2 = page.query_selector("#text1")43 assert text1_2.inner_text() == "20"44@pytest.mark.teste2e45def test_slider_action_on_change(page: "Page", gui: Gui, helpers):46 d = {"v1": 10, "v2": 10}47 def on_change(state, var, val):48 if var == "d.v2":49 d = {"v1": 2 * val, "v2": val}50 state.d.update(d)51 page_md = """52Value: <|{d.v1}|id=text1|>53Slider: <|{d.v2}|slider|id=slider1|>54"""55 gui._set_frame(inspect.currentframe())56 gui.add_page(name="test", page=page_md)57 helpers.run_e2e(gui)58 page.goto("/test")59 page.expect_websocket()60 page.wait_for_selector("#text1")61 text1 = page.query_selector("#text1")62 assert text1.inner_text() == "10"63 page.wait_for_selector("#slider1")64 page.fill("#slider1 input", "20")65 function_evaluated = True66 try:67 page.wait_for_function("document.querySelector('#text1').innerText !== '10'")68 except Exception as e:69 function_evaluated = False70 logging.getLogger().debug(f"Function evaluation timeout.\n{e}")71 if function_evaluated:72 text1_2 = page.query_selector("#text1")73 assert text1_2.inner_text() == "40"
Playwright tutorial
LambdaTest’s Playwright tutorial will give you a broader idea about the Playwright automation framework, its unique features, and use cases with examples to exceed your understanding of Playwright testing. This tutorial will give A to Z guidance, from installing the Playwright framework to some best practices and advanced concepts.
Chapters:
- What is Playwright : Playwright is comparatively new but has gained good popularity. Get to know some history of the Playwright with some interesting facts connected with it.
- How To Install Playwright : Learn in detail about what basic configuration and dependencies are required for installing Playwright and run a test. Get a step-by-step direction for installing the Playwright automation framework.
- Playwright Futuristic Features: Launched in 2020, Playwright gained huge popularity quickly because of some obliging features such as Playwright Test Generator and Inspector, Playwright Reporter, Playwright auto-waiting mechanism and etc. Read up on those features to master Playwright testing.
- What is Component Testing: Component testing in Playwright is a unique feature that allows a tester to test a single component of a web application without integrating them with other elements. Learn how to perform Component testing on the Playwright automation framework.
- Inputs And Buttons In Playwright: Every website has Input boxes and buttons; learn about testing inputs and buttons with different scenarios and examples.
- Functions and Selectors in Playwright: Learn how to launch the Chromium browser with Playwright. Also, gain a better understanding of some important functions like “BrowserContext,” which allows you to run multiple browser sessions, and “newPage” which interacts with a page.
- Handling Alerts and Dropdowns in Playwright : Playwright interact with different types of alerts and pop-ups, such as simple, confirmation, and prompt, and different types of dropdowns, such as single selector and multi-selector get your hands-on with handling alerts and dropdown in Playright testing.
- Playwright vs Puppeteer: Get to know about the difference between two testing frameworks and how they are different than one another, which browsers they support, and what features they provide.
- Run Playwright Tests on LambdaTest: Playwright testing with LambdaTest leverages test performance to the utmost. You can run multiple Playwright tests in Parallel with the LammbdaTest test cloud. Get a step-by-step guide to run your Playwright test on the LambdaTest platform.
- Playwright Python Tutorial: Playwright automation framework support all major languages such as Python, JavaScript, TypeScript, .NET and etc. However, there are various advantages to Python end-to-end testing with Playwright because of its versatile utility. Get the hang of Playwright python testing with this chapter.
- Playwright End To End Testing Tutorial: Get your hands on with Playwright end-to-end testing and learn to use some exciting features such as TraceViewer, Debugging, Networking, Component testing, Visual testing, and many more.
- Playwright Video Tutorial: Watch the video tutorials on Playwright testing from experts and get a consecutive in-depth explanation of Playwright automation testing.
Run Playwright Python 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?
