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_logger_09(tmp_path):
"""test PuppetLogger.clean_up() with inaccessible directory"""
with PuppetLogger(base_path=str(tmp_path)) as plog:
plog.add_log("test")
os.chmod(plog.working_path, stat.S_IRUSR)
working_path = plog.working_path
plog.close()
try:
assert os.path.isdir(plog.working_path)
with pytest.raises(OSError):
plog.clean_up()
assert plog.working_path is not None
plog.clean_up(ignore_errors=True)
assert plog.working_path is None
assert os.path.isdir(working_path)
finally:
os.chmod(working_path, stat.S_IRWXU)
def test_puppet_logger_08(tmp_path):
"""test PuppetLogger.add_log() with file not on disk"""
with PuppetLogger(base_path=str(tmp_path)) as plog:
with tempfile.SpooledTemporaryFile(max_size=2048) as log_fp:
plog.add_log("test", logfp=log_fp)
with pytest.raises(IOError, match="log file None does not exist"):
plog.get_fp("test")
def test_puppet_logger_04(tmp_path):
"""test PuppetLogger.reset()"""
with PuppetLogger(base_path=str(tmp_path)) as plog:
plog.add_log("test_new")
plog.clean_up()
plog.reset()
assert not plog.closed
assert not plog._logs
assert os.path.isdir(plog.working_path)
assert plog._base is not None
assert len(os.listdir(plog._base)) == 1
def test_puppet_logger_05(tmp_path):
"""test PuppetLogger.clone_log()"""
with PuppetLogger(base_path=str(tmp_path)) as plog:
plog.add_log("test_empty")
plog.add_log("test_extra")
plog.get_fp("test_extra").write(b"stuff")
plog.get_fp("test_extra").flush()
# test clone
plog.add_log("test_new")
pl_fp = plog.get_fp("test_new")
pl_fp.write(b"test1")
cloned = plog.clone_log("test_new")
assert os.path.isfile(cloned)
with open(cloned, "rb") as log_fp:
assert log_fp.read() == b"test1"
# test target exists
target = tmp_path / "target.txt"
target.touch()
pl_fp.write(b"test2")
def test_puppet_logger_03(tmp_path):
"""test PuppetLogger.clean_up()"""
with PuppetLogger(base_path=str(tmp_path)) as plog:
assert not plog.closed
assert not plog._logs
assert plog.working_path is not None
assert os.path.isdir(plog.working_path)
assert plog._base is not None
assert os.listdir(plog._base)
plog.add_log("test_new")
plog.clean_up()
assert plog.closed
assert not os.listdir(plog._base)
assert plog.working_path is None
assert plog.closed
assert not plog._logs
def test_puppet_logger_06(tmp_path):
"""test PuppetLogger.save_logs()"""
with PuppetLogger(base_path=str(tmp_path)) as plog:
plog.close()
# save when there are no logs
dest = tmp_path / "dest"
plog.save_logs(str(dest))
assert not any(dest.glob("*"))
plog.reset()
dest.rmdir()
# add small log
plog.add_log("test_1")
plog.get_fp("test_1").write(b"test1\ntest1\n")
# add binary data in log
plog.add_log("test_2")
plog.get_fp("test_2").write(b"\x00TEST\xFF\xEF")
# add empty log
plog.add_log("test_empty")
# add larger log (not a power of 2 to help catch buffer issues)
unprocessed = os.path.join(self.logs, "unprocessed")
os.mkdir(unprocessed)
with open(os.path.join(self.logs, "log_logcat.txt"), "wb") as log_fp:
# TODO: should this filter by pid or not?
log_fp.write(self._session.collect_logs())
#log_fp.write(self._session.collect_logs(pid=self._pid))
self._split_logcat(self.logs, self._package)
if not crash_reports:
return
# copy crash logs from the device
for fname in crash_reports:
self._session.pull(fname, unprocessed)
logger = PuppetLogger()
try:
syms_path = self._session.symbols_path(self._package)
process_minidumps(unprocessed, syms_path, logger.add_log)
logger.close()
logger.save_logs(self.logs)
finally:
logger.clean_up()
def __init__(self, use_profile=None, use_valgrind=False, use_xvfb=False, use_gdb=False, use_rr=False):
self._abort_tokens = set() # tokens used to notify log scanner to kill the browser process
self._checks = list()
self._last_bin_path = None
self._launches = 0 # number of successful browser launches
self._logs = PuppetLogger()
self._proc = None
self._profile_template = use_profile # profile that is used as a template
self._use_valgrind = use_valgrind
self._use_gdb = use_gdb
self._use_rr = use_rr
self._xvfb = None
self.profile = None # path to profile
self.reason = self.RC_CLOSED # why the target process was terminated
plat = platform.system().lower()
if use_valgrind:
assert not (use_gdb or use_rr), "only a single debugger can be enabled"
if not plat.startswith("linux"):
raise EnvironmentError("Valgrind is only supported on Linux")
try:
match = re.match(
"""
Add a log file to the log manager.
@type log_id: String
@param log_id: ID of the log to add.
@type logfp: file
@param logfp: logfp is a file object. If None is provided a new log file will be created.
@rtype: file
@return: file object of the newly added log file.
"""
assert log_id not in self._logs
assert not self.closed
if logfp is None:
logfp = PuppetLogger.open_unique(base_dir=self.working_path)
self._logs[log_id] = logfp
return logfp