diff --git a/src/openflexure_microscope_server/logging.py b/src/openflexure_microscope_server/logging.py index eb49f11f..ce00f5e3 100644 --- a/src/openflexure_microscope_server/logging.py +++ b/src/openflexure_microscope_server/logging.py @@ -16,6 +16,7 @@ from logging.handlers import RotatingFileHandler import os from fastapi.responses import PlainTextResponse +from fastapi import HTTPException OFM_LOG_FILE = None @@ -35,6 +36,13 @@ def configure_logging(log_folder): global OFM_LOG_FILE OFM_LOG_FILE = os.path.join(log_folder, "openflexure_microscope.log") + # Add OFM_HANDLER first so it can capture the error log if the + # log file can't be accessed. + ofm_format_str = "[%(asctime)s] [%(levelname)s] %(message)s" + OFM_HANDLER.setFormatter(logging.Formatter(ofm_format_str)) + OFM_HANDLER.addFilter(UvicornAccessFilter()) + root_logger.addHandler(OFM_HANDLER) + try: if not os.path.exists(log_folder): os.makedirs(log_folder) @@ -46,13 +54,12 @@ def configure_logging(log_folder): ) format_str = "[%(asctime)s] [%(levelname)s] <%(name)s> %(message)s" handler.setFormatter(logging.Formatter(format_str)) + handler.addFilter(UvicornAccessFilter()) root_logger.addHandler(handler) - ofm_format_str = "[%(asctime)s] [%(levelname)s] %(message)s" - OFM_HANDLER.setFormatter(logging.Formatter(ofm_format_str)) - root_logger.addHandler(OFM_HANDLER) except PermissionError as e: - logging.warning(f"Cannot create log file at {OFM_LOG_FILE}: {e}") + logging.error(f"Cannot create log file at {OFM_LOG_FILE}: {e}") + logging.info("") logging.info("****************************************************") logging.info("OFM server root logger has been set up at INFO level") @@ -80,12 +87,17 @@ def retrieve_log_from_file() -> PlainTextResponse: is written to while sending through FileResponse. """ if OFM_LOG_FILE is None: - raise RuntimeError( - "Cannot retrieve log file as logging directory hasn't been configured" + raise HTTPException( + 500, "Cannot retrieve log file as logging directory hasn't been configured." ) - with open(OFM_LOG_FILE, "r", encoding="utf-8") as logfile: - full_log = logfile.read() - return PlainTextResponse(full_log) + try: + with open(OFM_LOG_FILE, "r", encoding="utf-8") as logfile: + full_log = logfile.read() + return PlainTextResponse(full_log) + except IOError as e: + raise HTTPException( + 500, "An error occurred while trying to access the log file." + ) from e class OFMHandler(logging.Handler): @@ -116,11 +128,6 @@ class OFMHandler(logging.Handler): def emit(self, record): """Emit will save the logged record to the log.""" - # Basic filter for now that simply stops uvicorn.access logs - # These are the logs each time an API endpoint is accessed - # This is only the log for the UI. - if record.name.startswith("uvicorn.access"): - return try: if record.levelno >= self.level: self.append_record(record) @@ -137,4 +144,12 @@ class OFMHandler(logging.Handler): return "\n".join(self._log) +class UvicornAccessFilter(logging.Filter): + """A logging filter to filter out "uvicorn.access" messages.""" + + def filter(self, record): + """Return False if record is from "uvicorn.access".""" + return not record.name.startswith("uvicorn.access") + + OFM_HANDLER = OFMHandler() diff --git a/tests/test_logging.py b/tests/test_logging.py new file mode 100644 index 00000000..0a48125a --- /dev/null +++ b/tests/test_logging.py @@ -0,0 +1,157 @@ +"""Test the server's logging configuration and handling.""" + +import os +import tempfile +import logging + +from fastapi.responses import PlainTextResponse +from fastapi import HTTPException +import pytest + +from openflexure_microscope_server import logging as ofm_logging + + +def test_no_warnings_if_correct_permissions(caplog): + """Check that configure_logging no warnings on normal operation. + + Checking that: + + * The logger can be set up without warnings. + * That the log file is created. + * That the log file contains the expected message about setting up the logger. + """ + # Reset handler at start of test + ofm_logging.OFM_HANDLER = ofm_logging.OFMHandler() + with caplog.at_level(logging.WARNING): + with tempfile.TemporaryDirectory() as tmpdir: + ofm_logging.configure_logging(tmpdir) + assert len(caplog.records) == 0 + with open(ofm_logging.OFM_LOG_FILE, "r", encoding="utf-8") as log_file: + log_txt = log_file.read() + assert "OFM server root logger has been set up at INFO level" in log_txt + root_logger = logging.getLogger() + assert ofm_logging.OFM_HANDLER in root_logger.handlers + + +def test_permission_error_raises_warning(mocker, caplog): + """Check that configure_logging raises warning on PermissionError.""" + mocker.patch( + "openflexure_microscope_server.logging.RotatingFileHandler", + side_effect=PermissionError("Permission denied"), + ) + # Reset handler at start of test + ofm_logging.OFM_HANDLER = ofm_logging.OFMHandler() + with caplog.at_level(logging.WARNING): + with tempfile.TemporaryDirectory() as tmpdir: + ofm_logging.configure_logging(tmpdir) + assert len(caplog.records) == 1 + # Check OFM logger is added even if the file logger couldn't be. + root_logger = logging.getLogger() + assert ofm_logging.OFM_HANDLER in root_logger.handlers + + +def test_making_log_dir(caplog): + """Check that configure_logging will make a dir if needed.""" + # Reset handler at start of test + ofm_logging.OFM_HANDLER = ofm_logging.OFMHandler() + with caplog.at_level(logging.WARNING): + with tempfile.TemporaryDirectory() as tmpdir: + log_dir = os.path.join(tmpdir, "new_dir") + assert not os.path.isdir(log_dir) + ofm_logging.configure_logging(log_dir) + assert len(caplog.records) == 0 + assert os.path.isdir(log_dir) + + +def test_max_logs(): + """Proclaim that only the most recent 250 logs are stored in memory.""" + # Reset handler at start of test + ofm_logging.OFM_HANDLER = ofm_logging.OFMHandler() + with tempfile.TemporaryDirectory() as tmpdir: + ofm_logging.configure_logging(tmpdir) + # But I would log five hundred times. + for i in range(500): + logging.info("log %s", i) + logs = ofm_logging.OFM_HANDLER.log_history.split("\n") + assert len(logs) == 250 + assert logs[0].endswith("log 250") + assert logs[-1].endswith("log 499") + # And I would log five hundred more. + for i in range(500): + logging.info("log %s", 500 + i) + # Just to be the test who logged a thousand times to check your hand-el-or! + logs = ofm_logging.OFM_HANDLER.log_history.split("\n") + assert len(logs) == 250 + assert logs[0].endswith("log 750") + assert logs[-1].endswith("log 999") + + +def test_ofm_handler_ignores_uvicorn_access(): + """Check uvicorn.access logs are ignored by custom handler. + + The uvicorn.access logger logs every time an endpoint is triggered. + """ + # Reset handler at start of test + ofm_logging.OFM_HANDLER = ofm_logging.OFMHandler() + with tempfile.TemporaryDirectory() as tmpdir: + ofm_logging.configure_logging(tmpdir) + logs = ofm_logging.OFM_HANDLER.log_history.split("\n") + starting_len = len(logs) + logging.info("Root 1") + uvicorn_logger = logging.getLogger("uvicorn.access") + for i in range(500): + uvicorn_logger.info("Uvicorn log %s", i) + logging.info("Root 2") + logs = ofm_logging.OFM_HANDLER.log_history.split("\n") + assert len(logs) == starting_len + 2 + assert logs[starting_len].endswith("Root 1") + assert logs[-1].endswith("Root 2") + + +def test_server_responses(): + """Check that the FastAPI responses from the server are as expected. + + * retrieve_log() -> PlainTextResponse containing the last 250 logs separated by + a new line + * retrieve_log_from_file() -> All the logs from the log file. + """ + # Reset handler at start of test + ofm_logging.OFM_HANDLER = ofm_logging.OFMHandler() + with tempfile.TemporaryDirectory() as tmpdir: + ofm_logging.configure_logging(tmpdir) + for i in range(500): + logging.info("log %s", i) + log_response = ofm_logging.retrieve_log() + assert isinstance(log_response, PlainTextResponse) + logs = log_response.body.decode().split("\n") + assert len(logs) == 250 + file_log_response = ofm_logging.retrieve_log_from_file() + assert isinstance(file_log_response, PlainTextResponse) + file_logs = file_log_response.body.decode().split("\n") + assert len(file_logs) > 500 + + +def test_server_response_with_no_log_file(): + """Check that an HTTP exception is raised if logging isn't configured.""" + # Reset handler at start of test + ofm_logging.OFM_HANDLER = ofm_logging.OFMHandler() + # Also Reset log file global + ofm_logging.OFM_LOG_FILE = None + with pytest.raises(HTTPException) as excinfo: + ofm_logging.retrieve_log_from_file() + assert excinfo.value.status_code == 500 + + +def test_server_response_with_no_log_dir(): + """Check that an HTTP exception is raised if the log file cannot be accessed.""" + ofm_logging.OFM_HANDLER = ofm_logging.OFMHandler() + with tempfile.TemporaryDirectory() as tmpdir: + ofm_logging.configure_logging(tmpdir) + for i in range(500): + logging.info("log %s", i) + file_log_response = ofm_logging.retrieve_log_from_file() + assert isinstance(file_log_response, PlainTextResponse) + # Close and delete the log folder, this should create an error. + with pytest.raises(HTTPException) as excinfo: + ofm_logging.retrieve_log_from_file() + assert excinfo.value.status_code == 500