import argparse import json import optparse import os import sys import unittest from io import StringIO import mozfile from mozlog import commandline, formatters, handlers, reader, stdadapter, structuredlog class Handler: def __init__(self): self.items = [] def __call__(self, data): self.items.append(data) @property def last_item(self): return self.items[-1] @property def empty(self): return not self.items class BaseStructuredTest(unittest.TestCase): def setUp(self): self.logger = structuredlog.StructuredLogger("test") self.handler = Handler() self.logger.add_handler(self.handler) def pop_last_item(self): return self.handler.items.pop() def assert_log_equals(self, expected, actual=None): if actual is None: actual = self.pop_last_item() all_expected = {"pid": os.getpid(), "thread": "MainThread", "source": "test"} specials = set(["time"]) all_expected.update(expected) for key, value in all_expected.items(): self.assertEqual(actual[key], value) self.assertEqual(set(all_expected.keys()) | specials, set(actual.keys())) class TestStatusHandler(BaseStructuredTest): def setUp(self): super().setUp() self.handler = handlers.StatusHandler() self.logger.add_handler(self.handler) def test_failure_run(self): self.logger.suite_start([]) self.logger.test_start("test1") self.logger.test_status("test1", "sub1", status="PASS") self.logger.test_status("test1", "sub2", status="TIMEOUT") self.logger.test_status( "test1", "sub3", status="FAIL", expected="PASS", known_intermittent=["FAIL"] ) self.logger.test_end("test1", status="OK") self.logger.suite_end() summary = self.handler.summarize() self.assertIn("TIMEOUT", summary.unexpected_statuses) self.assertEqual(1, summary.unexpected_statuses["TIMEOUT"]) self.assertIn("PASS", summary.expected_statuses) self.assertEqual(1, summary.expected_statuses["PASS"]) self.assertIn("OK", summary.expected_statuses) self.assertEqual(1, summary.expected_statuses["OK"]) self.assertIn("FAIL", summary.expected_statuses) self.assertEqual(1, summary.expected_statuses["FAIL"]) self.assertIn("FAIL", summary.known_intermittent_statuses) self.assertEqual(1, summary.known_intermittent_statuses["FAIL"]) self.assertEqual(3, summary.action_counts["test_status"]) self.assertEqual(1, summary.action_counts["test_end"]) def test_precondition_failed_run(self): self.logger.suite_start([]) self.logger.test_start("test1") self.logger.test_end("test1", status="PRECONDITION_FAILED") self.logger.test_start("test2") self.logger.test_status("test2", "sub1", status="PRECONDITION_FAILED") self.logger.test_end("test2", status="OK") self.logger.suite_end() summary = self.handler.summarize() self.assertEqual(1, summary.expected_statuses["OK"]) self.assertEqual(2, summary.unexpected_statuses["PRECONDITION_FAILED"]) def test_error_run(self): self.logger.suite_start([]) self.logger.test_start("test1") self.logger.error("ERRR!") self.logger.test_end("test1", status="OK") self.logger.test_start("test2") self.logger.test_end("test2", status="OK") self.logger.suite_end() summary = self.handler.summarize() self.assertIn("ERROR", summary.log_level_counts) self.assertEqual(1, summary.log_level_counts["ERROR"]) self.assertIn("OK", summary.expected_statuses) self.assertEqual(2, summary.expected_statuses["OK"]) def test_crash_with_expected_crash_status(self): # Test that crashes are accounted for when test ends with expected CRASH status self.logger.suite_start([]) self.logger.test_start("test1") self.logger.crash( test="test1", process=1234, signature="test_signature", minidump_path="/path/to/dump", ) self.logger.crash( test="test1", process=5678, signature="test_signature2", minidump_path="/path/to/dump2", ) self.logger.test_end("test1", "CRASH", expected="CRASH") self.logger.suite_end() summary = self.handler.summarize() # Extra crashes were subtracted, keeping 1 to match the CRASH status self.assertEqual(1, summary.action_counts.get("crash", 0)) # Test had expected CRASH status self.assertEqual(1, summary.expected_statuses["CRASH"]) def test_crash_with_retry_pass(self): # Simulates xpcshell retry: test crashes producing 2 dumps, then passes on retry self.logger.suite_start([]) # First run: test crashes self.logger.test_start("test1") self.logger.crash( test="test1", process=1234, signature="test_signature", minidump_path="/path/to/dump", ) self.logger.crash( test="test1", process=5678, signature="test_signature2", minidump_path="/path/to/dump2", ) self.logger.test_end("test1", "CRASH", expected="CRASH") # Retry: test passes self.logger.test_start("test1") self.logger.test_end("test1", "PASS", expected="PASS") self.logger.suite_end() summary = self.handler.summarize() # Extra crashes were subtracted, keeping 1 to match the CRASH status self.assertEqual(1, summary.action_counts.get("crash", 0)) # Both test runs are counted self.assertEqual(1, summary.expected_statuses["CRASH"]) self.assertEqual(1, summary.expected_statuses["PASS"]) def test_crash_without_test_name(self): # Orphaned crash (e.g., shutdown crash) without associated test self.logger.suite_start([]) self.logger.test_start("test1") self.logger.test_end("test1", "PASS", expected="PASS") self.logger.crash( process=9999, signature="shutdown_crash", minidump_path="/path/to/dump", ) self.logger.suite_end() summary = self.handler.summarize() # Crash without test name remains unaccounted self.assertEqual(1, summary.action_counts["crash"]) self.assertEqual(1, summary.expected_statuses["PASS"]) class TestSummaryHandler(BaseStructuredTest): def setUp(self): super().setUp() self.handler = handlers.SummaryHandler() self.logger.add_handler(self.handler) def test_failure_run(self): self.logger.suite_start([]) self.logger.test_start("test1") self.logger.test_status("test1", "sub1", status="PASS") self.logger.test_status("test1", "sub2", status="TIMEOUT") self.logger.assertion_count("test1", 5, 1, 10) self.logger.assertion_count("test1", 5, 10, 15) self.logger.test_end("test1", status="OK") self.logger.suite_end() counts = self.handler.current["counts"] self.assertIn("timeout", counts["subtest"]["unexpected"]) self.assertEqual(1, counts["subtest"]["unexpected"]["timeout"]) self.assertIn("pass", counts["subtest"]["expected"]) self.assertEqual(1, counts["subtest"]["expected"]["pass"]) self.assertIn("ok", counts["test"]["expected"]) self.assertEqual(1, counts["test"]["expected"]["ok"]) self.assertIn("pass", counts["assert"]["unexpected"]) self.assertEqual(1, counts["assert"]["unexpected"]["pass"]) self.assertIn("fail", counts["assert"]["expected"]) self.assertEqual(1, counts["assert"]["expected"]["fail"]) logs = self.handler.current["unexpected_logs"] self.assertEqual(1, len(logs)) self.assertIn("test1", logs) self.assertEqual(1, len(logs["test1"])) self.assertEqual("sub2", logs["test1"][0]["subtest"]) def test_precondition_failed_run(self): self.logger.suite_start([]) self.logger.test_start("test1") self.logger.test_status("test1", "sub1", status="PASS") self.logger.test_end("test1", status="PRECONDITION_FAILED") self.logger.test_start("test2") self.logger.test_status("test2", "sub1", status="PRECONDITION_FAILED") self.logger.test_status("test2", "sub2", status="PRECONDITION_FAILED") self.logger.test_end("test2", status="OK") self.logger.suite_end() counts = self.handler.current["counts"] self.assertIn("precondition_failed", counts["test"]["unexpected"]) self.assertEqual(1, counts["test"]["unexpected"]["precondition_failed"]) self.assertIn("pass", counts["subtest"]["expected"]) self.assertEqual(1, counts["subtest"]["expected"]["pass"]) self.assertIn("ok", counts["test"]["expected"]) self.assertEqual(1, counts["test"]["expected"]["ok"]) self.assertIn("precondition_failed", counts["subtest"]["unexpected"]) self.assertEqual(2, counts["subtest"]["unexpected"]["precondition_failed"]) class TestStructuredLog(BaseStructuredTest): def test_suite_start(self): self.logger.suite_start(["test"], "logtest") self.assert_log_equals({ "action": "suite_start", "name": "logtest", "tests": {"default": ["test"]}, }) self.logger.suite_end() def test_suite_end(self): self.logger.suite_start([]) self.logger.suite_end() self.assert_log_equals({"action": "suite_end"}) def test_add_subsuite(self): self.logger.suite_start([]) self.logger.add_subsuite("other") self.assert_log_equals({ "action": "add_subsuite", "name": "other", "run_info": {"subsuite": "other"}, }) self.logger.suite_end() def test_add_subsuite_duplicate(self): self.logger.suite_start([]) self.logger.add_subsuite("other") # This should be a no-op self.logger.add_subsuite("other") self.assert_log_equals({ "action": "add_subsuite", "name": "other", "run_info": {"subsuite": "other"}, }) self.assert_log_equals({"action": "suite_start", "tests": {"default": []}}) self.logger.suite_end() def test_start(self): self.logger.suite_start([]) self.logger.test_start("test1") self.assert_log_equals({"action": "test_start", "test": "test1"}) self.logger.test_start(("test1", "==", "test1-ref"), path="path/to/test") self.assert_log_equals({ "action": "test_start", "test": ("test1", "==", "test1-ref"), "path": "path/to/test", }) self.logger.suite_end() def test_start_inprogress(self): self.logger.suite_start([]) self.logger.test_start("test1") self.logger.test_start("test1") self.assert_log_equals({ "action": "log", "message": "test_start for test1 logged while in progress.", "level": "ERROR", }) self.logger.suite_end() def test_start_inprogress_subsuite(self): self.logger.suite_start([]) self.logger.add_subsuite("other") self.logger.test_start("test1") self.logger.test_start("test1", subsuite="other") self.assert_log_equals({ "action": "test_start", "test": "test1", "subsuite": "other", }) self.logger.suite_end() def test_status(self): self.logger.suite_start([]) self.logger.test_start("test1") self.logger.test_status( "test1", "subtest name", "fail", expected="FAIL", message="Test message" ) self.assert_log_equals({ "action": "test_status", "subtest": "subtest name", "status": "FAIL", "message": "Test message", "test": "test1", }) self.logger.test_end("test1", "OK") self.logger.suite_end() def test_status_1(self): self.logger.suite_start([]) self.logger.test_start("test1") self.logger.test_status("test1", "subtest name", "fail") self.assert_log_equals({ "action": "test_status", "subtest": "subtest name", "status": "FAIL", "expected": "PASS", "test": "test1", }) self.logger.test_end("test1", "OK") self.logger.suite_end() def test_status_2(self): self.assertRaises( ValueError, self.logger.test_status, "test1", "subtest name", "XXXUNKNOWNXXX", ) def test_status_extra(self): self.logger.suite_start([]) self.logger.test_start("test1") self.logger.test_status( "test1", "subtest name", "FAIL", expected="PASS", extra={"data": 42} ) self.assert_log_equals({ "action": "test_status", "subtest": "subtest name", "status": "FAIL", "expected": "PASS", "test": "test1", "extra": {"data": 42}, }) self.logger.test_end("test1", "OK") self.logger.suite_end() def test_status_stack(self): self.logger.suite_start([]) self.logger.test_start("test1") self.logger.test_status( "test1", "subtest name", "FAIL", expected="PASS", stack="many\nlines\nof\nstack", ) self.assert_log_equals({ "action": "test_status", "subtest": "subtest name", "status": "FAIL", "expected": "PASS", "test": "test1", "stack": "many\nlines\nof\nstack", }) self.logger.test_end("test1", "OK") self.logger.suite_end() def test_status_known_intermittent(self): self.logger.suite_start([]) self.logger.test_start("test1") self.logger.test_status( "test1", "subtest name", "fail", known_intermittent=["FAIL"] ) self.assert_log_equals({ "action": "test_status", "subtest": "subtest name", "status": "FAIL", "expected": "PASS", "known_intermittent": ["FAIL"], "test": "test1", }) self.logger.test_end("test1", "OK") self.logger.suite_end() def test_status_not_started(self): self.logger.test_status("test_UNKNOWN", "subtest", "PASS") self.assertTrue( self.pop_last_item()["message"].startswith( "test_status for test_UNKNOWN logged while not in progress. Logged with data: {" ) ) def test_remove_optional_defaults(self): self.logger.suite_start([]) self.logger.test_start("test1") self.logger.test_status( "test1", "subtest name", "fail", message=None, stack=None ) self.assert_log_equals({ "action": "test_status", "subtest": "subtest name", "status": "FAIL", "expected": "PASS", "test": "test1", }) self.logger.test_end("test1", "OK") self.logger.suite_end() def test_remove_optional_defaults_raw_log(self): self.logger.log_raw({"action": "suite_start", "tests": [1], "name": None}) self.assert_log_equals({"action": "suite_start", "tests": {"default": ["1"]}}) self.logger.suite_end() def test_end(self): self.logger.suite_start([]) self.logger.test_start("test1") self.logger.test_end("test1", "fail", message="Test message") self.assert_log_equals({ "action": "test_end", "status": "FAIL", "expected": "OK", "message": "Test message", "test": "test1", }) self.logger.suite_end() def test_end_1(self): self.logger.suite_start([]) self.logger.test_start("test1") self.logger.test_end("test1", "PASS", expected="PASS", extra={"data": 123}) self.assert_log_equals({ "action": "test_end", "status": "PASS", "extra": {"data": 123}, "test": "test1", }) self.logger.suite_end() def test_end_2(self): self.assertRaises(ValueError, self.logger.test_end, "test1", "XXXUNKNOWNXXX") def test_end_stack(self): self.logger.suite_start([]) self.logger.test_start("test1") self.logger.test_end( "test1", "PASS", expected="PASS", stack="many\nlines\nof\nstack" ) self.assert_log_equals({ "action": "test_end", "status": "PASS", "test": "test1", "stack": "many\nlines\nof\nstack", }) self.logger.suite_end() def test_end_no_start(self): self.logger.test_end("test1", "PASS", expected="PASS") self.assertTrue( self.pop_last_item()["message"].startswith( "test_end for test1 logged while not in progress. Logged with data: {" ) ) self.logger.suite_end() def test_end_no_start_subsuite(self): self.logger.suite_start([]) self.logger.add_subsuite("other") self.logger.test_start("test1", subsuite="other") self.logger.test_end("test1", "PASS", expected="PASS") self.assertTrue( self.pop_last_item()["message"].startswith( "test_end for test1 logged while not in progress. Logged with data: {" ) ) self.logger.test_end("test1", "OK", subsuite="other") self.assert_log_equals({ "action": "test_end", "status": "OK", "test": "test1", "subsuite": "other", }) self.logger.suite_end() def test_end_twice(self): self.logger.suite_start([]) self.logger.test_start("test2") self.logger.test_end("test2", "PASS", expected="PASS") self.assert_log_equals({ "action": "test_end", "status": "PASS", "test": "test2", }) self.logger.test_end("test2", "PASS", expected="PASS") last_item = self.pop_last_item() self.assertEqual(last_item["action"], "log") self.assertEqual(last_item["level"], "ERROR") self.assertTrue( last_item["message"].startswith( "test_end for test2 logged while not in progress. Logged with data: {" ) ) self.logger.suite_end() def test_suite_start_twice(self): self.logger.suite_start([]) self.assert_log_equals({"action": "suite_start", "tests": {"default": []}}) self.logger.suite_start([]) last_item = self.pop_last_item() self.assertEqual(last_item["action"], "log") self.assertEqual(last_item["level"], "ERROR") self.logger.suite_end() def test_suite_end_no_start(self): self.logger.suite_start([]) self.assert_log_equals({"action": "suite_start", "tests": {"default": []}}) self.logger.suite_end() self.assert_log_equals({"action": "suite_end"}) self.logger.suite_end() last_item = self.pop_last_item() self.assertEqual(last_item["action"], "log") self.assertEqual(last_item["level"], "ERROR") def test_multiple_loggers_suite_start(self): logger1 = structuredlog.StructuredLogger("test") self.logger.suite_start([]) logger1.suite_start([]) last_item = self.pop_last_item() self.assertEqual(last_item["action"], "log") self.assertEqual(last_item["level"], "ERROR") def test_multiple_loggers_test_start(self): logger1 = structuredlog.StructuredLogger("test") self.logger.suite_start([]) self.logger.test_start("test") logger1.test_start("test") last_item = self.pop_last_item() self.assertEqual(last_item["action"], "log") self.assertEqual(last_item["level"], "ERROR") def test_process(self): self.logger.process_output(1234, "test output") self.assert_log_equals({ "action": "process_output", "process": "1234", "data": "test output", }) def test_process_start(self): self.logger.process_start(1234) self.assert_log_equals({"action": "process_start", "process": "1234"}) def test_process_exit(self): self.logger.process_exit(1234, 0) self.assert_log_equals({ "action": "process_exit", "process": "1234", "exitcode": 0, }) def test_log(self): for level in ["critical", "error", "warning", "info", "debug"]: getattr(self.logger, level)("message") self.assert_log_equals({ "action": "log", "level": level.upper(), "message": "message", }) def test_logging_adapter(self): import logging logging.basicConfig(level="DEBUG") old_level = logging.root.getEffectiveLevel() logging.root.setLevel("DEBUG") std_logger = logging.getLogger("test") std_logger.setLevel("DEBUG") logger = stdadapter.std_logging_adapter(std_logger) try: for level in ["critical", "error", "warning", "info", "debug"]: getattr(logger, level)("message") self.assert_log_equals({ "action": "log", "level": level.upper(), "message": "message", }) finally: logging.root.setLevel(old_level) def test_add_remove_handlers(self): handler = Handler() self.logger.add_handler(handler) self.logger.info("test1") self.assert_log_equals({"action": "log", "level": "INFO", "message": "test1"}) self.assert_log_equals( {"action": "log", "level": "INFO", "message": "test1"}, actual=handler.last_item, ) self.logger.remove_handler(handler) self.logger.info("test2") self.assert_log_equals({"action": "log", "level": "INFO", "message": "test2"}) self.assert_log_equals( {"action": "log", "level": "INFO", "message": "test1"}, actual=handler.last_item, ) def test_wrapper(self): file_like = structuredlog.StructuredLogFileLike(self.logger) file_like.write("line 1") self.assert_log_equals({"action": "log", "level": "INFO", "message": "line 1"}) file_like.write("line 2\n") self.assert_log_equals({"action": "log", "level": "INFO", "message": "line 2"}) file_like.write("line 3\r") self.assert_log_equals({"action": "log", "level": "INFO", "message": "line 3"}) file_like.write("line 4\r\n") self.assert_log_equals({"action": "log", "level": "INFO", "message": "line 4"}) def test_shutdown(self): # explicit shutdown log = structuredlog.StructuredLogger("test 1") log.add_handler(self.handler) log.info("line 1") self.assert_log_equals({ "action": "log", "level": "INFO", "message": "line 1", "source": "test 1", }) log.shutdown() self.assert_log_equals({"action": "shutdown", "source": "test 1"}) with self.assertRaises(structuredlog.LoggerShutdownError): log.info("bad log") with self.assertRaises(structuredlog.LoggerShutdownError): log.log_raw({"action": "log", "level": "info", "message": "bad log"}) # shutdown still applies to new instances del log log = structuredlog.StructuredLogger("test 1") with self.assertRaises(structuredlog.LoggerShutdownError): log.info("bad log") # context manager shutdown with structuredlog.StructuredLogger("test 2") as log: log.add_handler(self.handler) log.info("line 2") self.assert_log_equals({ "action": "log", "level": "INFO", "message": "line 2", "source": "test 2", }) self.assert_log_equals({"action": "shutdown", "source": "test 2"}) # shutdown prevents logging across instances log1 = structuredlog.StructuredLogger("test 3") log2 = structuredlog.StructuredLogger("test 3", component="bar") log1.shutdown() with self.assertRaises(structuredlog.LoggerShutdownError): log2.info("line 3") class TestTypeConversions(BaseStructuredTest): def test_raw(self): self.logger.log_raw({"action": "suite_start", "tests": [1], "time": "1234"}) self.assert_log_equals({ "action": "suite_start", "tests": {"default": ["1"]}, "time": 1234, }) self.logger.suite_end() def test_tuple(self): self.logger.suite_start([]) self.logger.test_start(( b"\xf0\x90\x8d\x84\xf0\x90\x8c\xb4\xf0\x90" b"\x8d\x83\xf0\x90\x8d\x84".decode(), 42, "\u16a4", )) self.assert_log_equals({ "action": "test_start", "test": ("\U00010344\U00010334\U00010343\U00010344", "42", "\u16a4"), }) self.logger.suite_end() def test_non_string_messages(self): self.logger.suite_start([]) self.logger.info(1) self.assert_log_equals({"action": "log", "message": "1", "level": "INFO"}) self.logger.info([1, (2, "3"), "s", "s" + chr(255)]) self.assert_log_equals({ "action": "log", "message": "[1, (2, '3'), 's', 's\xff']", "level": "INFO", }) self.logger.suite_end() def test_utf8str_write(self): with mozfile.NamedTemporaryFile() as logfile: _fmt = formatters.TbplFormatter() _handler = handlers.StreamHandler(logfile, _fmt) self.logger.add_handler(_handler) self.logger.suite_start([]) self.logger.info("☺") logfile.seek(0) data = logfile.readlines()[-1].strip() self.assertEqual(data.decode(), "☺") self.logger.suite_end() self.logger.remove_handler(_handler) def test_arguments(self): self.logger.info(message="test") self.assert_log_equals({"action": "log", "message": "test", "level": "INFO"}) self.logger.suite_start([], run_info={}) self.assert_log_equals({ "action": "suite_start", "tests": {"default": []}, "run_info": {}, }) self.logger.test_start(test="test1") self.logger.test_status("subtest1", "FAIL", test="test1", status="PASS") self.assert_log_equals({ "action": "test_status", "test": "test1", "subtest": "subtest1", "status": "PASS", "expected": "FAIL", }) self.logger.process_output(123, "data", "test") self.assert_log_equals({ "action": "process_output", "process": "123", "command": "test", "data": "data", }) self.assertRaises( TypeError, self.logger.test_status, subtest="subtest2", status="FAIL", expected="PASS", ) self.assertRaises( TypeError, self.logger.test_status, "test1", "subtest1", "group1", "PASS", "FAIL", "message", "stack", {}, [], None, "unexpected", ) self.assertRaises(TypeError, self.logger.test_status, "test1", test="test2") self.logger.suite_end() class TestComponentFilter(BaseStructuredTest): def test_filter_component(self): component_logger = structuredlog.StructuredLogger( self.logger.name, "test_component" ) component_logger.component_filter = handlers.LogLevelFilter(lambda x: x, "info") self.logger.debug("Test") self.assertFalse(self.handler.empty) self.assert_log_equals({"action": "log", "level": "DEBUG", "message": "Test"}) self.assertTrue(self.handler.empty) component_logger.info("Test 1") self.assertFalse(self.handler.empty) self.assert_log_equals({ "action": "log", "level": "INFO", "message": "Test 1", "component": "test_component", }) component_logger.debug("Test 2") self.assertTrue(self.handler.empty) component_logger.component_filter = None component_logger.debug("Test 3") self.assertFalse(self.handler.empty) self.assert_log_equals({ "action": "log", "level": "DEBUG", "message": "Test 3", "component": "test_component", }) def test_filter_default_component(self): component_logger = structuredlog.StructuredLogger( self.logger.name, "test_component" ) self.logger.debug("Test") self.assertFalse(self.handler.empty) self.assert_log_equals({"action": "log", "level": "DEBUG", "message": "Test"}) self.logger.component_filter = handlers.LogLevelFilter(lambda x: x, "info") self.logger.debug("Test 1") self.assertTrue(self.handler.empty) component_logger.debug("Test 2") self.assertFalse(self.handler.empty) self.assert_log_equals({ "action": "log", "level": "DEBUG", "message": "Test 2", "component": "test_component", }) self.logger.component_filter = None self.logger.debug("Test 3") self.assertFalse(self.handler.empty) self.assert_log_equals({"action": "log", "level": "DEBUG", "message": "Test 3"}) def test_filter_message_mutuate(self): def filter_mutate(msg): if msg["action"] == "log": msg["message"] = "FILTERED! %s" % msg["message"] return msg self.logger.component_filter = filter_mutate self.logger.debug("Test") self.assert_log_equals({ "action": "log", "level": "DEBUG", "message": "FILTERED! Test", }) self.logger.component_filter = None class TestCommandline(unittest.TestCase): def log_lines(self, log_file): log_file.seek(0) return [line.rstrip() for line in log_file.readlines()] def test_setup_logging(self): parser = argparse.ArgumentParser() commandline.add_logging_group(parser) args = parser.parse_args(["--log-raw=-"]) logger = commandline.setup_logging("test_setup_logging", args, {}) self.assertEqual(len(logger.handlers), 1) def test_setup_logging_optparse(self): parser = optparse.OptionParser() commandline.add_logging_group(parser) args, _ = parser.parse_args(["--log-raw=-"]) logger = commandline.setup_logging("test_optparse", args, {}) self.assertEqual(len(logger.handlers), 1) self.assertIsInstance(logger.handlers[0], handlers.StreamHandler) def test_limit_formatters(self): parser = argparse.ArgumentParser() commandline.add_logging_group(parser, include_formatters=["raw"]) other_formatters = [fmt for fmt in commandline.log_formatters if fmt != "raw"] # check that every formatter except raw is not present for fmt in other_formatters: with self.assertRaises(SystemExit): parser.parse_args(["--log-%s=-" % fmt]) with self.assertRaises(SystemExit): parser.parse_args(["--log-%s-level=error" % fmt]) # raw is still ok args = parser.parse_args(["--log-raw=-"]) logger = commandline.setup_logging("test_setup_logging2", args, {}) self.assertEqual(len(logger.handlers), 1) def test_setup_logging_optparse_unicode(self): parser = optparse.OptionParser() commandline.add_logging_group(parser) args, _ = parser.parse_args(["--log-raw=-"]) logger = commandline.setup_logging("test_optparse_unicode", args, {}) self.assertEqual(len(logger.handlers), 1) self.assertEqual(logger.handlers[0].stream, sys.stdout) self.assertIsInstance(logger.handlers[0], handlers.StreamHandler) def test_logging_defaultlevel(self): parser = argparse.ArgumentParser() commandline.add_logging_group(parser) with mozfile.NamedTemporaryFile() as log_file: args = parser.parse_args(["--log-tbpl=%s" % log_file.name]) logger = commandline.setup_logging("test_fmtopts", args, {}) logger.info("INFO message") logger.debug("DEBUG message") logger.error("ERROR message") # The debug level is not logged by default. self.assertEqual( [b"INFO message", b"ERROR message"], self.log_lines(log_file) ) logger.handlers[0].stream.close() def test_logging_errorlevel(self): parser = argparse.ArgumentParser() commandline.add_logging_group(parser) with mozfile.NamedTemporaryFile() as log_file: args = parser.parse_args([ "--log-tbpl=%s" % log_file.name, "--log-tbpl-level=error", ]) logger = commandline.setup_logging("test_fmtopts", args, {}) logger.info("INFO message") logger.debug("DEBUG message") logger.error("ERROR message") # Only the error level and above were requested. self.assertEqual([b"ERROR message"], self.log_lines(log_file)) logger.handlers[0].stream.close() def test_logging_debuglevel(self): parser = argparse.ArgumentParser() commandline.add_logging_group(parser) with mozfile.NamedTemporaryFile() as log_file: args = parser.parse_args([ "--log-tbpl=%s" % log_file.name, "--log-tbpl-level=debug", ]) logger = commandline.setup_logging("test_fmtopts", args, {}) logger.info("INFO message") logger.debug("DEBUG message") logger.error("ERROR message") # Requesting a lower log level than default works as expected. self.assertEqual( [b"INFO message", b"DEBUG message", b"ERROR message"], self.log_lines(log_file), ) logger.handlers[0].stream.close() def test_unused_options(self): parser = argparse.ArgumentParser() commandline.add_logging_group(parser) args = parser.parse_args(["--log-tbpl-level=error"]) self.assertRaises( ValueError, commandline.setup_logging, "test_fmtopts", args, {} ) class TestBuffer(BaseStructuredTest): def assert_log_equals(self, expected, actual=None): if actual is None: actual = self.pop_last_item() all_expected = { "pid": os.getpid(), "thread": "MainThread", "source": "testBuffer", } specials = set(["time"]) all_expected.update(expected) for key, value in all_expected.items(): self.assertEqual(actual[key], value) self.assertEqual(set(all_expected.keys()) | specials, set(actual.keys())) def setUp(self): self.logger = structuredlog.StructuredLogger("testBuffer") self.handler = handlers.BufferHandler(Handler(), message_limit=4) self.logger.add_handler(self.handler) def tearDown(self): self.logger.remove_handler(self.handler) def pop_last_item(self): return self.handler.inner.items.pop() def test_buffer_messages(self): self.logger.suite_start([]) self.logger.test_start("test1") self.logger.send_message("buffer", "off") self.logger.test_status("test1", "sub1", status="PASS") # Even for buffered actions, the buffer does not interfere if # buffering is turned off. self.assert_log_equals({ "action": "test_status", "test": "test1", "status": "PASS", "subtest": "sub1", }) self.logger.send_message("buffer", "on") self.logger.test_status("test1", "sub2", status="PASS") self.logger.test_status("test1", "sub3", status="PASS") self.logger.test_status("test1", "sub4", status="PASS") self.logger.test_status("test1", "sub5", status="PASS") self.logger.test_status("test1", "sub6", status="PASS") self.logger.test_status("test1", "sub7", status="PASS") self.logger.test_end("test1", status="OK") self.logger.send_message("buffer", "clear") self.assert_log_equals({"action": "test_end", "test": "test1", "status": "OK"}) self.logger.suite_end() def test_buffer_size(self): self.logger.suite_start([]) self.logger.test_start("test1") self.logger.test_status("test1", "sub1", status="PASS") self.logger.test_status("test1", "sub2", status="PASS") self.logger.test_status("test1", "sub3", status="PASS") self.logger.test_status("test1", "sub4", status="PASS") self.logger.test_status("test1", "sub5", status="PASS") self.logger.test_status("test1", "sub6", status="PASS") self.logger.test_status("test1", "sub7", status="PASS") # No test status messages made it to the underlying handler. self.assert_log_equals({"action": "test_start", "test": "test1"}) # The buffer's actual size never grows beyond the specified limit. self.assertEqual(len(self.handler._buffer), 4) self.logger.test_status("test1", "sub8", status="FAIL") # The number of messages deleted comes back in a list. self.assertEqual([4], self.logger.send_message("buffer", "flush")) # When the buffer is dumped, the failure is the last thing logged self.assert_log_equals({ "action": "test_status", "test": "test1", "subtest": "sub8", "status": "FAIL", "expected": "PASS", }) # Three additional messages should have been retained for context self.assert_log_equals({ "action": "test_status", "test": "test1", "status": "PASS", "subtest": "sub7", }) self.assert_log_equals({ "action": "test_status", "test": "test1", "status": "PASS", "subtest": "sub6", }) self.assert_log_equals({ "action": "test_status", "test": "test1", "status": "PASS", "subtest": "sub5", }) self.assert_log_equals({"action": "suite_start", "tests": {"default": []}}) class TestReader(unittest.TestCase): def to_file_like(self, obj): data_str = "\n".join(json.dumps(item) for item in obj) return StringIO(data_str) def test_read(self): data = [ {"action": "action_0", "data": "data_0"}, {"action": "action_1", "data": "data_1"}, ] f = self.to_file_like(data) self.assertEqual(data, list(reader.read(f))) def test_imap_log(self): data = [ {"action": "action_0", "data": "data_0"}, {"action": "action_1", "data": "data_1"}, ] f = self.to_file_like(data) def f_action_0(item): return ("action_0", item["data"]) def f_action_1(item): return ("action_1", item["data"]) res_iter = reader.imap_log( reader.read(f), {"action_0": f_action_0, "action_1": f_action_1} ) self.assertEqual( [("action_0", "data_0"), ("action_1", "data_1")], list(res_iter) ) def test_each_log(self): data = [ {"action": "action_0", "data": "data_0"}, {"action": "action_1", "data": "data_1"}, ] f = self.to_file_like(data) count = {"action_0": 0, "action_1": 0} def f_action_0(item): count[item["action"]] += 1 def f_action_1(item): count[item["action"]] += 2 reader.each_log( reader.read(f), {"action_0": f_action_0, "action_1": f_action_1} ) self.assertEqual({"action_0": 1, "action_1": 2}, count) def test_handler(self): data = [ {"action": "action_0", "data": "data_0"}, {"action": "action_1", "data": "data_1"}, ] f = self.to_file_like(data) test = self class ReaderTestHandler(reader.LogHandler): def __init__(self): self.action_0_count = 0 self.action_1_count = 0 def action_0(self, item): test.assertEqual(item["action"], "action_0") self.action_0_count += 1 def action_1(self, item): test.assertEqual(item["action"], "action_1") self.action_1_count += 1 handler = ReaderTestHandler() reader.handle_log(reader.read(f), handler) self.assertEqual(handler.action_0_count, 1) self.assertEqual(handler.action_1_count, 1) if __name__ == "__main__": import mozunit mozunit.main()