Linked Questions

Popular Questions

I'm new-ish to python logging and would like to create a solid logging setup for a project to uses watchdog to log the creation of DICOM files in a given directory. My project structure is:

planqa\
    src\
        watcher.py
        qa_logger.py
        logs\
            qa_watch.log
    tests\
        test_watchdog.py


So far I've got a LOGGING_DICT in qa_logger.py. LOGGING_DICT is imported into watcher.py (and eventually other modules), and then near the top of watcher.py, I've put the following lines:

logging.config.dictConfig(LOGGING_DICT)
logger = logging.getLogger()

When I run python watcher.py, all logs are generated nicely. Logs are both printed to the console and written to the file src\logs\qa_watch.log.

The problem arises when I run pytest. No logs are generated at all. Yet, strangely, the file the file src\logs\qa_watch.log is created if it doesn't exist, but it's just never written to (unlike when I run python watcher.py)!

I'd like to use pytest to verify that a log was generated in src\logs\qa_watch.log when a valid DICOM file is created in the watched folder.

Any help would be greatly appreciated, including any comments on how to better structure things! I'm still (and always will be) learning!

EDIT: Log level isn't the issue. I have the same problem if I use logger.warning() instead of logger.info() for the logs in watcher.py.

Files:

# qa_logger.py

import logging
from logging.config import dictConfig
import os
from os.path import dirname, join as pjoin
import sys

LOG_DIR = pjoin(dirname(dirname(__file__)), "logs")

LOGGING_DICT = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
    'f': {
        'format': '%(asctime)s %(name)-12s %(levelname)-8s: %(message)s'
        }
    },
'handlers': {
    'console': {
        'class': 'logging.StreamHandler',
        'formatter': 'f',
        'level': logging.DEBUG
        },
    'file': {
        'class': 'logging.handlers.TimedRotatingFileHandler',
        'level': logging.DEBUG,
        'formatter': 'f',
        'filename': pjoin(LOG_DIR, "qa_watch.log"),
        'when': 'midnight',
        'interval': 1,
        'backupCount': 30
        }
    },
'root': {
    'handlers': ['console', 'file'],
    'level': logging.DEBUG,
    },
}


# watcher.py

import logging
from time import sleep
from os.path import abspath, dirname, join as pjoin
from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler
from src.qa_logger import LOGGING_DICT

logging.config.dictConfig(LOGGING_DICT)
logger = logging.getLogger()

class NewFileHandler(PatternMatchingEventHandler):
    patterns = ['*.dcm']
    ignore_patterns = None
    case_sensitive = False
    ignore_directories = True 

    def on_created(self, event):
        msg = "Caught creation of {}".format(event.src_path)
        logger.info(msg)


def watch_for_dicoms_created_in_folder(folder=None):

    if folder is None:
        folder = abspath(pjoin(dirname(dirname(__file__)), "tests", "data"))

    dcm_handler = NewFileHandler()

    observer = Observer()
    observer.schedule(dcm_handler, folder, recursive=True)
    observer.start()
    logger.info("----- Started watcher -----")
    try:
        while True:
            sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

if __name__ == "__main__":
    watch_for_dicoms_created_in_folder()


# test_watchdog.py

import logging
import os
from os.path import dirname, join as pjoin, isfile
from multiprocessing import Process
from time import sleep
from src.watcher import watch_for_dicoms_created_in_folder
from src.qa_logger import LOG_DIR

DATA_DIR = pjoin(dirname(__file__), "data")
LOG_PATH = pjoin(LOG_DIR, "qa_watch.log")

def test_watching_for_file_creation():

    dcm_test_filepath = pjoin(DATA_DIR, "a_dcm_to_catch.dcm")
    nondcm_test_filepath = pjoin(DATA_DIR, "not_a_dcm_to_catch.txt")

    # Watcher is run in a separate process or subsequent code will not execute
    watch1 = Process(name='watch1', target=watch_for_dicoms_created_in_folder)

    watch1.start()

    dcm_test_file = open(dcm_test_filepath, 'w')
    nondcm_test_file = open(nondcm_test_filepath, 'w')
    dcm_test_file.close()
    nondcm_test_file.close()

    sleep(0.2) # Give watcher time to see files and create logfile
    watch1.terminate()

    assert isfile(LOG_PATH)

    with open(LOG_PATH) as logfile:
        assert dcm_test_filepath in logfile
        assert not nondcm_test_filepath in logfile

    # Cleanup
    try:
        os.remove(dcm_test_filepath)
        os.remove(nondcm_test_filepath)
    except OSError:
        pass

Related Questions