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

code review after enhancements of Yaroslav

Browse Source
pull/1483/head
sebres 8 years ago
parent 3c42b4f56e
commit 7242c9cbde
2 changed files with 60 additions and 68 deletions
Show Stats Download Patch File Download Diff File

4
bin/fail2ban-testcases
Unescape View File

@ -30,7 +30,7 @@ import sys
import time
import unittest
# Check if local fail2ban module exists, and use if it exists by
# Check if local fail2ban module exists, and use if it exists by
# modifying the path. This is such that tests can be used in dev
# environment.
if os.path.exists("fail2ban/__init__.py"):
@ -122,7 +122,7 @@ if verbosity > 1: # pragma: no cover
if verbosity > 3:
fmt = ' | %(module)15.15s-%(levelno)-2d: %(funcName)-20.20s |' + fmt
if verbosity > 2:
fmt = ' +%(relativeCreated)5d %(thread)X %(name)-25s %(levelname)-5.5s' + fmt
fmt = ' +%(relativeCreated)5d %(thread)X %(name)-25.25s %(levelname)-5.5s' + fmt
else:
fmt = ' %(asctime)-15s %(thread)X %(levelname)-5.5s' + fmt
#

124
fail2ban/tests/fail2banclienttestcase.py
Unescape View File

@ -192,9 +192,9 @@ def _kill_srv(pidfile):
logSys.debug("cleanup: %r", (pidfile, isdir(pidfile)))
if isdir(pidfile):
piddir = pidfile
pidfile = piddir + "/f2b.pid"
if not isfile(pidfile):
pidfile = piddir + "/fail2ban.pid"
pidfile = pjoin(piddir, "f2b.pid")
if not isfile(pidfile): # pragma: no cover
pidfile = pjoin(piddir, "fail2ban.pid")
if not isfile(pidfile):
logSys.debug("cleanup: no pidfile for %r", piddir)
@ -243,6 +243,8 @@ def with_kill_srv(f):
class Fail2banClientServerBase(LogCaptureTestCase):
_orig_exit = Fail2banCmdLine._exit
def setUp(self):
"""Call before every test case."""
LogCaptureTestCase.setUp(self)
@ -253,8 +255,6 @@ class Fail2banClientServerBase(LogCaptureTestCase):
Fail2banCmdLine._exit = self._orig_exit
LogCaptureTestCase.tearDown(self)
_orig_exit = Fail2banCmdLine._exit
@staticmethod
def _test_exit(code=0):
if code == 0:
@ -288,10 +288,10 @@ class Fail2banClientServerBase(LogCaptureTestCase):
logSys.debug("No log file %s to examine details of error", log)
raise
def assertExitsNormally(self, args):
def execSuccess(self, startparams, *args):
raise NotImplementedError("To be defined in subclass")
def assertExitsAbnormally(self, args):
def execFailed(self, startparams, *args):
raise NotImplementedError("To be defined in subclass")
#
@ -301,7 +301,7 @@ class Fail2banClientServerBase(LogCaptureTestCase):
# start and wait to end (foreground):
logSys.debug("start of test worker")
phase['start'] = True
self.assertExitsNormally(("-f",) + startparams + ("start",))
self.execSuccess(("-f",) + startparams, "start")
# end :
phase['end'] = True
logSys.debug("end of test worker")
@ -330,13 +330,13 @@ class Fail2banClientServerBase(LogCaptureTestCase):
self._wait_for_srv(tmp, True, startparams=startparams)
self.pruneLog()
# several commands to server:
self.assertExitsNormally(startparams + ("ping",))
self.assertExitsAbnormally(startparams + ("~~unknown~cmd~failed~~",))
self.assertExitsNormally(startparams + ("echo", "TEST-ECHO",))
self.execSuccess(startparams, "ping")
self.execFailed(startparams, "~~unknown~cmd~failed~~")
self.execSuccess(startparams, "echo", "TEST-ECHO")
finally:
self.pruneLog()
# stop:
self.assertExitsNormally(startparams + ("stop",))
self.execSuccess(startparams, "stop")
# wait for end:
Utils.wait_for(lambda: phase.get('end', None) is not None, MAX_WAITTIME)
self.assertTrue(phase.get('end', None))
@ -349,30 +349,31 @@ class Fail2banClientServerBase(LogCaptureTestCase):
class Fail2banClientTest(Fail2banClientServerBase):
def assertExitsNormally(self, args):
self.assertRaises(ExitException, _exec_client, ((CLIENT,) + args))
def execSuccess(self, startparams, *args):
self.assertRaises(ExitException, _exec_client,
((CLIENT,) + startparams + args))
def assertExitsAbnormally(self, args):
self.assertRaises(FailExitException, _exec_client, ((CLIENT,) + args))
def execFailed(self, startparams, *args):
self.assertRaises(FailExitException, _exec_client,
((CLIENT,) + startparams + args))
def testConsistency(self):
self.assertTrue(isfile(pjoin(BIN, CLIENT)))
self.assertTrue(isfile(pjoin(BIN, SERVER)))
def testClientUsage(self):
self.assertExitsNormally(("-h",))
self.execSuccess((), "-h")
self.assertLogged("Usage: " + CLIENT)
self.assertLogged("Report bugs to ")
self.pruneLog()
self.assertExitsNormally(("-vq", "-V",))
self.execSuccess((), "-vq", "-V")
self.assertLogged("Fail2Ban v" + fail2bancmdline.version)
@with_tmpdir
def testClientDump(self, tmp):
# use here the stock configuration (if possible)
startparams = _start_params(tmp, True)
self.assertExitsNormally(
(startparams + ("-vvd",)))
self.execSuccess(startparams, "-vvd")
self.assertLogged("Loading files")
self.assertLogged("logtarget")
@ -382,34 +383,28 @@ class Fail2banClientTest(Fail2banClientServerBase):
# use once the stock configuration (to test starting also)
startparams = _start_params(tmp, True)
# start:
self.assertExitsNormally(
("-b",) + startparams + ("start",))
self.execSuccess(("-b",) + startparams, "start")
# wait for server (socket and ready):
self._wait_for_srv(tmp, True, startparams=startparams)
self.assertLogged("Server ready")
self.assertLogged("Exit with code 0")
try:
self.assertExitsNormally(
startparams + ("echo", "TEST-ECHO",))
self.assertExitsAbnormally(
startparams + ("~~unknown~cmd~failed~~",))
self.execSuccess(startparams, "echo", "TEST-ECHO")
self.execFailed(startparams, "~~unknown~cmd~failed~~")
self.pruneLog()
# start again (should fail):
self.assertExitsAbnormally(
("-b",) + startparams + ("start",))
self.execFailed(("-b",) + startparams, "start")
self.assertLogged("Server already running")
finally:
self.pruneLog()
# stop:
self.assertExitsNormally(
startparams + ("stop",))
self.execSuccess(startparams, "stop")
self.assertLogged("Shutdown successful")
self.assertLogged("Exit with code 0")
self.pruneLog()
# stop again (should fail):
self.assertExitsAbnormally(
startparams + ("stop",))
self.execFailed(startparams, "stop")
self.assertLogged("Failed to access socket path")
self.assertLogged("Is fail2ban running?")
@ -430,7 +425,7 @@ class Fail2banClientTest(Fail2banClientServerBase):
self.pruneLog()
try:
# echo from client (inside):
self.assertExitsNormally(startparams + ("echo", "TEST-ECHO",))
self.execSuccess(startparams, "echo", "TEST-ECHO")
self.assertLogged("TEST-ECHO")
self.assertLogged("Exit with code 0")
self.pruneLog()
@ -440,8 +435,7 @@ class Fail2banClientTest(Fail2banClientServerBase):
"status",
"exit"
]
self.assertExitsNormally(
startparams + ("-i",))
self.execSuccess(startparams, "-i")
self.assertLogged("INTERACT-ECHO")
self.assertLogged("Status", "Number of jail:")
self.assertLogged("Exit with code 0")
@ -452,8 +446,7 @@ class Fail2banClientTest(Fail2banClientServerBase):
"restart",
"exit"
]
self.assertExitsNormally(
startparams + ("-i",))
self.execSuccess(startparams, "-i")
self.assertLogged("Reading config files:")
self.assertLogged("Shutdown successful")
self.assertLogged("Server ready")
@ -464,20 +457,18 @@ class Fail2banClientTest(Fail2banClientServerBase):
"reload ~~unknown~jail~fail~~",
"exit"
]
self.assertExitsNormally(
startparams + ("-i",))
self.execSuccess(startparams, "-i")
self.assertLogged("Failed during configuration: No section: '~~unknown~jail~fail~~'")
self.pruneLog()
# test reload missing jail (direct):
self.assertExitsAbnormally(
startparams + ("reload", "~~unknown~jail~fail~~"))
self.execFailed(startparams, "reload", "~~unknown~jail~fail~~")
self.assertLogged("Failed during configuration: No section: '~~unknown~jail~fail~~'")
self.assertLogged("Exit with code -1")
self.pruneLog()
finally:
self.pruneLog()
# stop:
self.assertExitsNormally(startparams + ("stop",))
self.execSuccess(startparams, "stop")
self.assertLogged("Shutdown successful")
self.assertLogged("Exit with code 0")
@ -488,34 +479,33 @@ class Fail2banClientTest(Fail2banClientServerBase):
startparams = _start_params(tmp, logtarget="INHERITED")
## wrong config directory
self.assertExitsAbnormally(
("--async", "-c", pjoin(tmp, "miss"), "start",))
self.execFailed((),
"--async", "-c", pjoin(tmp, "miss"), "start")
self.assertLogged("Base configuration directory " + pjoin(tmp, "miss") + " does not exist")
self.pruneLog()
## wrong socket
self.assertExitsAbnormally(
("--async", "-c", pjoin(tmp, "config"), "-s", pjoin(tmp, "miss/f2b.sock"), "start",))
self.execFailed((),
"--async", "-c", pjoin(tmp, "config"), "-s", pjoin(tmp, "miss/f2b.sock"), "start")
self.assertLogged("There is no directory " + pjoin(tmp, "miss") + " to contain the socket file")
self.pruneLog()
## not running
self.assertExitsAbnormally(
("-c", pjoin(tmp, "config"), "-s", pjoin(tmp, "f2b.sock"), "reload",))
self.execFailed((),
"-c", pjoin(tmp, "config"), "-s", pjoin(tmp, "f2b.sock"), "reload")
self.assertLogged("Could not find server")
self.pruneLog()
## already exists:
open(pjoin(tmp, "f2b.sock"), 'a').close()
self.assertExitsAbnormally(
("--async", "-c", pjoin(tmp, "config"), "-s", pjoin(tmp, "f2b.sock"), "start",))
self.execFailed((),
"--async", "-c", pjoin(tmp, "config"), "-s", pjoin(tmp, "f2b.sock"), "start")
self.assertLogged("Fail2ban seems to be in unexpected state (not running but the socket exists)")
self.pruneLog()
os.remove(pjoin(tmp, "f2b.sock"))
## wrong option:
self.assertExitsAbnormally(
("-s",))
self.execFailed((), "-s")
self.assertLogged("Usage: ")
self.pruneLog()
@ -533,14 +523,16 @@ class Fail2banClientTest(Fail2banClientServerBase):
class Fail2banServerTest(Fail2banClientServerBase):
def assertExitsNormally(self, args):
self.assertRaises(ExitException, _exec_server, ((SERVER,) + args))
def execSuccess(self, startparams, *args):
self.assertRaises(ExitException, _exec_server,
((SERVER,) + startparams + args))
def assertExitsAbnormally(self, args):
self.assertRaises(FailExitException, _exec_server, ((SERVER,) + args))
def execFailed(self, startparams, *args):
self.assertRaises(FailExitException, _exec_server,
((SERVER,) + startparams + args))
def testServerUsage(self):
self.assertExitsNormally(("-h",))
self.execSuccess((), "-h")
self.assertLogged("Usage: " + SERVER)
self.assertLogged("Report bugs to ")
@ -560,12 +552,12 @@ class Fail2banServerTest(Fail2banClientServerBase):
self.assertLogged("Server ready")
self.pruneLog()
try:
self.assertExitsNormally(startparams + ("echo", "TEST-ECHO",))
self.assertExitsAbnormally(startparams + ("~~unknown~cmd~failed~~",))
self.execSuccess(startparams, "echo", "TEST-ECHO")
self.execFailed(startparams, "~~unknown~cmd~failed~~")
finally:
self.pruneLog()
# stop:
self.assertExitsNormally(startparams + ("stop",))
self.execSuccess(startparams, "stop")
self.assertLogged("Shutdown successful")
self.assertLogged("Exit with code 0")
@ -576,21 +568,21 @@ class Fail2banServerTest(Fail2banClientServerBase):
startparams = _start_params(tmp, logtarget="INHERITED")
## wrong config directory
self.assertExitsAbnormally(
("-c", pjoin(tmp, "miss"),))
self.execFailed((),
"-c", pjoin(tmp, "miss"))
self.assertLogged("Base configuration directory " + pjoin(tmp, "miss") + " does not exist")
self.pruneLog()
## wrong socket
self.assertExitsAbnormally(
("-c", pjoin(tmp, "config"), "-x", "-s", pjoin(tmp, "miss/f2b.sock"),))
self.execFailed((),
"-c", pjoin(tmp, "config"), "-x", "-s", pjoin(tmp, "miss/f2b.sock"))
self.assertLogged("There is no directory " + pjoin(tmp, "miss") + " to contain the socket file")
self.pruneLog()
## already exists:
open(pjoin(tmp, "f2b.sock"), 'a').close()
self.assertExitsAbnormally(
("-c", pjoin(tmp, "config"), "-s", pjoin(tmp, "f2b.sock"),))
self.execFailed((),
"-c", pjoin(tmp, "config"), "-s", pjoin(tmp, "f2b.sock"))
self.assertLogged("Fail2ban seems to be in unexpected state (not running but the socket exists)")
self.pruneLog()
os.remove(pjoin(tmp, "f2b.sock"))

Write Preview
Loading…
Cancel
Save