feat: add TRACE log level and patch_logging() (#84)

* feat: add TRACE log level and `patch_logging()` for fine-grained debugging

Ports the TRACE log level support from netius (hivesolutions/netius#49)
to the appier framework, enabling protocol-level verbose logging below
DEBUG for fine-grained debugging of low-level operations.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: use tempfile instead of /dev/null in rotating handler tests

/dev/null does not exist on Windows, causing test failures on CI.
Use tempfile.mkstemp() for cross-platform compatibility.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* chore: small cosmetic fix

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: joamag

CHANGELOG.md

@@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-*
+* TRACE log level and `patch_logging()` for fine-grained protocol-level debugging
 
 ### Changed
 

src/appier/__init__.py

@@ -229,13 +229,16 @@
 from .graph import Graph
 from .http import file_g, get_f, get, post, put, delete, patch, basic_auth, HTTPResponse
 from .log import (
+    SILENT,
+    TRACE,
     MemoryHandler,
     BaseFormatter,
     ThreadFormatter,
     DummyLogger,
     reload_format,
     rotating_handler,
     smtp_handler,
+    patch_logging,
     in_signature,
 )
 from .meta import Ordered, Indexed

src/appier/base.py

@@ -3671,6 +3671,11 @@ def is_child(self):
     def is_main(self):
         return threading.current_thread().ident == self.tid
 
+    def is_trace(self):
+        if not self.level:
+            return False
+        return self.level <= log.TRACE
+
     def is_devel(self):
         if not self.level:
             return False
@@ -4182,6 +4187,8 @@ def _level(cls, level):
             return level
         if level == "SILENT":
             return log.SILENT
+        if level == "TRACE":
+            return log.TRACE
         if hasattr(logging, "_checkLevel"):
             return logging._checkLevel(level)
         return logging.getLevelName(level)
@@ -4512,6 +4519,11 @@ def _load_config(self, apply=True):
     def _load_logging(
         self, level=None, set_default=True, format_base=None, format_tid=None
     ):
+        # patches the logging infra-structure so that the TRACE level
+        # is properly registered and available for usage, this call
+        # is idempotent and safe to be called multiple times
+        log.patch_logging()
+
         format_base = format_base or log.LOGGING_FORMAT
         format_tid = format_tid or log.LOGGING_FORMAT_TID
 

src/appier/log.py

@@ -72,6 +72,12 @@
 or an handler, this is used as an utility for debugging
 purposes more that a real feature for production systems """
 
+TRACE = logging.DEBUG - 5
+""" The trace level used for extremely detailed and verbose
+logging of protocol-level operations, this is meant to be
+used for fine-grained debugging of low-level operations
+like raw byte transfers and frame parsing """
+
 LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")
 """ The sequence of levels from the least sever to the
 most sever this sequence may be used to find all the
@@ -292,6 +298,9 @@ def set_tid(self, value, *args, **kwargs):
 
 
 class DummyLogger(object):
+    def trace(self, object):
+        pass
+
     def debug(self, object):
         pass
 
@@ -355,11 +364,28 @@ def smtp_handler(
     )
 
 
+def patch_logging():
+    if hasattr(logging, "_appier_patched"):
+        return
+
+    # patches the logging infra-structure adding the trace level
+    # support and the corresponding trace method to the logger
+    logging.addLevelName(TRACE, "TRACE")
+    logging.Logger.trace = _trace
+
+    logging._appier_patched = True
+
+
 def in_signature(callable, name):
     has_full = hasattr(inspect, "getfullargspec")
     if has_full:
         spec = inspect.getfullargspec(callable)
     else:
         spec = inspect.getargspec(callable)
     args, _varargs, kwargs = spec[:3]
-    return (args and name in args) or (kwargs and "secure" in kwargs)
+    return bool((args and name in args) or (kwargs and "secure" in kwargs))
+
+
+def _trace(self, message, *args, **kwargs):
+    if self.isEnabledFor(TRACE):
+        self._log(TRACE, message, args, **kwargs)

src/appier/test/log.py

@@ -28,13 +28,207 @@
 __license__ = "Apache License, Version 2.0"
 """ The license for the module """
 
+import os
 import logging
+import tempfile
 import unittest
 
+import logging.handlers
+
 import appier
 
+from appier import log
+
 
 class LogTest(unittest.TestCase):
+    def test_silent_value(self):
+        self.assertEqual(appier.SILENT, logging.CRITICAL + 1)
+        self.assertEqual(type(appier.SILENT), int)
+
+    def test_silent_above_critical(self):
+        self.assertTrue(appier.SILENT > logging.CRITICAL)
+
+    def test_trace_value(self):
+        self.assertEqual(appier.TRACE, 5)
+        self.assertEqual(appier.TRACE, logging.DEBUG - 5)
+        self.assertEqual(type(appier.TRACE), int)
+
+    def test_trace_below_debug(self):
+        self.assertTrue(appier.TRACE < logging.DEBUG)
+
+    def test_level_ordering(self):
+        self.assertTrue(appier.TRACE < logging.DEBUG)
+        self.assertTrue(logging.DEBUG < logging.INFO)
+        self.assertTrue(logging.INFO < logging.WARNING)
+        self.assertTrue(logging.WARNING < logging.ERROR)
+        self.assertTrue(logging.ERROR < logging.CRITICAL)
+        self.assertTrue(logging.CRITICAL < appier.SILENT)
+
+    def test_rotating_handler(self):
+        fd, path = tempfile.mkstemp()
+        os.close(fd)
+        try:
+            handler = appier.rotating_handler(path=path, max_bytes=1024, max_log=3)
+
+            self.assertEqual(type(handler), logging.handlers.RotatingFileHandler)
+            self.assertEqual(handler.maxBytes, 1024)
+            self.assertEqual(handler.backupCount, 3)
+
+            handler.close()
+        finally:
+            os.unlink(path)
+
+    def test_rotating_handler_defaults(self):
+        fd, path = tempfile.mkstemp()
+        os.close(fd)
+        try:
+            handler = appier.rotating_handler(path=path)
+
+            self.assertEqual(handler.maxBytes, 1048576)
+            self.assertEqual(handler.backupCount, 5)
+
+            handler.close()
+        finally:
+            os.unlink(path)
+
+    def test_patch_logging(self):
+        appier.patch_logging()
+
+        result = logging.getLevelName(appier.TRACE)
+
+        self.assertEqual(result, "TRACE")
+
+    def test_patch_logging_reverse(self):
+        appier.patch_logging()
+
+        result = logging.getLevelName("TRACE")
+
+        self.assertEqual(result, appier.TRACE)
+
+    def test_patch_logging_idempotent(self):
+        appier.patch_logging()
+        appier.patch_logging()
+
+        result = logging.getLevelName(appier.TRACE)
+
+        self.assertEqual(result, "TRACE")
+
+    def test_patch_logging_logger_trace(self):
+        appier.patch_logging()
+
+        logger = logging.getLogger("appier.test.trace")
+
+        self.assertTrue(hasattr(logger, "trace"))
+        self.assertTrue(callable(logger.trace))
+
+    def test_patch_logging_logger_trace_call(self):
+        appier.patch_logging()
+
+        logger = logging.getLogger("appier.test.trace.call")
+        logger.setLevel(appier.TRACE)
+        records = []
+        handler = logging.Handler()
+        handler.setLevel(appier.TRACE)
+        handler.emit = lambda record: records.append(record)
+        logger.addHandler(handler)
+
+        try:
+            logger.trace("trace test message")
+
+            self.assertEqual(len(records), 1)
+            self.assertEqual(records[0].getMessage(), "trace test message")
+            self.assertEqual(records[0].levelno, appier.TRACE)
+            self.assertEqual(records[0].levelname, "TRACE")
+        finally:
+            logger.removeHandler(handler)
+
+    def test_patch_logging_logger_trace_filtered(self):
+        appier.patch_logging()
+
+        logger = logging.getLogger("appier.test.trace.filtered")
+        logger.setLevel(logging.DEBUG)
+        records = []
+        handler = logging.Handler()
+        handler.setLevel(appier.TRACE)
+        handler.emit = lambda record: records.append(record)
+        logger.addHandler(handler)
+
+        try:
+            # the trace message should be filtered since the logger
+            # level is set to DEBUG which is above TRACE
+            logger.trace("this should be filtered")
+
+            self.assertEqual(len(records), 0)
+        finally:
+            logger.removeHandler(handler)
+
+    def test_level_trace_before_patch(self):
+        # temporarily removes the patched state to simulate a
+        # scenario where patch_logging() has not been called yet
+        patched = getattr(logging, "_appier_patched", None)
+        if patched:
+            del logging._appier_patched
+        trace_method = getattr(logging.Logger, "trace", None)
+        if trace_method:
+            del logging.Logger.trace
+        try:
+            result = appier.App._level("TRACE")
+
+            self.assertEqual(result, appier.TRACE)
+            self.assertEqual(result, 5)
+        finally:
+            if patched:
+                logging._appier_patched = patched
+            if trace_method:
+                logging.Logger.trace = trace_method
+
+    def test_level_trace_after_patch(self):
+        appier.patch_logging()
+
+        result = appier.App._level("TRACE")
+
+        self.assertEqual(result, appier.TRACE)
+        self.assertEqual(result, 5)
+
+    def test_level_silent(self):
+        result = appier.App._level("SILENT")
+
+        self.assertEqual(result, appier.SILENT)
+
+    def test_level_integer(self):
+        result = appier.App._level(logging.DEBUG)
+
+        self.assertEqual(result, logging.DEBUG)
+
+    def test_level_none(self):
+        result = appier.App._level(None)
+
+        self.assertEqual(result, None)
+
+    def test_in_signature(self):
+        def sample(a, b, secure=None):
+            pass
+
+        result = log.in_signature(sample, "secure")
+
+        self.assertEqual(result, True)
+
+    def test_in_signature_missing(self):
+        def sample(a, b):
+            pass
+
+        result = log.in_signature(sample, "secure")
+
+        self.assertEqual(result, False)
+
+    def test_in_signature_args(self):
+        def sample(a, b, secure):
+            pass
+
+        result = log.in_signature(sample, "secure")
+
+        self.assertEqual(result, True)
+
     def test_memory_handler(self):
         memory_handler = appier.MemoryHandler()
         formatter = logging.Formatter("%(message)s")