"""
Tests for L{eliot.testing}.
"""

from unittest import SkipTest, TestResult, TestCase, skipUnless

try:
    import numpy as np
except ImportError:
    np = None

from ..testing import (
    issuperset,
    assertContainsFields,
    LoggedAction,
    LoggedMessage,
    validateLogging,
    UnflushedTracebacks,
    assertHasMessage,
    assertHasAction,
    validate_logging,
    capture_logging,
    swap_logger,
    check_for_errors,
)
from .._output import MemoryLogger
from .._action import start_action
from .._message import Message
from .._validation import ActionType, MessageType, ValidationError, Field
from .._traceback import write_traceback
from .. import add_destination, remove_destination, _output, log_message
from .common import CustomObject, CustomJSONEncoder


class IsSuperSetTests(TestCase):
    """
    Tests for L{issuperset}.
    """

    def test_equal(self):
        """
        Equal dictionaries are supersets of each other.
        """
        a = {"a": 1}
        b = a.copy()
        self.assertTrue(issuperset(a, b))

    def test_additionalIsSuperSet(self):
        """
        If C{A} is C{B} plus some extra entries, C{A} is superset of C{B}.
        """
        a = {"a": 1, "b": 2, "c": 3}
        b = {"a": 1, "c": 3}
        self.assertTrue(issuperset(a, b))

    def test_missingIsNotSuperSet(self):
        """
        If C{A} is C{B} minus some entries, C{A} is not a superset of C{B}.
        """
        a = {"a": 1, "c": 3}
        b = {"a": 1, "b": 2, "c": 3}
        self.assertFalse(issuperset(a, b))


class LoggedActionTests(TestCase):
    """
    Tests for L{LoggedAction}.
    """

    def test_values(self):
        """
        The values given to the L{LoggedAction} constructor are stored on it.
        """
        d1 = {"x": 1}
        d2 = {"y": 2}
        root = LoggedAction(d1, d2, [])
        self.assertEqual((root.startMessage, root.endMessage), (d1, d2))

    def fromMessagesIndex(self, messages, index):
        """
        Call L{LoggedAction.fromMessages} using action specified by index in
        a list of message dictionaries.

        @param messages: A C{list} of message dictionaries.

        @param index: Index to the logger's messages.

        @return: Result of L{LoggedAction.fromMessages}.
        """
        uuid = messages[index]["task_uuid"]
        level = messages[index]["task_level"]
        return LoggedAction.fromMessages(uuid, level, messages)

    def test_fromMessagesCreatesLoggedAction(self):
        """
        L{LoggedAction.fromMessages} returns a L{LoggedAction}.
        """
        logger = MemoryLogger()
        with start_action(logger, "test"):
            pass
        logged = self.fromMessagesIndex(logger.messages, 0)
        self.assertIsInstance(logged, LoggedAction)

    def test_fromMessagesStartAndSuccessfulFinish(self):
        """
        L{LoggedAction.fromMessages} finds the start and successful finish
        messages of an action and stores them in the result.
        """
        logger = MemoryLogger()
        Message.new(x=1).write(logger)
        with start_action(logger, "test"):
            Message.new(x=1).write(logger)
        # Now we should have x message, start action message, another x message
        # and finally finish message.
        logged = self.fromMessagesIndex(logger.messages, 1)
        self.assertEqual(
            (logged.startMessage, logged.endMessage),
            (logger.messages[1], logger.messages[3]),
        )

    def test_fromMessagesStartAndErrorFinish(self):
        """
        L{LoggedAction.fromMessages} finds the start and successful finish
        messages of an action and stores them in the result.
        """
        logger = MemoryLogger()
        try:
            with start_action(logger, "test"):
                raise KeyError()
        except KeyError:
            pass
        logged = self.fromMessagesIndex(logger.messages, 0)
        self.assertEqual(
            (logged.startMessage, logged.endMessage),
            (logger.messages[0], logger.messages[1]),
        )

    def test_fromMessagesStartNotFound(self):
        """
        L{LoggedAction.fromMessages} raises a L{ValueError} if a start message
        is not found.
        """
        logger = MemoryLogger()
        with start_action(logger, action_type="test"):
            pass
        self.assertRaises(ValueError, self.fromMessagesIndex, logger.messages[1:], 0)

    def test_fromMessagesFinishNotFound(self):
        """
        L{LoggedAction.fromMessages} raises a L{ValueError} if a finish message
        is not found.
        """
        logger = MemoryLogger()
        with start_action(logger, action_type="test"):
            pass
        with self.assertRaises(ValueError) as cm:
            self.fromMessagesIndex(logger.messages[:1], 0)
        self.assertEqual(cm.exception.args[0], "Missing end message of type test")

    def test_fromMessagesAddsChildMessages(self):
        """
        L{LoggedAction.fromMessages} adds direct child messages to the
        constructed L{LoggedAction}.
        """
        logger = MemoryLogger()
        # index 0:
        Message.new(x=1).write(logger)
        # index 1 - start action
        with start_action(logger, "test"):
            # index 2
            Message.new(x=2).write(logger)
            # index 3
            Message.new(x=3).write(logger)
        # index 4 - end action
        # index 5
        Message.new(x=4).write(logger)
        logged = self.fromMessagesIndex(logger.messages, 1)

        expectedChildren = [
            LoggedMessage(logger.messages[2]),
            LoggedMessage(logger.messages[3]),
        ]
        self.assertEqual(logged.children, expectedChildren)

    def test_fromMessagesAddsChildActions(self):
        """
        L{LoggedAction.fromMessages} recursively adds direct child actions to
        the constructed L{LoggedAction}.
        """
        logger = MemoryLogger()
        # index 0
        with start_action(logger, "test"):
            # index 1:
            with start_action(logger, "test2"):
                # index 2
                Message.new(message_type="end", x=2).write(logger)
            # index 3 - end action
            with start_action(logger, "test3"):
                # index 4
                pass
            # index 5 - end action
        # index 6 - end action
        logged = self.fromMessagesIndex(logger.messages, 0)

        self.assertEqual(logged.children[0], self.fromMessagesIndex(logger.messages, 1))
        self.assertEqual(
            logged.type_tree(), {"test": [{"test2": ["end"]}, {"test3": []}]}
        )

    def test_ofType(self):
        """
        L{LoggedAction.ofType} returns a list of L{LoggedAction} created by the
        specified L{ActionType}.
        """
        ACTION = ActionType("myaction", [], [], "An action!")
        logger = MemoryLogger()
        # index 0
        with start_action(logger, "test"):
            # index 1:
            with ACTION(logger):
                # index 2
                Message.new(x=2).write(logger)
            # index 3 - end action
            # index 4 - end action
            # index 5
        with ACTION(logger):
            pass
        # index 6 - end action
        logged = LoggedAction.ofType(logger.messages, ACTION)
        self.assertEqual(
            logged,
            [
                self.fromMessagesIndex(logger.messages, 1),
                self.fromMessagesIndex(logger.messages, 5),
            ],
        )

        # String-variant of ofType:
        logged2 = LoggedAction.ofType(logger.messages, "myaction")
        self.assertEqual(logged, logged2)

    def test_ofTypeNotFound(self):
        """
        L{LoggedAction.ofType} returns an empty list if actions of the given
        type cannot be found.
        """
        ACTION = ActionType("myaction", [], [], "An action!")
        logger = MemoryLogger()
        self.assertEqual(LoggedAction.ofType(logger.messages, ACTION), [])

    def test_descendants(self):
        """
        L{LoggedAction.descendants} returns all descendants of the
        L{LoggedAction}.
        """
        ACTION = ActionType("myaction", [], [], "An action!")
        logger = MemoryLogger()
        # index 0
        with ACTION(logger):
            # index 1:
            with start_action(logger, "test"):
                # index 2
                Message.new(x=2).write(logger)
            # index 3 - end action
            # index 4
            Message.new(x=2).write(logger)
        # index 5 - end action

        loggedAction = LoggedAction.ofType(logger.messages, ACTION)[0]
        self.assertEqual(
            list(loggedAction.descendants()),
            [
                self.fromMessagesIndex(logger.messages, 1),
                LoggedMessage(logger.messages[2]),
                LoggedMessage(logger.messages[4]),
            ],
        )

    def test_succeeded(self):
        """
        If the action succeeded, L{LoggedAction.succeeded} will be true.
        """
        logger = MemoryLogger()
        with start_action(logger, "test"):
            pass
        logged = self.fromMessagesIndex(logger.messages, 0)
        self.assertTrue(logged.succeeded)

    def test_notSucceeded(self):
        """
        If the action failed, L{LoggedAction.succeeded} will be false.
        """
        logger = MemoryLogger()
        try:
            with start_action(logger, "test"):
                raise KeyError()
        except KeyError:
            pass
        logged = self.fromMessagesIndex(logger.messages, 0)
        self.assertFalse(logged.succeeded)


class LoggedMessageTest(TestCase):
    """
    Tests for L{LoggedMessage}.
    """

    def test_values(self):
        """
        The values given to the L{LoggedMessage} constructor are stored on it.
        """
        message = {"x": 1}
        logged = LoggedMessage(message)
        self.assertEqual(logged.message, message)

    def test_ofType(self):
        """
        L{LoggedMessage.ofType} returns a list of L{LoggedMessage} created by the
        specified L{MessageType}.
        """
        MESSAGE = MessageType("mymessage", [], "A message!")
        logger = MemoryLogger()
        # index 0
        MESSAGE().write(logger)
        # index 1
        Message.new(x=2).write(logger)
        # index 2
        MESSAGE().write(logger)
        logged = LoggedMessage.ofType(logger.messages, MESSAGE)
        self.assertEqual(
            logged,
            [LoggedMessage(logger.messages[0]), LoggedMessage(logger.messages[2])],
        )

        # Lookup by string type:
        logged2 = LoggedMessage.ofType(logger.messages, "mymessage")
        self.assertEqual(logged, logged2)

    def test_ofTypeNotFound(self):
        """
        L{LoggedMessage.ofType} returns an empty list if messages of the given
        type cannot be found.
        """
        MESSAGE = MessageType("mymessage", [], "A message!")
        logger = MemoryLogger()
        self.assertEqual(LoggedMessage.ofType(logger.messages, MESSAGE), [])


class AssertContainsFields(TestCase):
    """
    Tests for L{assertContainsFields}.
    """

    class ContainsTest(TestCase):
        """
        A test case that uses L{assertContainsFields}.
        """

        def __init__(self, message, expectedFields):
            TestCase.__init__(self)
            self.message = message
            self.expectedFields = expectedFields

        def runTest(self):
            assertContainsFields(self, self.message, self.expectedFields)

    def test_equal(self):
        """
        Equal dictionaries contain each other.
        """
        message = {"a": 1}
        expected = message.copy()
        test = self.ContainsTest(message, expected)
        # No exception raised:
        test.debug()

    def test_additionalIsSuperSet(self):
        """
        If C{A} is C{B} plus some extra entries, C{A} contains the fields in
        C{B}.
        """
        message = {"a": 1, "b": 2, "c": 3}
        expected = {"a": 1, "c": 3}
        test = self.ContainsTest(message, expected)
        # No exception raised:
        test.debug()

    def test_missingFields(self):
        """
        If C{A} is C{B} minus some entries, C{A} does not contain the fields in
        C{B}.
        """
        message = {"a": 1, "c": 3}
        expected = {"a": 1, "b": 2, "c": 3}
        test = self.ContainsTest(message, expected)
        self.assertRaises(AssertionError, test.debug)

    def test_differentValues(self):
        """
        If C{A} has a different value for a specific field than C{B}, C{A} does
        not contain the fields in C{B}.
        """
        message = {"a": 1, "c": 3}
        expected = {"a": 1, "c": 2}
        test = self.ContainsTest(message, expected)
        self.assertRaises(AssertionError, test.debug)


class ValidateLoggingTestsMixin(object):
    """
    Tests for L{validateLogging} and L{capture_logging}.
    """

    validate = None

    def test_decoratedFunctionCalledWithMemoryLogger(self):
        """
        The underlying function decorated with L{validateLogging} is called with
        a L{MemoryLogger} instance.
        """
        result = []

        class MyTest(TestCase):
            @self.validate(None)
            def test_foo(this, logger):
                result.append((this, logger.__class__))

        theTest = MyTest("test_foo")
        theTest.run()
        self.assertEqual(result, [(theTest, MemoryLogger)])

    def test_decorated_function_passthrough(self):
        """
        Additional arguments are passed to the underlying function.
        """
        result = []

        def another_wrapper(f):
            def g(this):
                f(this, 1, 2, c=3)

            return g

        class MyTest(TestCase):
            @another_wrapper
            @self.validate(None)
            def test_foo(this, a, b, logger, c=None):
                result.append((a, b, c))

        theTest = MyTest("test_foo")
        theTest.debug()
        self.assertEqual(result, [(1, 2, 3)])

    def test_newMemoryLogger(self):
        """
        The underlying function decorated with L{validateLogging} is called with
        a new L{MemoryLogger} every time the wrapper is called.
        """
        result = []

        class MyTest(TestCase):
            @self.validate(None)
            def test_foo(this, logger):
                result.append(logger)

        theTest = MyTest("test_foo")
        theTest.run()
        theTest.run()
        self.assertIsNot(result[0], result[1])

    def test_returns(self):
        """
        The result of the underlying function is returned by wrapper when called.
        """

        class MyTest(TestCase):
            @self.validate(None)
            def test_foo(self, logger):
                return 123

        self.assertEqual(MyTest("test_foo").test_foo(), 123)

    def test_raises(self):
        """
        The exception raised by the underlying function is passed through by the
        wrapper when called.
        """
        exc = Exception()

        class MyTest(TestCase):
            @self.validate(None)
            def test_foo(self, logger):
                raise exc

        raised = None
        try:
            MyTest("test_foo").debug()
        except Exception as e:
            raised = e
        self.assertIs(exc, raised)

    def test_name(self):
        """
        The wrapper has the same name as the wrapped function.
        """

        class MyTest(TestCase):
            @self.validate(None)
            def test_foo(self, logger):
                pass

        self.assertEqual(MyTest.test_foo.__name__, "test_foo")

    def test_addCleanupValidate(self):
        """
        When a test method is decorated with L{validateLogging} it has
        L{MemoryLogger.validate} registered as a test cleanup.
        """
        MESSAGE = MessageType("mymessage", [], "A message")

        class MyTest(TestCase):
            @self.validate(None)
            def runTest(self, logger):
                self.logger = logger
                logger.write({"message_type": "wrongmessage"}, MESSAGE._serializer)

        test = MyTest()
        with self.assertRaises(ValidationError) as context:
            test.debug()
        # Some reference to the reason:
        self.assertIn("wrongmessage", str(context.exception))
        # Some reference to which file caused the problem:
        self.assertIn("test_testing.py", str(context.exception))

    def test_addCleanupTracebacks(self):
        """
        When a test method is decorated with L{validateLogging} it has has a
        check unflushed tracebacks in the L{MemoryLogger} registered as a
        test cleanup.
        """

        class MyTest(TestCase):
            @self.validate(None)
            def runTest(self, logger):
                try:
                    1 / 0
                except ZeroDivisionError:
                    write_traceback(logger)

        test = MyTest()
        self.assertRaises(UnflushedTracebacks, test.debug)

    def test_assertion(self):
        """
        If a callable is passed to L{validateLogging}, it is called with the
        L{TestCase} instance and the L{MemoryLogger} passed to the test
        method.
        """
        result = []

        class MyTest(TestCase):
            def assertLogging(self, logger):
                result.append((self, logger))

            @self.validate(assertLogging)
            def runTest(self, logger):
                self.logger = logger

        test = MyTest()
        test.run()
        self.assertEqual(result, [(test, test.logger)])

    def test_assertionArguments(self):
        """
        If a callable together with additional arguments and keyword arguments are
        passed to L{validateLogging}, the callable is called with the additional
        args and kwargs.
        """
        result = []

        class MyTest(TestCase):
            def assertLogging(self, logger, x, y):
                result.append((self, logger, x, y))

            @self.validate(assertLogging, 1, y=2)
            def runTest(self, logger):
                self.logger = logger

        test = MyTest()
        test.run()
        self.assertEqual(result, [(test, test.logger, 1, 2)])

    def test_assertionAfterTest(self):
        """
        If a callable is passed to L{validateLogging}, it is called with the
        after the main test code has run, allowing it to make assertions
        about log messages from the test.
        """

        class MyTest(TestCase):
            def assertLogging(self, logger):
                self.result.append(2)

            @self.validate(assertLogging)
            def runTest(self, logger):
                self.result = [1]

        test = MyTest()
        test.run()
        self.assertEqual(test.result, [1, 2])

    def test_assertionBeforeTracebackCleanup(self):
        """
        If a callable is passed to L{validateLogging}, it is called with the
        before the check for unflushed tracebacks, allowing it to flush
        traceback log messages.
        """

        class MyTest(TestCase):
            def assertLogging(self, logger):
                logger.flushTracebacks(ZeroDivisionError)
                self.flushed = True

            @self.validate(assertLogging)
            def runTest(self, logger):
                self.flushed = False
                try:
                    1 / 0
                except ZeroDivisionError:
                    write_traceback(logger)

        test = MyTest()
        test.run()
        self.assertTrue(test.flushed)


class ValidateLoggingTests(ValidateLoggingTestsMixin, TestCase):
    """
    Tests for L{validate_logging}.
    """

    validate = staticmethod(validate_logging)


class CaptureLoggingTests(ValidateLoggingTestsMixin, TestCase):
    """
    Tests for L{capture_logging}.
    """

    validate = staticmethod(capture_logging)

    def setUp(self):
        # Since we're not always calling the test method via the TestCase
        # infrastructure, sometimes cleanup methods are not called. This
        # means the original default logger is not restored. So we do so
        # manually. If the issue is a bug in capture_logging itself the
        # tests below will catch that.
        original_logger = _output._DEFAULT_LOGGER

        def cleanup():
            _output._DEFAULT_LOGGER = original_logger

        self.addCleanup(cleanup)

    def test_default_logger(self):
        """
        L{capture_logging} captures messages from logging that
        doesn't specify a L{Logger}.
        """

        class MyTest(TestCase):
            @capture_logging(None)
            def runTest(self, logger):
                Message.log(some_key=1234)
                self.logger = logger

        test = MyTest()
        test.run()
        self.assertEqual(test.logger.messages[0]["some_key"], 1234)

    def test_global_cleanup(self):
        """
        After the function wrapped with L{capture_logging} finishes,
        logging that doesn't specify a logger is logged normally.
        """

        class MyTest(TestCase):
            @capture_logging(None)
            def runTest(self, logger):
                pass

        test = MyTest()
        test.run()
        messages = []
        add_destination(messages.append)
        self.addCleanup(remove_destination, messages.append)
        Message.log(some_key=1234)
        self.assertEqual(messages[0]["some_key"], 1234)

    def test_global_cleanup_exception(self):
        """
        If the function wrapped with L{capture_logging} throws an exception,
        logging that doesn't specify a logger is logged normally.
        """

        class MyTest(TestCase):
            @capture_logging(None)
            def runTest(self, logger):
                raise RuntimeError()

        test = MyTest()
        test.run()
        messages = []
        add_destination(messages.append)
        self.addCleanup(remove_destination, messages.append)
        Message.log(some_key=1234)
        self.assertEqual(messages[0]["some_key"], 1234)

    def test_validationNotRunForSkip(self):
        """
        If the decorated test raises L{SkipTest} then the logging validation is
        also skipped.
        """

        class MyTest(TestCase):
            recorded = False

            def record(self, logger):
                self.recorded = True

            @validateLogging(record)
            def runTest(self, logger):
                raise SkipTest("Do not run this test.")

        test = MyTest()
        result = TestResult()
        test.run(result)

        # Verify that the validation function did not run and that the test was
        # nevertheless marked as a skip with the correct reason.
        self.assertEqual(
            (test.recorded, result.skipped, result.errors, result.failures),
            (False, [(test, "Do not run this test.")], [], []),
        )


class JSONEncodingTests(TestCase):
    """Tests for L{capture_logging} JSON encoder support."""

    @skipUnless(np, "NumPy is not installed.")
    @capture_logging(None)
    def test_default_JSON_encoder(self, logger):
        """
        L{capture_logging} validates using L{EliotJSONEncoder} by default.
        """
        # Default JSON encoder can't handle NumPy:
        log_message(message_type="hello", number=np.uint32(12))

    @capture_logging(None, encoder_=CustomJSONEncoder)
    def test_custom_JSON_encoder(self, logger):
        """
        L{capture_logging} can be called with a custom JSON encoder, which is then
        used for validation.
        """
        # Default JSON encoder can't handle this custom object:
        log_message(message_type="hello", object=CustomObject())


MESSAGE1 = MessageType(
    "message1", [Field.forTypes("x", [int], "A number")], "A message for testing."
)
MESSAGE2 = MessageType("message2", [], "A message for testing.")


class AssertHasMessageTests(TestCase):
    """
    Tests for L{assertHasMessage}.
    """

    class UnitTest(TestCase):
        """
        Test case that can be instantiated.
        """

        def runTest(self):
            pass

    def test_failIfNoMessagesOfType(self):
        """
        L{assertHasMessage} raises L{AssertionError} if the given L{MemoryLogger}
        has no messages of the given L{MessageType}.
        """
        test = self.UnitTest()
        logger = MemoryLogger()
        MESSAGE1(x=123).write(logger)
        self.assertRaises(AssertionError, assertHasMessage, test, logger, MESSAGE2)

    def test_returnsIfMessagesOfType(self):
        """
        L{assertHasMessage} returns the first message of the given L{MessageType}.
        """
        test = self.UnitTest()
        logger = MemoryLogger()
        MESSAGE1(x=123).write(logger)
        self.assertEqual(
            assertHasMessage(test, logger, MESSAGE1),
            LoggedMessage.ofType(logger.messages, MESSAGE1)[0],
        )

    def test_failIfNotSubset(self):
        """
        L{assertHasMessage} raises L{AssertionError} if the found message doesn't
        contain the given fields.
        """
        test = self.UnitTest()
        logger = MemoryLogger()
        MESSAGE1(x=123).write(logger)
        self.assertRaises(
            AssertionError, assertHasMessage, test, logger, MESSAGE1, {"x": 24}
        )

    def test_returnsIfSubset(self):
        """
        L{assertHasMessage} returns the first message of the given L{MessageType} if
        it contains the given fields.
        """
        test = self.UnitTest()
        logger = MemoryLogger()
        MESSAGE1(x=123).write(logger)
        self.assertEqual(
            assertHasMessage(test, logger, MESSAGE1, {"x": 123}),
            LoggedMessage.ofType(logger.messages, MESSAGE1)[0],
        )


ACTION1 = ActionType(
    "action1",
    [Field.forTypes("x", [int], "A number")],
    [Field.forTypes("result", [int], "A number")],
    "A action for testing.",
)
ACTION2 = ActionType("action2", [], [], "A action for testing.")


class AssertHasActionTests(TestCase):
    """
    Tests for L{assertHasAction}.
    """

    class UnitTest(TestCase):
        """
        Test case that can be instantiated.
        """

        def runTest(self):
            pass

    def test_failIfNoActionsOfType(self):
        """
        L{assertHasAction} raises L{AssertionError} if the given L{MemoryLogger}
        has no actions of the given L{ActionType}.
        """
        test = self.UnitTest()
        logger = MemoryLogger()
        with ACTION1(logger, x=123):
            pass
        self.assertRaises(AssertionError, assertHasAction, test, logger, ACTION2, True)

    def test_failIfWrongSuccessStatus(self):
        """
        L{assertHasAction} raises L{AssertionError} if the given success status does
        not match that of the found actions.
        """
        test = self.UnitTest()
        logger = MemoryLogger()
        with ACTION1(logger, x=123):
            pass
        try:
            with ACTION2(logger):
                1 / 0
        except ZeroDivisionError:
            pass
        self.assertRaises(AssertionError, assertHasAction, test, logger, ACTION1, False)
        self.assertRaises(AssertionError, assertHasAction, test, logger, ACTION2, True)

    def test_returnsIfMessagesOfType(self):
        """
        A successful L{assertHasAction} returns the first message of the given
        L{ActionType}.
        """
        test = self.UnitTest()
        logger = MemoryLogger()
        with ACTION1(logger, x=123):
            pass
        self.assertEqual(
            assertHasAction(test, logger, ACTION1, True),
            LoggedAction.ofType(logger.messages, ACTION1)[0],
        )

    def test_failIfNotStartSubset(self):
        """
        L{assertHasAction} raises L{AssertionError} if the found action doesn't
        contain the given start fields.
        """
        test = self.UnitTest()
        logger = MemoryLogger()
        with ACTION1(logger, x=123):
            pass
        self.assertRaises(
            AssertionError, assertHasAction, test, logger, ACTION1, True, {"x": 24}
        )

    def test_failIfNotEndSubset(self):
        """
        L{assertHasAction} raises L{AssertionError} if the found action doesn't
        contain the given end fields.
        """
        test = self.UnitTest()
        logger = MemoryLogger()
        with ACTION1(logger, x=123) as act:
            act.addSuccessFields(result=5)
        self.assertRaises(
            AssertionError,
            assertHasAction,
            test,
            logger,
            ACTION1,
            True,
            startFields={"x": 123},
            endFields={"result": 24},
        )

    def test_returns(self):
        """
        A successful L{assertHasAction} returns the first message of the given
        L{ActionType} after doing all validation.
        """
        test = self.UnitTest()
        logger = MemoryLogger()
        with ACTION1(logger, x=123) as act:
            act.addSuccessFields(result=5)
        self.assertEqual(
            assertHasAction(test, logger, ACTION1, True, {"x": 123}, {"result": 5}),
            LoggedAction.ofType(logger.messages, ACTION1)[0],
        )


class PEP8Tests(TestCase):
    """
    Tests for PEP 8 method compatibility.
    """

    def test_LoggedAction_from_messages(self):
        """
        L{LoggedAction.from_messages} is the same as
        L{LoggedAction.fromMessages}.
        """
        self.assertEqual(LoggedAction.from_messages, LoggedAction.fromMessages)

    def test_LoggedAction_of_type(self):
        """
        L{LoggedAction.of_type} is the same as
        L{LoggedAction.ofType}.
        """
        self.assertEqual(LoggedAction.of_type, LoggedAction.ofType)

    def test_LoggedAction_end_message(self):
        """
        L{LoggedAction.end_message} is the same as L{LoggedAction.endMessage}.
        """
        action = LoggedAction({1: 2}, {3: 4}, [])
        self.assertEqual(action.end_message, action.endMessage)

    def test_LoggedAction_start_message(self):
        """
        L{LoggedAction.start_message} is the same as
        L{LoggedAction.startMessage}.
        """
        action = LoggedAction({1: 2}, {3: 4}, [])
        self.assertEqual(action.start_message, action.startMessage)

    def test_LoggedMessage_of_type(self):
        """
        L{LoggedMessage.of_type} is the same as
        L{LoggedMessage.ofType}.
        """
        self.assertEqual(LoggedMessage.of_type, LoggedMessage.ofType)

    def test_validate_logging(self):
        """
        L{validate_logging} is the same as L{validateLogging}.
        """
        self.assertEqual(validate_logging, validateLogging)


class LowLevelTestingHooks(TestCase):
    """Tests for lower-level APIs for setting up MemoryLogger."""

    @capture_logging(None)
    def test_swap_logger(self, logger):
        """C{swap_logger} swaps out the current logger."""
        new_logger = MemoryLogger()
        old_logger = swap_logger(new_logger)
        Message.log(message_type="hello")

        # We swapped out old logger for new:
        self.assertIs(old_logger, logger)
        self.assertEqual(new_logger.messages[0]["message_type"], "hello")

        # Now restore old logger:
        intermediate_logger = swap_logger(old_logger)
        Message.log(message_type="goodbye")
        self.assertIs(intermediate_logger, new_logger)
        self.assertEqual(logger.messages[0]["message_type"], "goodbye")

    def test_check_for_errors_unflushed_tracebacks(self):
        """C{check_for_errors} raises on unflushed tracebacks."""
        logger = MemoryLogger()

        # No errors initially:
        check_for_errors(logger)

        try:
            1 / 0
        except ZeroDivisionError:
            write_traceback(logger)
        logger.flush_tracebacks(ZeroDivisionError)

        # Flushed tracebacks don't count:
        check_for_errors(logger)

        # But unflushed tracebacks do:
        try:
            raise RuntimeError
        except RuntimeError:
            write_traceback(logger)
        with self.assertRaises(UnflushedTracebacks):
            check_for_errors(logger)

    def test_check_for_errors_validation(self):
        """C{check_for_errors} raises on validation errors."""
        logger = MemoryLogger()
        logger.write({"x": 1, "message_type": "mem"})

        # No errors:
        check_for_errors(logger)

        # Now long something unserializable to JSON:
        logger.write({"message_type": object()})
        with self.assertRaises(TypeError):
            check_for_errors(logger)