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_bootstrapper_05(mocker):
"""test Bootstrapper.wait() target crashed"""
fake_sock = mocker.Mock(socket.socket)
fake_conn = mocker.Mock(socket.socket)
# return empty buffer for test coverage
fake_conn.recv.return_value = ""
fake_sock.accept.return_value = (fake_conn, None)
mocker.patch("ffpuppet.bootstrapper.socket.socket", return_value=fake_sock)
with Bootstrapper() as bts:
with pytest.raises(BrowserTerminatedError, match="Failure during browser startup"):
bts.wait(lambda: False)
assert fake_conn.close.call_count == 1
def test_bootstrapper_03(mocker):
"""test Bootstrapper.wait() failure waiting for request"""
fake_sock = mocker.Mock(socket.socket)
fake_conn = mocker.Mock(socket.socket)
fake_conn.recv.side_effect = socket.timeout
fake_sock.accept.return_value = (fake_conn, None)
mocker.patch("ffpuppet.bootstrapper.socket.socket", return_value=fake_sock)
with Bootstrapper() as bts:
# test failure
with pytest.raises(BrowserTerminatedError, match="Failure waiting for request"):
bts.wait(lambda: False)
assert fake_conn.recv.call_count == 1
assert fake_conn.close.call_count == 1
fake_conn.reset_mock()
# test timeout
mocker.patch("ffpuppet.bootstrapper.time.time", side_effect=(1, 1, 1, 1, 2))
with pytest.raises(BrowserTimeoutError, match="Timeout waiting for request"):
bts.wait(lambda: True, timeout=0.1)
# should call recv() at least 2x for positive and negative timeout check
assert fake_conn.recv.call_count > 1
assert fake_conn.close.call_count == 1
def test_bootstrapper_07(mocker):
"""test Bootstrapper.wait() successful with redirect"""
fake_sock = mocker.Mock(socket.socket)
fake_conn = mocker.Mock(socket.socket)
fake_conn.recv.return_value = "AAAA"
fake_sock.accept.return_value = (fake_conn, None)
mocker.patch("ffpuppet.bootstrapper.socket.socket", return_value=fake_sock)
# without redirect
with Bootstrapper() as bts:
bts.wait(lambda: True, url="http://127.0.0.1:9999/test.html")
assert fake_conn.close.call_count == 1
assert fake_conn.recv.call_count == 1
assert fake_conn.sendall.call_count == 1
def test_bootstrapper_04(mocker):
"""test Bootstrapper.wait() failure sending response"""
fake_sock = mocker.Mock(socket.socket)
fake_conn = mocker.Mock(socket.socket)
fake_conn.recv.return_value = "A"
fake_conn.sendall.side_effect = socket.timeout
fake_sock.accept.return_value = (fake_conn, None)
mocker.patch("ffpuppet.bootstrapper.socket.socket", return_value=fake_sock)
with Bootstrapper() as bts:
# test timeout
with pytest.raises(BrowserTimeoutError, match="Timeout sending response"):
bts.wait(lambda: True)
assert fake_conn.recv.call_count == 1
assert fake_conn.sendall.call_count == 1
assert fake_conn.close.call_count == 1
fake_conn.reset_mock()
# test failure
with pytest.raises(BrowserTerminatedError, match="Failure during browser startup"):
bts.wait(lambda: False)
assert fake_conn.recv.call_count == 1
assert fake_conn.sendall.call_count == 1
assert fake_conn.close.call_count == 1
try:
conn.connect(("127.0.0.1", port))
break
except socket.timeout:
if not attempt:
raise
# send request and receive response
try:
conn.settimeout(10)
conn.sendall(b"A" * payload_size)
conn.send(b"")
conn.recv(8192)
finally:
conn.close()
with Bootstrapper() as bts:
browser_thread = threading.Thread(target=_fake_browser, args=(bts.port,))
try:
browser_thread.start()
bts.wait(lambda: True, timeout=10)
finally:
browser_thread.join()
def test_bootstrapper_01():
"""test simple Bootstrapper()"""
with Bootstrapper() as bts:
assert bts._socket is not None
assert bts.location.startswith("http://127.0.0.1:")
assert int(bts.location.split(":")[-1]) > 1024
assert bts.port > 1024
bts.close()
assert bts._socket is None
def test_bootstrapper_02(mocker):
"""test Bootstrapper.wait() failure waiting for initial connection"""
fake_sock = mocker.Mock(socket.socket)
fake_sock.accept.side_effect = socket.timeout
mocker.patch("ffpuppet.bootstrapper.socket.socket", return_value=fake_sock)
with Bootstrapper() as bts:
# test failure
with pytest.raises(BrowserTerminatedError, match="Failure waiting for browser connection"):
bts.wait(lambda: False)
assert fake_sock.accept.call_count == 1
fake_sock.reset_mock()
# test timeout
mocker.patch("ffpuppet.bootstrapper.time.time", side_effect=(1, 1, 1, 2))
with pytest.raises(BrowserTimeoutError, match="Timeout waiting for browser connection"):
bts.wait(lambda: True, timeout=0.1)
# should call accept() at least 2x for positive and negative timeout check
assert fake_sock.accept.call_count > 1
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
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)
log.debug("launch timeout: %d", launch_timeout)
# create and modify a profile
self.profile = create_profile(
extension=extension,
prefs_js=prefs_js,
template=self._profile_template)
# performing the bootstrap helps guarantee that the browser
# will be loaded and ready to accept input when launch() returns
bootstrapper = Bootstrapper()
try:
prefs = {
"capability.policy.policynames": "'localfilelinks'",
"capability.policy.localfilelinks.sites": "'%s'" % bootstrapper.location,
"capability.policy.localfilelinks.checkloaduri.enabled": "'allAccess'",
"privacy.partition.network_state": "false"}
if self._use_rr or self._use_valgrind:
# when the browser is running slowly socket reads can fail if this is > 0
prefs["network.http.speculative-parallel-limit"] = "0"
append_prefs(self.profile, prefs)
launch_args = [bootstrapper.location]
is_windows = platform.system() == "Windows"
if is_windows:
# disable launcher process