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

:author: <NAME>
"""

# Standard library
import logging
import sys
import time

try:
    import unittest2 as unittest
except ImportError:
    import unittest

# 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

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

__version_info__ = (1, 0, 0)
__version__ = ".".join(str(x) for x in __version_info__)

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


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.logger = self._get_logger()
        self.reader = self._get_reader()

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

    def _get_logger(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 _get_reader(self):
        """
        Returns the log reader service
        """
        context = self.framework.get_bundle_context()
        ref = context.get_service_reference(pelix.misc.LOG_READER_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.logger.log(level, logging.getLevelName(level))

            # Get new logs
            new_logs = self.reader.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.assertIs(latest.bundle, self.framework, "No 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.reader.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.reader.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.logger = self._get_logger()
                self.reader = self._get_reader()

                # 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.reader.get_log()[-1]
                        if level >= filter_level:
                            self.assertIn(logging.getLevelName(level),
                                          latest.message)
                    except IndexError:<commit_msg>Set logging level in log tests<commit_after>#!/usr/bin/env python
# -- Content-Encoding: UTF-8 --
"""
Tests the log service

:author: <NAME>
"""

# Standard library
import logging
import sys
import time

try:
    import unittest2 as unittest
except ImportError:
    import unittest  # type: ignore

# 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

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

__version_info__ = (1, 0, 0)
__version__ = ".".join(str(x) for x in __version_info__)

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


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.logger = self._get_logger()
        self.reader = self._get_reader()

        # Setup logging level
        logging.getLogger().setLevel(logging.INFO)

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

    def _get_logger(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 _get_reader(self):
        """
        Returns the log reader service
        """
        context = self.framework.get_bundle_context()
        ref = context.get_service_reference(pelix.misc.LOG_READER_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.logger.log(level, logging.getLevelName(level))

            # Get new logs
            new_logs = self.reader.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.assertIs(latest.bundle, self.framework, "No 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.reader.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.reader.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()

                # Ensure logging emits all records
                logging.getLogger().setLevel(logging.DEBUG)

                # 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.logger = self._get_logger()
                self.reader = self._get_reader()

                # 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.reader.get_log()[-1]
                        if level >= filter_level:
                            self.assertIn(logging.getLevelName(level),
                                          latest.message)
                    except IndexError:
{'fraction_non_alphanumeric': 0.10953068095925239, 'fraction_numerical': 0.001924001924001924, 'mean_word_length': 1.5850799289520425, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2062311', 'n_tokens_mistral': 3774, 'n_tokens_neox': 3606, 'n_words': 922}