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_puppet_target_03(mocker, tmp_path):
"""test PuppetTarget.detect_failure()"""
fake_ffp = mocker.patch("grizzly.target.puppet_target.FFPuppet", autospec=True)
fake_ffp.RC_ALERT = FFPuppet.RC_ALERT
fake_ffp.RC_CLOSED = FFPuppet.RC_CLOSED
fake_ffp.RC_EXITED = FFPuppet.RC_EXITED
fake_ffp.RC_WORKER = FFPuppet.RC_WORKER
fake_file = tmp_path / "fake"
fake_file.touch()
target = PuppetTarget(str(fake_file), None, 300, 25, 5000, str(fake_file), 25)
# no failures
fake_ffp.return_value.is_healthy.return_value = True
fake_ffp.return_value.reason = None
assert target.detect_failure([], False) == Target.RESULT_NONE
assert target.detect_failure(["memory"], False) == Target.RESULT_NONE
assert not target.closed
# test close
fake_ffp.return_value.is_healthy.call_count = 0
fake_ffp.return_value.is_healthy.return_value = False
fake_ffp.return_value.is_running.return_value = False
fake_ffp.return_value.reason = FFPuppet.RC_CLOSED
def test_puppet_target_01(mocker, tmp_path):
"""test creating a PuppetTarget"""
fake_ffp = mocker.patch("grizzly.target.puppet_target.FFPuppet", autospec=True)
fake_ffp.return_value.reason = FFPuppet.RC_CLOSED
fake_ffp.return_value.log_length.return_value = 562
fake_file = tmp_path / "fake"
fake_file.touch()
target = PuppetTarget(str(fake_file), None, 300, 25, 5000, None, 25)
assert target.closed
assert target.detect_failure([], False) == Target.RESULT_NONE
assert target.log_size() == 1124
fake_ffp.return_value.log_length.assert_any_call("stderr")
fake_ffp.return_value.log_length.assert_any_call("stdout")
assert target.monitor is not None
target.add_abort_token("test")
assert fake_ffp.return_value.add_abort_token.call_count == 1
target.save_logs("fake_dest")
assert fake_ffp.return_value.save_logs.call_count == 1
target.cleanup()
assert fake_ffp.return_value.clean_up.call_count == 1
fake_ffp.RC_EXITED = FFPuppet.RC_EXITED
fake_ffp.RC_WORKER = FFPuppet.RC_WORKER
fake_file = tmp_path / "fake"
fake_file.touch()
target = PuppetTarget(str(fake_file), None, 300, 25, 5000, str(fake_file), 25)
# no failures
fake_ffp.return_value.is_healthy.return_value = True
fake_ffp.return_value.reason = None
assert target.detect_failure([], False) == Target.RESULT_NONE
assert target.detect_failure(["memory"], False) == Target.RESULT_NONE
assert not target.closed
# test close
fake_ffp.return_value.is_healthy.call_count = 0
fake_ffp.return_value.is_healthy.return_value = False
fake_ffp.return_value.is_running.return_value = False
fake_ffp.return_value.reason = FFPuppet.RC_CLOSED
assert target.detect_failure([], False) == Target.RESULT_NONE
assert fake_ffp.return_value.is_running.call_count == 1
assert fake_ffp.return_value.is_healthy.call_count == 1
assert fake_ffp.return_value.close.call_count == 1
# test single process crash
fake_ffp.return_value.close.call_count = 0
fake_ffp.return_value.is_healthy.return_value = False
fake_ffp.return_value.is_running.return_value = False
fake_ffp.return_value.reason = FFPuppet.RC_ALERT
assert target.detect_failure([], False) == Target.RESULT_FAILURE
assert fake_ffp.return_value.close.call_count == 1
# test multiprocess crash
fake_ffp.return_value.close.call_count = 0
fake_ffp.return_value.is_healthy.return_value = False
fake_ffp.return_value.is_running.return_value = True
fake_ffp.return_value.reason = FFPuppet.RC_ALERT
if self.expect_close and not was_timeout:
# give the browser a moment to close if needed
self._puppet.wait(timeout=30)
is_healthy = self._puppet.is_healthy()
# check if there has been a crash, hang, etc...
if not is_healthy or was_timeout:
if self._puppet.is_running():
log.debug("terminating browser...")
if was_timeout and "timeout" not in ignored and platform.system() == "Linux":
self._abort_hung_proc()
# give the process a moment to start dump
self._puppet.wait(timeout=1)
self.close()
# if something has happened figure out what
if not is_healthy:
if self._puppet.reason == FFPuppet.RC_CLOSED:
log.info("target.close() was called")
elif self._puppet.reason == FFPuppet.RC_EXITED:
log.info("Target closed itself")
elif (self._puppet.reason == FFPuppet.RC_WORKER
and "memory" in ignored
and "ffp_worker_memory_usage" in self._puppet.available_logs()):
status = self.RESULT_IGNORED
log.info("Memory limit exceeded")
elif (self._puppet.reason == FFPuppet.RC_WORKER
and "log-limit" in ignored
and "ffp_worker_log_size" in self._puppet.available_logs()):
status = self.RESULT_IGNORED
log.info("Log size limit exceeded")
else:
log.debug("failure detected, ffpuppet return code: %r", self._puppet.reason)
status = self.RESULT_FAILURE