Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_ffpuppet_10():
"""test calling launch() multiple times"""
with FFPuppet() as ffp:
with HTTPTestServer() as srv:
for _ in range(10):
ffp.launch(TESTFF_BIN, location=srv.get_addr())
ffp.close()
# call 2x without calling close()
ffp.launch(TESTFF_BIN, location=srv.get_addr())
with pytest.raises(LaunchError, match="Process is already running"):
ffp.launch(TESTFF_BIN)
assert ffp.launches == 11
ffp.close()
def test_bootstrapper_09(mocker):
"""test Bootstrapper() exhaust port range"""
fake_sock = mocker.Mock(socket.socket)
fake_sock.bind.side_effect = socket.error(10013, "TEST")
mocker.patch("ffpuppet.bootstrapper.socket.socket", return_value=fake_sock)
with pytest.raises(LaunchError, match="Could not find available port"):
with Bootstrapper():
pass
def test_ffpuppet_19(tmp_path):
"""test launching with Valgrind"""
if platform.system() != "Linux":
with pytest.raises(EnvironmentError, match="Valgrind is only supported on Linux"):
FFPuppet(use_valgrind=True)
return
vmv = FFPuppet.VALGRIND_MIN_VERSION
try:
FFPuppet.VALGRIND_MIN_VERSION = 9999999999.99
with pytest.raises(EnvironmentError, match=r"Valgrind >= \d+\.\d+ is required"):
FFPuppet(use_valgrind=True)
FFPuppet.VALGRIND_MIN_VERSION = 0
with FFPuppet(use_valgrind=True) as ffp:
bin_path = str(check_output(["which", "echo"]).strip().decode("ascii"))
# launch will fail b/c 'echo' will exit right away but that's fine
with pytest.raises(LaunchError, match="Failure waiting for browser connection"):
ffp.launch(bin_path)
ffp.close()
ffp.save_logs(str(tmp_path / "logs"))
log_data = (tmp_path / "logs" / "log_stderr.txt").read_bytes()
# verify Valgrind ran and executed the script
assert b"valgrind -q" in log_data
assert b"[ffpuppet] Reason code: EXITED" in log_data
finally:
FFPuppet.VALGRIND_MIN_VERSION = vmv
"Connection: close\r\n\r\n" % url
conn.settimeout(max(int(time_limit - time.time()), 1))
log.debug("sending response (redirect %r)", url)
try:
conn.sendall(resp.encode("ascii"))
except socket.timeout:
resp_timeout = True
else:
resp_timeout = False
if not cb_continue():
raise BrowserTerminatedError("Failure during browser startup")
if resp_timeout:
raise BrowserTimeoutError("Timeout sending response")
log.debug("bootstrap complete (%0.2fs)", time.time() - start_time)
except socket.error as soc_e: # pragma: no cover
raise LaunchError("Error attempting to launch browser: %s" % soc_e)
finally:
if conn is not None:
conn.close()
exceeds the amount specified here.
@type prefs_js: String
@param prefs_js: Path to a prefs.js file to install in the Firefox profile.
@type extension: String, or list of Strings
@param extension: Path to an extension (e.g. DOMFuzz fuzzPriv extension) to be installed.
@rtype: None
@return: None
"""
assert self._launches > -1, "clean_up() has been called"
assert log_limit >= 0
assert memory_limit >= 0
if self._proc is not None:
raise LaunchError("Process is already running")
bin_path = os.path.abspath(bin_path)
if not os.path.isfile(bin_path) or not os.access(bin_path, os.X_OK):
raise IOError("%s is not an executable" % bin_path)
self._last_bin_path = os.path.dirname(bin_path) # need the path for minidump_stackwalk
log.debug("requested location: %r", location)
if location is not None:
if os.path.isfile(location):
location = "///".join(
["file:", pathname2url(os.path.realpath(location)).lstrip("/")])
elif re.match(r"http(s)?://", location, re.IGNORECASE) is None:
raise IOError("Cannot find %r" % location)
self.reason = None
launch_timeout = max(launch_timeout, self.LAUNCH_TIMEOUT_MIN)
# 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
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
class LaunchError(Exception):
"""
Raised when the browser process does not appear to be in a functional state during launch
"""
class BrowserTerminatedError(LaunchError):
"""
Raised when the browser process goes away during launch
"""
class BrowserTimeoutError(LaunchError):
"""
Raised when the browser process appears to hang during launch
"""
class InvalidPrefs(LaunchError):
"""
Raised when an invalid prefs.js file is used
"""
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
class LaunchError(Exception):
"""
Raised when the browser process does not appear to be in a functional state during launch
"""
class BrowserTerminatedError(LaunchError):
"""
Raised when the browser process goes away during launch
"""
class BrowserTimeoutError(LaunchError):
"""
Raised when the browser process appears to hang during launch
"""
class InvalidPrefs(LaunchError):
"""
Raised when an invalid prefs.js file is used
"""
class TerminateError(Exception):
"""
Raised when attempts to terminate the browser fail
"""
class BrowserTerminatedError(LaunchError):
"""
Raised when the browser process goes away during launch
"""
class BrowserTimeoutError(LaunchError):
"""
Raised when the browser process appears to hang during launch
"""
class InvalidPrefs(LaunchError):
"""
Raised when an invalid prefs.js file is used
"""
class TerminateError(Exception):
"""
Raised when attempts to terminate the browser fail
socket.SO_EXCLUSIVEADDRUSE, # pylint: disable=no-member
1)
self._socket.settimeout(poll_wait)
for _ in range(self.PORT_RETRIES):
try:
self._socket.bind(("127.0.0.1", random.randint(self.PORT_MIN, self.PORT_MAX)))
self._socket.listen(5)
except socket.error as soc_e:
if soc_e.errno in (errno.EADDRINUSE, 10013):
# Address already in use
continue
raise soc_e # pragma: no cover
break
else:
self._socket.close()
raise LaunchError("Could not find available port")