github
/
fail2ban
mirror of https://github.com/fail2ban/fail2ban
1
0
Fork 0
Code Issues Releases Wiki Activity

ENH: Move traceback formatter to from tests.utils to helpers

Browse Source 
Now allows for tests to be removed from package if desired
pull/700/head
Steven Hiscocks 11 years ago
parent 78c82b3da7
commit 6a740f684a
5 changed files with 91 additions and 84 deletions
Show Stats Download Patch File Download Diff File

2
bin/fail2ban-regex
Unescape View File

@ -45,7 +45,7 @@ from fail2ban.client.filterreader import FilterReader
from fail2ban.server.filter import Filter
from fail2ban.server.failregex import RegexException
from fail2ban.tests.utils import FormatterWithTraceBack
from fail2ban.helpers import FormatterWithTraceBack
# Gets the instance of the logger.
logSys = logging.getLogger("fail2ban")

3
bin/fail2ban-testcases
Unescape View File

@ -34,7 +34,8 @@ if os.path.exists("fail2ban/__init__.py"):
sys.path.insert(0, ".")
from fail2ban.version import version
from fail2ban.tests.utils import FormatterWithTraceBack, gatherTests
from fail2ban.tests.utils import gatherTests
from fail2ban.helpers import FormatterWithTraceBack
from fail2ban.server.mytime import MyTime
from optparse import OptionParser, Option

83
fail2ban/helpers.py
Unescape View File

@ -20,9 +20,90 @@
__author__ = "Cyril Jaquier, Arturo 'Buanzo' Busleiman, Yaroslav Halchenko"
__license__ = "GPL"
import sys
import os
import traceback
import re
import logging
def formatExceptionInfo():
""" Consistently format exception information """
import sys
cla, exc = sys.exc_info()[:2]
return (cla.__name__, str(exc))
#
# Following "traceback" functions are adopted from PyMVPA distributed
# under MIT/Expat and copyright by PyMVPA developers (i.e. me and
# Michael). Hereby I re-license derivative work on these pieces under GPL
# to stay in line with the main Fail2Ban license
#
def mbasename(s):
"""Custom function to include directory name if filename is too common
Also strip .py at the end
"""
base = os.path.basename(s)
if base.endswith('.py'):
base = base[:-3]
if base in set(['base', '__init__']):
base = os.path.basename(os.path.dirname(s)) + '.' + base
return base
class TraceBack(object):
"""Customized traceback to be included in debug messages
"""
def __init__(self, compress=False):
"""Initialize TrackBack metric
Parameters
----------
compress : bool
if True then prefix common with previous invocation gets
replaced with ...
"""
self.__prev = ""
self.__compress = compress
def __call__(self):
ftb = traceback.extract_stack(limit=100)[:-2]
entries = [
[mbasename(x[0]), os.path.dirname(x[0]), str(x[1])] for x in ftb]
entries = [ [e[0], e[2]] for e in entries
if not (e[0] in ['unittest', 'logging.__init__']
or e[1].endswith('/unittest'))]
# lets make it more concise
entries_out = [entries[0]]
for entry in entries[1:]:
if entry[0] == entries_out[-1][0]:
entries_out[-1][1] += ',%s' % entry[1]
else:
entries_out.append(entry)
sftb = '>'.join(['%s:%s' % (mbasename(x[0]),
x[1]) for x in entries_out])
if self.__compress:
# lets remove part which is common with previous invocation
prev_next = sftb
common_prefix = os.path.commonprefix((self.__prev, sftb))
common_prefix2 = re.sub('>[^>]*$', '', common_prefix)
if common_prefix2 != "":
sftb = '...' + sftb[len(common_prefix2):]
self.__prev = prev_next
return sftb
class FormatterWithTraceBack(logging.Formatter):
"""Custom formatter which expands %(tb) and %(tbc) with tracebacks
TODO: might need locking in case of compressed tracebacks
"""
def __init__(self, fmt, *args, **kwargs):
logging.Formatter.__init__(self, fmt=fmt, *args, **kwargs)
compress = '%(tbc)s' in fmt
self._tb = TraceBack(compress=compress)
def format(self, record):
record.tbc = record.tb = self._tb()
return logging.Formatter.format(self, record)

3
fail2ban/tests/misctestcase.py
Unescape View File

@ -32,8 +32,7 @@ import datetime
from glob import glob
from StringIO import StringIO
from .utils import mbasename, TraceBack, FormatterWithTraceBack
from ..helpers import formatExceptionInfo
from ..helpers import formatExceptionInfo, mbasename, TraceBack, FormatterWithTraceBack
from ..server.datetemplate import DatePatternRegex

84
fail2ban/tests/utils.py
Unescape View File

@ -22,90 +22,17 @@ __author__ = "Yaroslav Halchenko"
__copyright__ = "Copyright (c) 2013 Yaroslav Halchenko"
__license__ = "GPL"
import logging, os, re, traceback, time, unittest
from os.path import basename, dirname
import logging
import os
import re
import time
import unittest
from StringIO import StringIO
from ..server.mytime import MyTime
logSys = logging.getLogger(__name__)
#
# Following "traceback" functions are adopted from PyMVPA distributed
# under MIT/Expat and copyright by PyMVPA developers (i.e. me and
# Michael). Hereby I re-license derivative work on these pieces under GPL
# to stay in line with the main Fail2Ban license
#
def mbasename(s):
"""Custom function to include directory name if filename is too common
Also strip .py at the end
"""
base = basename(s)
if base.endswith('.py'):
base = base[:-3]
if base in set(['base', '__init__']):
base = basename(dirname(s)) + '.' + base
return base
class TraceBack(object):
"""Customized traceback to be included in debug messages
"""
def __init__(self, compress=False):
"""Initialize TrackBack metric
Parameters
----------
compress : bool
if True then prefix common with previous invocation gets
replaced with ...
"""
self.__prev = ""
self.__compress = compress
def __call__(self):
ftb = traceback.extract_stack(limit=100)[:-2]
entries = [[mbasename(x[0]), dirname(x[0]), str(x[1])] for x in ftb]
entries = [ [e[0], e[2]] for e in entries
if not (e[0] in ['unittest', 'logging.__init__']
or e[1].endswith('/unittest'))]
# lets make it more concise
entries_out = [entries[0]]
for entry in entries[1:]:
if entry[0] == entries_out[-1][0]:
entries_out[-1][1] += ',%s' % entry[1]
else:
entries_out.append(entry)
sftb = '>'.join(['%s:%s' % (mbasename(x[0]),
x[1]) for x in entries_out])
if self.__compress:
# lets remove part which is common with previous invocation
prev_next = sftb
common_prefix = os.path.commonprefix((self.__prev, sftb))
common_prefix2 = re.sub('>[^>]*$', '', common_prefix)
if common_prefix2 != "":
sftb = '...' + sftb[len(common_prefix2):]
self.__prev = prev_next
return sftb
class FormatterWithTraceBack(logging.Formatter):
"""Custom formatter which expands %(tb) and %(tbc) with tracebacks
TODO: might need locking in case of compressed tracebacks
"""
def __init__(self, fmt, *args, **kwargs):
logging.Formatter.__init__(self, fmt=fmt, *args, **kwargs)
compress = '%(tbc)s' in fmt
self._tb = TraceBack(compress=compress)
def format(self, record):
record.tbc = record.tb = self._tb()
return logging.Formatter.format(self, record)
def mtimesleep():
# no sleep now should be necessary since polling tracks now not only
# mtime but also ino and size
@ -146,7 +73,6 @@ def gatherTests(regexps=None, no_network=False):
if not regexps: # pragma: no cover
tests = unittest.TestSuite()
else: # pragma: no cover
import re
class FilteredTestSuite(unittest.TestSuite):
_regexps = [re.compile(r) for r in regexps]
def addTest(self, suite):

Write Preview
Loading…
Cancel
Save