121 lines
3.8 KiB
Python
121 lines
3.8 KiB
Python
import logging
|
|
from queue import Queue, Empty
|
|
|
|
from pyucc.utils import logger as ulogger
|
|
|
|
|
|
class FakeTextWidget:
|
|
def __init__(self):
|
|
self._lines = []
|
|
self._tags = {}
|
|
self._state = 'normal'
|
|
self._yview = (0.0, 1.0)
|
|
|
|
def winfo_exists(self):
|
|
return True
|
|
|
|
def tag_config(self, level_name, **kwargs):
|
|
self._tags[level_name] = kwargs
|
|
|
|
def configure(self, **kwargs):
|
|
# noop or record state
|
|
if 'state' in kwargs:
|
|
self._state = kwargs['state']
|
|
|
|
def insert(self, pos, msg, tags=()):
|
|
self._lines.append((pos, msg, tags))
|
|
|
|
def index(self, what):
|
|
# return line count
|
|
return f"{len(self._lines)+1}.0"
|
|
|
|
def delete(self, a, b):
|
|
self._lines = []
|
|
|
|
def see(self, what):
|
|
pass
|
|
|
|
def yview(self):
|
|
return self._yview
|
|
|
|
def get_all_text(self):
|
|
return "\n".join([m for _, m, _ in self._lines])
|
|
|
|
|
|
def test_queue_putting_handler_puts_record():
|
|
q = Queue()
|
|
h = ulogger.QueuePuttingHandler(q)
|
|
rec = logging.LogRecord(name='test', level=logging.INFO, pathname=__file__, lineno=1, msg='hi', args=(), exc_info=None)
|
|
h.emit(rec)
|
|
got = q.get_nowait()
|
|
assert isinstance(got, logging.LogRecord)
|
|
assert got.msg == 'hi'
|
|
|
|
|
|
def test_tkinter_text_handler_emit_and_flush_pending(monkeypatch):
|
|
fake = FakeTextWidget()
|
|
handler = ulogger.TkinterTextHandler(fake, {logging.INFO: 'black'}, max_lines=10)
|
|
fmt = logging.Formatter('%(message)s')
|
|
handler.setFormatter(fmt)
|
|
# create record
|
|
rec = logging.LogRecord(name='test', level=logging.INFO, pathname=__file__, lineno=10, msg='hello', args=(), exc_info=None)
|
|
handler.emit(rec)
|
|
# pending buffer should have one
|
|
assert len(handler._pending_records) == 1
|
|
# flush pending should insert into fake widget
|
|
handler.flush_pending()
|
|
text = fake.get_all_text()
|
|
assert 'hello' in text
|
|
|
|
|
|
def test_process_global_log_queue_dispatch(monkeypatch):
|
|
# prepare global queue with one record
|
|
q = Queue()
|
|
rec = logging.LogRecord(name='proc', level=logging.INFO, pathname=__file__, lineno=1, msg='msg1', args=(), exc_info=None)
|
|
q.put(rec)
|
|
monkeypatch.setattr(ulogger, '_global_log_queue', q)
|
|
|
|
# fake console and file handlers
|
|
class DummyHandler:
|
|
def __init__(self):
|
|
self.handled = []
|
|
def handle(self, record):
|
|
self.handled.append(record)
|
|
|
|
console = DummyHandler()
|
|
fileh = DummyHandler()
|
|
monkeypatch.setattr(ulogger, '_actual_console_handler', console)
|
|
monkeypatch.setattr(ulogger, '_actual_file_handler', fileh)
|
|
|
|
# fake tkinter handler
|
|
fake = FakeTextWidget()
|
|
tkhandler = ulogger.TkinterTextHandler(fake, {logging.INFO: 'black'})
|
|
fmt = logging.Formatter('%(message)s')
|
|
tkhandler.setFormatter(fmt)
|
|
monkeypatch.setattr(ulogger, '_actual_tkinter_handler', tkhandler)
|
|
|
|
# fake root with after
|
|
class FakeRoot:
|
|
def after(self, ms, func):
|
|
return 'after-id'
|
|
def winfo_exists(self):
|
|
return True
|
|
monkeypatch.setattr(ulogger, '_tk_root_instance_for_processing', FakeRoot())
|
|
|
|
monkeypatch.setattr(ulogger, '_logging_system_active', True)
|
|
# call processor
|
|
ulogger._process_global_log_queue()
|
|
# console and file handlers should have handled the record
|
|
assert len(console.handled) == 1
|
|
assert len(fileh.handled) == 1
|
|
# tkinter handler flush_pending should have cleared pending
|
|
assert len(tkhandler._pending_records) == 0
|
|
|
|
# cleanup
|
|
monkeypatch.setattr(ulogger, '_logging_system_active', False)
|
|
monkeypatch.setattr(ulogger, '_actual_console_handler', None)
|
|
monkeypatch.setattr(ulogger, '_actual_file_handler', None)
|
|
monkeypatch.setattr(ulogger, '_actual_tkinter_handler', None)
|
|
monkeypatch.setattr(ulogger, '_global_log_queue', None)
|
|
monkeypatch.setattr(ulogger, '_tk_root_instance_for_processing', None)
|