Source code for toil.test.src.realtimeLoggerTest

# Copyright (C) 2015-2021 Regents of the University of California
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

import logging

from toil.job import Job
from toil.realtimeLogger import RealtimeLogger
from toil.test import ToilTest

[docs] class RealtimeLoggerTest(ToilTest):
[docs] def testRealtimeLogger(self): options = Job.Runner.getDefaultOptions(self._getTestJobStorePath()) options.realTimeLogging = True options.logLevel = 'INFO' detector = MessageDetector() # Set up a log message detector to the root logger logging.getLogger().addHandler(detector) # I believe coloredlogs replaces handlers with its own when doing handler formatting, preserving only filters # options.colored_logs = False Job.Runner.startToil(LogTest(), options) # We need the message we're supposed to see self.assertTrue(detector.detected) # But not the message that shouldn't be logged. self.assertFalse(detector.overLogged)
[docs] class MessageDetector(logging.StreamHandler): """ Detect the secret message and set a flag. """ def __init__(self): self.detected = False # Have we seen the message we want? self.overLogged = False # Have we seen the message we don't want? super().__init__()
[docs] def emit(self, record): if record.msg == 'This should be logged at info level': self.detected = True if record.msg == 'This should be logged at debug level': self.overLogged = True
[docs] class LogTest(Job): def __init__(self): Job.__init__(self, memory=100000, cores=1, disk='3G')
[docs] def run(self, fileStore):'This should be logged at info level') RealtimeLogger.debug('This should be logged at debug level')