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_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_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_ffpuppet_02(tmp_path):
"""test crash on start"""
with FFPuppet() as ffp:
prefs = (tmp_path / "prefs.js")
prefs.write_bytes(b"//fftest_startup_crash\n")
with pytest.raises(BrowserTerminatedError, match="Failure waiting for browser connection"):
ffp.launch(TESTFF_BIN, prefs_js=str(prefs))
ffp.close()
assert not ffp.is_running() # process should be gone
assert ffp.launches == 0
assert ffp.reason == ffp.RC_ALERT
def wait(self, cb_continue, timeout=60, url=None):
assert self._socket is not None
start_time = time.time()
time_limit = start_time + timeout
try:
conn = None
log.debug("waiting for browser connection...")
while True:
try:
conn, _ = self._socket.accept()
except socket.timeout:
if not cb_continue():
raise BrowserTerminatedError("Failure waiting for browser connection")
if time.time() >= time_limit:
raise BrowserTimeoutError("Timeout waiting for browser connection")
continue
break
conn.settimeout(1)
received = False
log.debug("waiting to receive browser request...")
while True:
try:
request = conn.recv(self.BUF_SIZE)
except socket.timeout:
if not cb_continue():
raise BrowserTerminatedError("Failure waiting for request")
if time.time() >= time_limit:
raise BrowserTimeoutError("Timeout waiting for request")
if not received:
except socket.timeout:
if not cb_continue():
raise BrowserTerminatedError("Failure waiting for browser connection")
if time.time() >= time_limit:
raise BrowserTimeoutError("Timeout waiting for browser connection")
continue
break
conn.settimeout(1)
received = False
log.debug("waiting to receive browser request...")
while True:
try:
request = conn.recv(self.BUF_SIZE)
except socket.timeout:
if not cb_continue():
raise BrowserTerminatedError("Failure waiting for request")
if time.time() >= time_limit:
raise BrowserTimeoutError("Timeout waiting for request")
if not received:
continue
if not received and not request:
log.warning("Empty request received from browser during bootstrap!")
elif len(request) == self.BUF_SIZE:
# maybe there is more to read...
received = True
continue
break
# build response
if url is None:
resp = "HTTP/1.1 204 No Content\r\nConnection: close\r\n\r\n"
else:
resp = "HTTP/1.1 301 Moved Permanently\r\n" \
if url is None:
resp = "HTTP/1.1 204 No Content\r\nConnection: close\r\n\r\n"
else:
resp = "HTTP/1.1 301 Moved Permanently\r\n" \
"Location: %s\r\n" \
"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()