[Author Prev][Author Next][Thread Prev][Thread Next][Author Index][Thread Index]

[tor-commits] [bridgedb/master] Add unittests for bridgedb.safelog module.



commit 0dc4e5b040e7dbe9714ea5b563caa32182a07cc7
Author: Isis Lovecruft <isis@xxxxxxxxxxxxxx>
Date:   Mon Apr 7 12:28:07 2014 +0000

    Add unittests for bridgedb.safelog module.
    
      * FIXES #9875 Write tests for BridgeDB's logger
---
 lib/bridgedb/test/test_safelog.py |  336 +++++++++++++++++++++++++++++++++++++
 1 file changed, 336 insertions(+)

diff --git a/lib/bridgedb/test/test_safelog.py b/lib/bridgedb/test/test_safelog.py
new file mode 100644
index 0000000..bc92305
--- /dev/null
+++ b/lib/bridgedb/test/test_safelog.py
@@ -0,0 +1,336 @@
+# -*- coding: utf-8 -*-
+
+"""Unittests for :mod:`bridgedb.safelog`."""
+
+import re
+
+from twisted.internet import defer
+from twisted.test.proto_helpers import StringTransport
+from twisted.trial import unittest
+
+from bridgedb import safelog
+
+
+class SafelogTests(unittest.TestCase):
+    """Tests for functions and attributes in :mod:`bridgedb.safelog`."""
+
+    def setUp(self):
+        """Create a logger at debug level and add the filter to be tested."""
+        self.logfile = StringTransport()
+        self.handler = safelog.logging.StreamHandler(self.logfile)
+        self.logger = safelog.logging.getLogger(self.__class__.__name__)
+        self.logger.setLevel(10)
+        self.logger.addHandler(self.handler)
+        self.sensitiveData = 'Nicholas Bourbaki'
+
+    def tearDown(self):
+        """Rewind and truncate the logfile so that we have an empty one."""
+        self.logfile.clear()
+
+    def test_setSafeLogging_off(self):
+        """Calls to ``logSafely()`` should return the original data when
+        ``safe_logging`` is disabled.
+        """
+        safelog.setSafeLogging(False)
+        self.logger.warn("Got a connection from %s..."
+                         % safelog.logSafely(self.sensitiveData))
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)
+        #self.assertSubstring("Got a connection from", contents)
+        #self.assertSubstring(self.sensitiveData, contents)
+        #self.failIfSubstring("[scrubbed]", contents)
+
+    def test_setSafeLogging_on(self):
+        """Calls to ``logSafely()`` should return ``"[scrubbed]"`` for any
+        arbitrary data when ``safe_logging`` is enabled.
+        """
+        safelog.setSafeLogging(True)
+        self.logger.warn("Got a connection from %s..."
+                         % safelog.logSafely(self.sensitiveData))
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)
+        #self.assertSubstring("Got a connection from", contents)
+        #self.failIfSubstring(self.sensitiveData, contents)
+        #self.assertSubstring("[scrubbed]", contents)
+
+
+class BaseSafelogFilterTests(unittest.TestCase):
+    """Unittests for :class:`bridgedb.safelog.BaseSafelogFilter`."""
+
+    def setUp(self):
+        safelog.setSafeLogging(True)
+        self.logfile = StringTransport()
+        self.handler = safelog.logging.StreamHandler(self.logfile)
+        self.logger = safelog.logging.getLogger(self.__class__.__name__)
+        self.logger.setLevel(10)
+        self.logger.addHandler(self.handler)
+        self.filter = safelog.BaseSafelogFilter()
+        self.logger.addFilter(self.filter)
+
+        self.logMessage = "testing 1 2 3"
+        self.record = safelog.logging.LogRecord('name', 10, __file__, 1337,
+                                                self.logMessage, {}, None)
+
+    def test_doubleCheck(self):
+        """BaseSafelogFilter.doubleCheck() should always return True."""
+        checked = self.filter.doubleCheck(self.logMessage)
+        self.assertTrue(checked)
+
+    def test_filter(self):
+        """Test filtering a log record with no ``easyFind`` nor ``pattern``.
+
+        The ``LogRecord.message`` shouldn't change.
+        """
+        filtered = self.filter.filter(self.record)
+        self.assertEqual(filtered.getMessage(), self.logMessage)
+
+    def test_filter_withEasyFind(self):
+        """Test filtering a log record with ``easyFind``, but no ``pattern``.
+
+        The ``LogRecord.message`` shouldn't change.
+        """
+        self.filter.easyFind = "2"
+        filtered = self.filter.filter(self.record)
+        self.assertEqual(filtered.getMessage(), self.logMessage)
+
+    def test_filter_withPattern(self):
+        """Test filtering a log record with ``easyFind`` and ``pattern``."""
+        self.filter.easyFind = "2"
+        self.filter.pattern = re.compile("1 2 3")
+        filtered = self.filter.filter(self.record)
+        self.assertEqual(filtered.msg, "testing [scrubbed]")
+
+
+class SafelogEmailFilterTests(unittest.TestCase):
+    """Unittests for :class:`bridgedb.safelog.SafelogEmailFilter`."""
+
+    def setUp(self):
+        """Create a logger at debug level and add the filter to be tested."""
+        self.logfile = StringTransport()
+        self.handler = safelog.logging.StreamHandler(self.logfile)
+        self.filter = safelog.SafelogEmailFilter()
+        self.logger = safelog.logging.getLogger(self.__class__.__name__)
+        self.logger.setLevel(10)
+        self.logger.addHandler(self.handler)
+        self.logger.addFilter(self.filter)
+        self.s1 = "Here is an email address: "
+        self.s2 = "blackhole@xxxxxxxxxxxxxx"
+
+    def test_filter_withPattern(self):
+        """Test filtering a log record with ``easyFind`` and ``pattern``."""
+        record = safelog.logging.LogRecord('name', 10, __file__, 1337,
+                                           "testing blackhole@xxxxxxxxxxxxxx",
+                                           {}, None)
+        filtered = self.filter.filter(record)
+        self.assertEqual(filtered.msg, "testing [scrubbed]")
+
+    def test_debugLevel(self):
+        self.logger.debug("%s %s" % (self.s1, self.s2))
+        self.logfile.io.seek(0)
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)
+        # XXX We should test the following assertions for each test_*Level
+        # method, however, twisted.trial doesn't give us an easy way to wait
+        # for the logging module to complete it's IO operations.
+        #self.assertSubstring(self.s1, contents)
+        #self.failIfSubstring(self.s2, contents)
+
+    def test_infoLevel(self):
+        self.logger.info("%s %s" % (self.s1, self.s2))
+        self.logfile.io.seek(0)
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)
+
+    def test_warnLevel(self):
+        self.logger.warn("%s %s" % (self.s1, self.s2))
+        self.logfile.io.seek(0)
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)
+
+    def test_errorLevel(self):
+        self.logger.error("%s %s" % (self.s1, self.s2))
+        self.logfile.io.seek(0)
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)
+
+    def test_exceptionLevel(self):
+        try:
+            raise Exception("%s %s" % (self.s1, self.s2))
+        except Exception as error:
+            self.logger.exception(error)
+
+        self.logfile.io.seek(0)
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)
+        #self.assertSubstring(self.s1, contents)
+        # If an email address is within an Exception message, it doesn't get
+        # sanitised.
+        #self.assertSubstring(self.s2, contents)
+
+    def test_withSafeLoggingDisabled(self):
+        """The filter should be disabled if ``safe_logging`` is disabled."""
+        safelog.setSafeLogging(False)
+        self.logger.info("%s %s" % (self.s1, self.s2))
+        self.logfile.io.flush()
+        self.logfile.io.seek(0)
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)
+        #self.assertSubstring(self.s1, contents)
+        #self.assertSubstring(self.s2, contents)
+
+
+class SafelogIPv4FilterTests(unittest.TestCase):
+    """Unittests for :class:`bridgedb.safelog.SafelogIPv4Filter`."""
+
+    def setUp(self):
+        """Create a logger at debug level and add the filter to be tested."""
+        self.logfile = StringTransport()
+        self.handler = safelog.logging.StreamHandler(self.logfile)
+        self.filter = safelog.SafelogIPv4Filter()
+        self.logger = safelog.logging.getLogger(str(self.__class__.__name__))
+        self.logger.addHandler(self.handler)
+        self.logger.addFilter(self.filter)
+        self.logger.setLevel(10)
+        self.s1 = "There's an IPv4 address at the end of this book: "
+        self.s2 = "1.2.3.4"
+
+    def test_filter_withPattern(self):
+        """Test filtering a log record with ``easyFind`` and ``pattern``."""
+        record = safelog.logging.LogRecord('name', 10, __file__, 1337,
+                                           "testing 1.2.3.4",
+                                           {}, None)
+        filtered = self.filter.filter(record)
+        self.assertIsInstance(filtered, safelog.logging.LogRecord)
+
+    def test_doubleCheck_IPv4(self):
+        checked = self.filter.doubleCheck("1.2.3.4")
+        self.assertIs(checked, True)
+
+    def test_doubleCheck_IPv6(self):
+        checked = self.filter.doubleCheck("2af1:a470:9b36::a1:3:82")
+        self.assertIsNot(checked, True)
+
+    def test_debugLevel(self):
+        self.logger.debug("%s %s" % (self.s1, self.s2))
+        self.logfile.io.seek(0)
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)
+
+    def test_infoLevel(self):
+        self.logger.info("%s %s" % (self.s1, self.s2))
+        self.logfile.io.seek(0)
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)
+
+    def test_warnLevel(self):
+        self.logger.warn("%s %s" % (self.s1, self.s2))
+        self.logfile.io.seek(0)
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)
+
+    def test_errorLevel(self):
+        self.logger.error("%s %s" % (self.s1, self.s2))
+        self.logfile.io.seek(0)
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)
+
+    def test_exceptionLevel(self):
+        try:
+            raise Exception("%s %s" % (self.s1, self.s2))
+        except Exception as error:
+            self.logger.exception(error)
+
+        self.logfile.io.flush()
+        self.logfile.io.seek(0)
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)
+
+    def test_withSafeLoggingDisabled(self):
+        """The filter should be disabled if ``safe_logging`` is disabled."""
+        safelog.setSafeLogging(False)
+        self.logger.info("%s %s" % (self.s1, self.s2))
+        self.logfile.io.flush()
+        self.logfile.io.seek(0)
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)
+
+
+class SafelogIPv6FilterTests(unittest.TestCase):
+    """Unittests for :class:`bridgedb.safelog.SafelogIPv6Filter`."""
+
+    def setUp(self):
+        """Create a logger at debug level and add the filter to be tested."""
+        self.logfile = StringTransport()
+        self.handler = safelog.logging.StreamHandler(self.logfile)
+        self.filter = safelog.SafelogIPv6Filter()
+        self.logger = safelog.logging.getLogger(str(self.__class__.__name__))
+        self.logger.addHandler(self.handler)
+        self.logger.addFilter(self.filter)
+        self.logger.setLevel(10)
+        self.s1 = "There's an IPv6 address at the end of this book: "
+        self.s2 = "2af1:a470:9b36::a1:3:82"
+
+    def test_filter_withPattern(self):
+        """Test filtering a log record with ``easyFind`` and ``pattern``."""
+        record = safelog.logging.LogRecord('name', 10, __file__, 1337,
+                                           "2af1:a470:9b36::a1:3:82",
+                                           {}, None)
+        filtered = self.filter.filter(record)
+        self.assertIsInstance(filtered, safelog.logging.LogRecord)
+
+    def test_doubleCheck_IPv4(self):
+        checked = self.filter.doubleCheck("1.2.3.4")
+        self.assertIsNot(checked, True)
+
+    def test_doubleCheck_IPv6(self):
+        checked = self.filter.doubleCheck("2af1:a470:9b36::a1:3:82")
+        self.assertIs(checked, True)
+
+    def test_debugLevel(self):
+        self.logger.debug("%s %s" % (self.s1, self.s2))
+        self.logfile.io.seek(0)
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)
+
+    def test_infoLevel(self):
+        self.logger.info("%s %s" % (self.s1, self.s2))
+        self.logfile.io.seek(0)
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)
+
+    def test_warnLevel(self):
+        self.logger.warn("%s %s" % (self.s1, self.s2))
+        self.logfile.io.seek(0)
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)
+
+    def test_errorLevel(self):
+        self.logger.error("%s %s" % (self.s1, self.s2))
+        self.logfile.io.seek(0)
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)
+
+    def test_exceptionLevel(self):
+        try:
+            raise Exception("%s %s" % (self.s1, self.s2))
+        except Exception as error:
+            self.logger.exception(error)
+
+        self.logfile.io.flush()
+        self.logfile.io.seek(0)
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)
+        #self.assertSubstring(self.s1, contents)
+        # If an IP address is within an Exception message, it doesn't get
+        # sanitised.
+        #self.assertSubstring(self.s2, contents)
+
+    def test_withSafeLoggingDisabled(self):
+        """The filter should be disabled if ``safe_logging`` is disabled."""
+        safelog.setSafeLogging(False)
+        self.logger.info("%s %s" % (self.s1, self.s2))
+
+        self.logfile.io.flush()
+        self.logfile.io.seek(0)
+        contents = self.logfile.value()
+        self.assertIsNotNone(contents)



_______________________________________________
tor-commits mailing list
tor-commits@xxxxxxxxxxxxxxxxxxxx
https://lists.torproject.org/cgi-bin/mailman/listinfo/tor-commits