How to use expect_websocket method in Playwright Python

Best Python code snippet using playwright-python

realtimechess_test.py

Source:realtimechess_test.py Github

copy

Full Screen

...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())...

Full Screen

Full Screen

conftest.py

Source:conftest.py Github

copy

Full Screen

...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)...

Full Screen

Full Screen

test_html_rendering.py

Source:test_html_rendering.py Github

copy

Full Screen

...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 (...

Full Screen

Full Screen

test_slider_action.py

Source:test_slider_action.py Github

copy

Full Screen

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

Full Screen

Full Screen

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:

  1. 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.
  2. 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.
  3. 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.
  4. 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.
  5. Inputs And Buttons In Playwright: Every website has Input boxes and buttons; learn about testing inputs and buttons with different scenarios and examples.
  6. 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.
  7. 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.
  8. 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.
  9. 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.
  10. 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.
  11. 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.
  12. 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!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful