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_02(self):
"test CheckLogContents()"
# input contains token
with open(self.tmpfn, "w") as in_fp:
in_fp.write("blah\nfoo\ntest\n123")
checker = CheckLogContents([self.tmpfn], [re.compile("test")])
self.assertTrue(checker.check())
with open(self.tmpfn, "wb") as log_fp:
checker.dump_log(log_fp)
self.assertGreater(log_fp.tell(), 1)
# input does not contains token
checker = CheckLogContents([self.tmpfn], [re.compile("no_token")])
self.assertFalse(checker.check())
# check a 2nd time
self.assertFalse(checker.check())
with open(self.tmpfn, "wb") as log_fp:
checker.dump_log(log_fp)
self.assertEqual(log_fp.tell(), 0)
# log does not exist
checker = CheckLogContents(["missing_log"], [re.compile("no_token")])
self.assertFalse(checker.check())
with open(self.tmpfn, "wb") as log_fp:
checker = CheckLogContents([self.tmpfn], [re.compile("no_token")])
self.assertFalse(checker.check())
# check a 2nd time
self.assertFalse(checker.check())
with open(self.tmpfn, "wb") as log_fp:
checker.dump_log(log_fp)
self.assertEqual(log_fp.tell(), 0)
# log does not exist
checker = CheckLogContents(["missing_log"], [re.compile("no_token")])
self.assertFalse(checker.check())
with open(self.tmpfn, "wb") as log_fp:
checker.dump_log(log_fp)
self.assertEqual(log_fp.tell(), 0)
# input exceeds chunk_size
try:
CheckLogContents.chunk_size = CheckLogContents.buf_limit
with open(self.tmpfn, "w") as in_fp:
in_fp.write("A" * (CheckLogContents.buf_limit - 2))
in_fp.write("test123")
in_fp.write("A" * 20)
checker = CheckLogContents([self.tmpfn], [re.compile("test123")])
self.assertFalse(checker.check())
self.assertTrue(checker.check())
finally:
CheckLogContents.chunk_size = 0x20000
with open(self.tmpfn, "wb") as log_fp:
checker.dump_log(log_fp)
self.assertGreater(log_fp.tell(), 1)
checker.dump_log(log_fp)
self.assertEqual(log_fp.tell(), 0)
# log does not exist
checker = CheckLogContents(["missing_log"], [re.compile("no_token")])
self.assertFalse(checker.check())
with open(self.tmpfn, "wb") as log_fp:
checker.dump_log(log_fp)
self.assertEqual(log_fp.tell(), 0)
# input exceeds chunk_size
try:
CheckLogContents.chunk_size = CheckLogContents.buf_limit
with open(self.tmpfn, "w") as in_fp:
in_fp.write("A" * (CheckLogContents.buf_limit - 2))
in_fp.write("test123")
in_fp.write("A" * 20)
checker = CheckLogContents([self.tmpfn], [re.compile("test123")])
self.assertFalse(checker.check())
self.assertTrue(checker.check())
finally:
CheckLogContents.chunk_size = 0x20000
with open(self.tmpfn, "wb") as log_fp:
checker.dump_log(log_fp)
self.assertGreater(log_fp.tell(), 1)
log.debug("launched firefox with pid: %d", self._proc.pid)
bootstrapper.wait(self.is_healthy, timeout=launch_timeout, url=location)
finally:
bootstrapper.close()
if prefs_js is not None and os.path.isfile(os.path.join(self.profile, "Invalidprefs.js")):
raise InvalidPrefs("%r is invalid" % prefs_js)
if log_limit:
self._checks.append(CheckLogSize(
log_limit,
self._logs.get_fp("stderr").name,
self._logs.get_fp("stdout").name))
if memory_limit:
self._checks.append(CheckMemoryUsage(self.get_pid(), memory_limit))
if self._abort_tokens:
self._checks.append(CheckLogContents(
[self._logs.get_fp("stderr").name, self._logs.get_fp("stdout").name],
self._abort_tokens))
self._launches += 1
def __init__(self, log_files, search_tokens):
assert log_files, "log_files is empty"
assert search_tokens, "search_tokens is empty"
super(CheckLogContents, self).__init__()
self.logs = list()
for log_file in log_files:
self.logs.append({"fname": log_file, "buffer": "", "offset": 0})
self.tokens = search_tokens