#!/usr/bin/env python
# -- Content-Encoding: UTF-8 --
"""
Tests the log service

:author: Thomas Calmant
"""

# Pelix
import pelix.framework
import pelix.misc
from pelix.ipopo.constants import use_ipopo
from pelix.misc.log import LOG_DEBUG, LOG_INFO, LOG_WARNING, LOG_ERROR

# Standard library
import logging
import sys
import time

try:
    import unittest2 as unittest
except ImportError:
    import unittest

# ------------------------------------------------------------------------------

__version__ = "1.0.0"


# ------------------------------------------------------------------------------


class LogServiceTest(unittest.TestCase):
    """
    Tests the log service
    """

    def setUp(self):
        """
        Prepares a framework and a registers a service to export
        """
        # Create the framework
        self.framework = pelix.framework.create_framework(
            ('pelix.ipopo.core', 'pelix.misc.log'))
        self.framework.start()

        # Get the service
        self.service = self._get_service()

    def tearDown(self):
        """
        Cleans up for next test
        """
        # Stop the framework
        pelix.framework.FrameworkFactory.delete_framework(self.framework)
        self.service = None
        self.framework = None

    def _get_service(self):
        """
        Returns the log service
        """
        context = self.framework.get_bundle_context()
        ref = context.get_service_reference(pelix.misc.LOG_SERVICE)
        return context.get_service(ref)

    def test_log(self):
        """
        Basic tests for the log service
        """
        # Try to log at various log levels
        prev_logs = []
        for level, osgi_level in (
                (logging.DEBUG, LOG_DEBUG), (logging.INFO, LOG_INFO),
                (logging.WARNING, LOG_WARNING), (logging.ERROR, LOG_ERROR),
                (logging.CRITICAL, LOG_ERROR)):
            # Log at the expected level
            self.service.log(level, logging.getLevelName(level))

            # Get new logs
            new_logs = self.service.get_log()
            latest = new_logs[-1]

            # Check time stamp
            self.assertLessEqual(
                latest.time, time.time() + .5, "Log in future")
            self.assertGreaterEqual(
                latest.time, time.time() - 10, "Log too far in past")

            # Check stored info
            self.assertEqual(latest.level, level, "Wrong log level")
            self.assertEqual(latest.osgi_level, osgi_level,
                             "Wrong OSGi log level")
            self.assertEqual(latest.message, logging.getLevelName(level),
                             "Wrong log message")
            self.assertIsNone(latest.bundle, "Unexpected bundle info")
            self.assertIsNone(latest.exception, "Unexpected exception data")
            self.assertIsNone(latest.reference, "Unexpected reference data")

            # Compare list (not tuples)
            new_logs = list(new_logs)
            self.assertListEqual(new_logs, prev_logs + [latest],
                                 "Logs list changed")
            prev_logs = new_logs

    def test_logging(self):
        """
        Tests if logs made with the logging are handled
        """
        # Debug logs aren't taken into account
        logging.debug("Some log message at %s",
                      logging.getLevelName(logging.DEBUG))
        self.assertListEqual(
            list(self.service.get_log()), [], "Debug message logged")

        # Try to log at various log levels
        prev_logs = []
        for level, osgi_level in (
                (logging.INFO, LOG_INFO), (logging.WARNING, LOG_WARNING),
                (logging.ERROR, LOG_ERROR), (logging.CRITICAL, LOG_ERROR)):
            # Log at the expected level
            logging.log(level, "Some log message at %s",
                        logging.getLevelName(level))

            # Get new logs
            new_logs = self.service.get_log()
            latest = new_logs[-1]

            # Check time stamp
            self.assertLessEqual(
                latest.time, time.time() + .5, "Log in future")
            self.assertGreaterEqual(
                latest.time, time.time() - 10, "Log too far in past")

            # Check stored info
            self.assertEqual(latest.level, level, "Wrong log level")
            self.assertEqual(latest.osgi_level, osgi_level,
                             "Wrong OSGi log level")
            self.assertIn(logging.getLevelName(level), latest.message,
                          "Wrong log message")
            self.assertIsNone(latest.bundle, "Unexpected bundle info")
            self.assertIsNone(latest.exception, "Unexpected exception data")
            self.assertIsNone(latest.reference, "Unexpected reference data")

            # Check string representation
            self.assertIn(logging.getLevelName(level), str(latest))

            # Compare list (not tuples)
            new_logs = list(new_logs)
            self.assertListEqual(new_logs, prev_logs + [latest],
                                 "Logs list changed")
            prev_logs = new_logs

    def test_logging_filter_level(self):
        """
        Tests the change of filter for the logging handler
        """
        for filter_level in (logging.DEBUG, logging.INFO, logging.WARNING,
                             logging.ERROR):
            for int_level in (True, False):
                # Restart the framework
                self.tearDown()
                self.setUp()

                # Change the framework property and reload the log service
                if int_level:
                    self.framework.add_property(
                        pelix.misc.PROPERTY_LOG_LEVEL, filter_level)
                else:
                    self.framework.add_property(
                        pelix.misc.PROPERTY_LOG_LEVEL,
                        logging.getLevelName(filter_level))

                self.framework.get_bundle_by_name("pelix.misc.log").update()
                self.service = self._get_service()

                # Log for each level
                for level in (logging.DEBUG, logging.INFO, logging.WARNING,
                              logging.ERROR):
                    # Log something
                    logging.log(level, "Some log at %s",
                                logging.getLevelName(level))

                    try:
                        latest = self.service.get_log()[-1]
                        if level >= filter_level:
                            self.assertIn(logging.getLevelName(level),
                                          latest.message)
                    except IndexError:
                        if level >= filter_level:
                            self.fail("Missing a log matching the filter")

        # Try with invalid levels, default level is INFO
        filter_level = logging.INFO
        for invalid in (None, "", "deb", "ug", "foobar", {1: 2}, [1, 2]):
            # Restart the framework
            self.tearDown()
            self.setUp()

            # Change the framework property and reload the log service
            self.framework.add_property(pelix.misc.PROPERTY_LOG_LEVEL, invalid)

            self.framework.get_bundle_by_name("pelix.misc.log").update()
            self.service = self._get_service()

            # Log for each level
            for level in (logging.DEBUG, logging.INFO, logging.WARNING,
                          logging.ERROR):
                # Log something
                logging.log(level, "Some log at %s",
                            logging.getLevelName(level))

                try:
                    latest = self.service.get_log()[-1]
                    if level >= filter_level:
                        self.assertIn(logging.getLevelName(level),
                                      latest.message)
                except IndexError:
                    if level >= filter_level:
                        self.fail("Missing a log matching the filter")

    def test_listener(self):
        """
        Tests when log listeners are notified
        """
        entries = []

        # Prepare the listener
        class Listener:
            @staticmethod
            def logged(entry):
                entries.append(entry)

        listener = Listener()

        # Register it twice
        self.service.add_log_listener(listener)
        self.service.add_log_listener(listener)