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()