Source code
Revision control
Copy as Markdown
Other Tools
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
""" Helpers to build scenarii
"""
from condprof.util import logger
_SUPPORTED_MOBILE_BROWSERS = "fenix", "gecko", "firefox"
def is_mobile(platform):
return any(mobile in platform for mobile in _SUPPORTED_MOBILE_BROWSERS)
class TabSwitcher:
"""Helper used to create tabs and circulate in them."""
def __init__(self, session, options):
self.handles = None
self.current = 0
self.session = session
self._max = options.get("max_urls", 10)
self.platform = options.get("platform", "")
self.num_tabs = self._max >= 100 and 100 or self._max
self._mobile = is_mobile(self.platform)
async def create_windows(self):
# on mobile we don't use tabs for now
if self._mobile:
return
# creating tabs
for i in range(self.num_tabs):
await self.session._request(
url="/window/new", method="POST", data={"type": "tab"}
)
async def switch(self):
if self._mobile:
return
try:
if self.handles is None:
self.handles = await self.session.get_window_handles()
self.current = 0
except Exception:
logger.error("Could not get window handles")
return
handle = self.handles[self.current]
if self.current == len(self.handles) - 1:
self.current = 0
else:
self.current += 1
try:
await self.session.switch_to_window(handle)
except Exception:
logger.error("Could not switch to handle %s" % str(handle))
# 10 minutes
_SCRIPT_TIMEOUT = 10 * 60 * 1000
async def execute_async_script(session, script, *args):
# switch to the right context if needed
current_context = await session._request(url="/moz/context", method="GET")
if current_context != "chrome":
await session._request(
url="/moz/context", method="POST", data={"context": "chrome"}
)
switch_back = True
else:
switch_back = False
await session._request(
url="/timeouts", method="POST", data={"script": _SCRIPT_TIMEOUT}
)
try:
attempts = 0
while True:
try:
return await session._request(
url="/execute/async",
method="POST",
data={"script": script, "args": list(args)},
)
except Exception as e:
attempts += 1
logger.error("The script failed.", exc_info=True)
if attempts > 2:
return {
"result": 1,
"result_message": str(e),
"result_exc": e,
"logs": {},
}
finally:
if switch_back:
await session._request(
url="/moz/context", method="POST", data={"context": current_context}
)
async def close_extra_windows(session):
logger.info("Closing all tabs")
handles = await session.get_window_handles()
# we're closing all tabs except the last one
for handle in handles[:-1]:
await session.switch_to_window(handle)
await session._request(url="/window", method="DELETE")