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