fix(logging-filter): add some changes for discussion

Author: open-dynaMIX

anonip.py

@@ -304,46 +304,66 @@ def truncate_address(self, ip):
         return ip.supernet(new_prefix=self._prefixes[ip.version])[0]
 
 
-class AnonipFilter:
-    def __init__(self, args=None, extra=None, anonip=None):
+class AnonipFilter(logging.Filter):
+    def __init__(self, name="", args=None, extra=None, anonip=None):
         """
         An implementation of Python logging.Filter using anonip.
 
+        :param name: str
         :param args: list of log message args to filter. Defaults to []
         :param extra: list of LogRecord attributes to filter. Defaults to []
         :param anonip: dict of parameters for Anonip instance
         """
+        super(AnonipFilter, self).__init__(name)
         self.args = [] if args is None else args
         self.extra = [] if extra is None else extra
         self.anonip = Anonip(**(anonip or {}))
 
+    def _set_args_attr(self, args, key):
+        value = args[key]
+        if not isinstance(value, str):
+            return args
+
+        orig_type = type(args)
+        temp_type = list
+        if isinstance(args, abc.Mapping):
+            temp_type = dict
+        has_setitem = hasattr(args, "__setitem__")
+        if not has_setitem:
+            args = temp_type(args)
+        ip = self.anonip.extract_ip(value)[1]
+        args[key] = str(self.anonip.process_ip(ip))
+        if not has_setitem:
+            args = orig_type(args)
+        return args
+
     def filter(self, record):
         """
-        See logging.Filter.filter()
+         Apply anonip IP masking.
+
+        :param record: logging.LogRecord
+        :return: bool
         """
-        if record.name != "anonip":
-            for key in self.args:
-                if isinstance(record.args, abc.Mapping):
-                    if key in record.args:
-                        value = record.args[key]
-                        if isinstance(value, str):
-                            record.args[key] = self.anonip.process_line(value)
-                elif isinstance(record.args, abc.Sequence):
-                    if key < len(record.args):
-                        value = record.args[key]
-                        if isinstance(value, str):
-                            is_tuple = isinstance(record.args, tuple)
-                            if is_tuple:
-                                record.args = list(record.args)
-                            record.args[key] = self.anonip.process_line(value)
-                            if is_tuple:
-                                record.args = tuple(record.args)
-
-            for key in self.extra:
-                if hasattr(record, key):
-                    value = getattr(record, key)
-                    if (isinstance(value, str)):
-                        setattr(record, key, self.anonip.process_line(value))
+        if not super(AnonipFilter, self).filter(record):
+            return False
+
+        for key in self.args:
+            if isinstance(record.args, abc.Mapping):
+                if key in record.args:
+                    record.args = self._set_args_attr(record.args, key)
+            elif isinstance(record.args, abc.Sequence):
+                if isinstance(key, int) and key < len(record.args):
+                    record.args = self._set_args_attr(record.args, key)
+
+        for key in self.extra:
+            if hasattr(record, key):
+                value = getattr(record, key)
+                if isinstance(value, str):
+                    ip = self.anonip.extract_ip(value)[1]
+                    setattr(record, key, str(self.anonip.process_ip(ip)))
+
+        if not self.args and not self.extra:
+            record.msg = self.anonip.process_line(record.msg)
 
         return True
 

conftest.py

@@ -1,3 +1,4 @@
+import logging
 import sys
 
 import pytest
@@ -8,3 +9,11 @@ def backup_and_restore_sys_argv():
     old_sys_argv = sys.argv
     yield
     sys.argv = old_sys_argv
+
+
+@pytest.fixture()
+def enable_logging():
+    logging.disable(logging.NOTSET)
+    logging.getLogger("anonip").setLevel(logging.CRITICAL)
+    yield
+    logging.disable(logging.CRITICAL)

tests.py

@@ -432,27 +432,44 @@ def test_logging_filter_defaults(caplog):
     logging.disable(logging.CRITICAL)
 
 
-def test_logging_filter_args(caplog):
-    logging.disable(logging.NOTSET)
-    logging.getLogger("anonip").setLevel(logging.CRITICAL)
-
+@pytest.mark.parametrize(
+    "string,args,filter_attr,expected",
+    [
+        ("%(ip)s string", {"ip": "192.168.100.200"}, "ip", "192.168.96.0 string"),
+        ("%s string", "192.168.100.200", 0, "192.168.96.0 string"),
+        (
+            "%(ip)s string",
+            {"ip": "2001:0db8:85a3:0000:0000:8a2e:0370:7334"},
+            "ip",
+            "2001:db8:85a0:: string",
+        ),
+        ("string", None, "ip", "string"),
+        ("string", {"ip": ["in a list"]}, "ip", "string"),
+        ("", "", "", ""),
+    ],
+)
+def test_logging_filter_args(
+    string, args, filter_attr, expected, mocker, caplog, enable_logging
+):
     logger = logging.getLogger("filter_args")
-    logger.addFilter(anonip.AnonipFilter(args=["ip", "non-existing-attr"], extra=[]))
+    logger.addFilter(
+        anonip.AnonipFilter(args=[filter_attr, "non-existing-attr"], extra=[])
+    )
     logger.setLevel(logging.INFO)
 
-    logger.info("%(ip)s string", {"ip": "192.168.100.200"})
-    logger.info("string %(ip)s", {"ip": "1.2.3.4"})
-    logger.info("%(ip)s string", {"ip": "2001:0db8:85a3:0000:0000:8a2e:0370:7334"})
-    logger.info("string")
-
-    assert caplog.record_tuples == [
-        ("filter_args", logging.INFO, "192.168.96.0 string"),
-        ("filter_args", logging.INFO, "string 1.2.0.0"),
-        ("filter_args", logging.INFO, "2001:db8:85a0:: string"),
-        ("filter_args", logging.INFO, "string"),
-    ]
+    log_args = [string]
+    if args:
+        log_args.append(args)
 
-    logging.disable(logging.CRITICAL)
+    if string == args == filter_attr == expected == "":
+        mocker.patch.object(logging.Filter, "filter", return_value=False)
+        logger.info(*log_args)
+        assert len(caplog.record_tuples) == 0
+    else:
+        logger.info(*log_args)
+        assert caplog.record_tuples == [
+            ("filter_args", logging.INFO, expected),
+        ]
 
 
 def test_logging_filter_extra(caplog):

tox.ini

@@ -13,6 +13,7 @@ envlist = py{27,36,37,38,39,310}, pypy3, flake8, black
 deps=
     pytest
     pytest-cov
+    pytest-mock
 commands=pytest -r a -vv tests.py anonip.py
 
 [testenv:flake8]