Gecko [ fernandoquevedodop.com ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
action_d	[ DIR ]	drwxr-xr-x
config	[ DIR ]	drwxr-xr-x
files	[ DIR ]	drwxr-xr-x
__init__.py	974 B	-rw-r--r--
actionstestcase.py	18.4 KB	-rw-r--r--
actiontestcase.py	23.46 KB	-rw-r--r--
banmanagertestcase.py	9.56 KB	-rw-r--r--
clientbeautifiertestcase.py	9.05 KB	-rw-r--r--
clientreadertestcase.py	41.59 KB	-rw-r--r--
databasetestcase.py	22.47 KB	-rw-r--r--
datedetectortestcase.py	26.33 KB	-rw-r--r--
dummyjail.py	2.02 KB	-rw-r--r--
fail2banclienttestcase.py	58.69 KB	-rw-r--r--
fail2banregextestcase.py	21.1 KB	-rw-r--r--
failmanagertestcase.py	8.76 KB	-rw-r--r--
filtertestcase.py	81.07 KB	-rw-r--r--
misctestcase.py	17.58 KB	-rw-r--r--
observertestcase.py	22.46 KB	-rw-r--r--
samplestestcase.py	12.2 KB	-rw-r--r--
servertestcase.py	87.43 KB	-rw-r--r--
sockettestcase.py	7.63 KB	-rw-r--r--
tickettestcase.py	7.13 KB	-rw-r--r--
utils.py	29.72 KB	-rw-r--r--

Code Editor : actiontestcase.py

# emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: t -*-
# vi: set ft=python sts=4 ts=4 sw=4 noet :

# This file is part of Fail2Ban.
#
# Fail2Ban is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# Fail2Ban is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Fail2Ban; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.

# Author: Cyril Jaquier
#

__author__ = "Cyril Jaquier"
__copyright__ = "Copyright (c) 2004 Cyril Jaquier"
__license__ = "GPL"

import os
import tempfile
import time
import unittest

from ..server.action import CommandAction, CallingMap, substituteRecursiveTags
from ..server.actions import OrderedDict, Actions
from ..server.utils import Utils

from .dummyjail import DummyJail
from .utils import pid_exists, with_tmpdir, LogCaptureTestCase

class CommandActionTest(LogCaptureTestCase):

def setUp(self):
		"""Call before every test case."""
		LogCaptureTestCase.setUp(self)
		self.__action = CommandAction(None, "Test")
		# prevent execute stop if start fails (or event not started at all):
		self.__action_started = False
		orgstart = self.__action.start
		def _action_start():
			self.__action_started = True
			return orgstart()
		self.__action.start = _action_start

def tearDown(self):
		"""Call after every test case."""
		if self.__action_started:
			self.__action.stop()
		LogCaptureTestCase.tearDown(self)

def testSubstituteRecursiveTags(self):
		aInfo = {
			'HOST': "192.0.2.0",
			'ABC': "123 <HOST>",
			'xyz': "890 <ABC>",
		}
		# Recursion is bad
		self.assertRaises(ValueError,
			lambda: substituteRecursiveTags({'A': '<A>'}))
		self.assertRaises(ValueError,
			lambda: substituteRecursiveTags({'A': '', 'B': '<A>'}))
		self.assertRaises(ValueError,
			lambda: substituteRecursiveTags({'A': '', 'B': '<C>', 'C': '<A>'}))
		# Unresolveable substition
		self.assertRaises(ValueError,
			lambda: substituteRecursiveTags({'A': 'to= fromip=<IP>', 'C': '', 'B': '<C>', 'D': ''}))
		self.assertRaises(ValueError,
			lambda: substituteRecursiveTags({'failregex': 'to=<honeypot> fromip=<IP>', 'sweet': '<honeypot>', 'honeypot': '<sweet>', 'ignoreregex': ''}))
		# We need here an ordered, because the sequence of iteration is very important for this test
		if OrderedDict:
			# No cyclic recursion, just multiple replacement of tag <T>, should be successful:
			self.assertEqual(substituteRecursiveTags( OrderedDict(
					(('X', 'x=x<T>'), ('T', '1'), ('Z', '<X> <T> <Y>'), ('Y', 'y=y<T>')))
				), {'X': 'x=x1', 'T': '1', 'Y': 'y=y1', 'Z': 'x=x1 1 y=y1'}
			)
			# No cyclic recursion, just multiple replacement of tag <T> in composite tags, should be successful:
			self.assertEqual(substituteRecursiveTags( OrderedDict(
				 (('X', 'x=x<T> <Z> <<R1>> <<R2>>'), ('R1', 'Z'), ('R2', 'Y'), ('T', '1'), ('Z', '<T> <Y>'), ('Y', 'y=y<T>')))
				), {'X': 'x=x1 1 y=y1 1 y=y1 y=y1', 'R1': 'Z', 'R2': 'Y', 'T': '1', 'Z': '1 y=y1', 'Y': 'y=y1'}
			)
			# No cyclic recursion, just multiple replacement of same tags, should be successful:
			self.assertEqual(substituteRecursiveTags( OrderedDict((
					('actionstart', 'ipset create <ipmset> hash:ip timeout <bantime> family <ipsetfamily>\n<iptables> -I <chain> <actiontype>'),
					('ipmset', 'f2b-<name>'),
					('name', 'any'),
					('bantime', '600'),
					('ipsetfamily', 'inet'),
					('iptables', 'iptables <lockingopt>'),
					('lockingopt', '-w'),
					('chain', 'INPUT'),
					('actiontype', '<multiport>'),
					('multiport', '-p <protocol> -m multiport --dports <port> -m set --match-set <ipmset> src -j <blocktype>'),
					('protocol', 'tcp'),
					('port', 'ssh'),
					('blocktype', 'REJECT',),
				))
				), OrderedDict((
					('actionstart', 'ipset create f2b-any hash:ip timeout 600 family inet\niptables -w -I INPUT -p tcp -m multiport --dports ssh -m set --match-set f2b-any src -j REJECT'),
					('ipmset', 'f2b-any'),
					('name', 'any'),
					('bantime', '600'),
					('ipsetfamily', 'inet'),
					('iptables', 'iptables -w'),
					('lockingopt', '-w'),
					('chain', 'INPUT'),
					('actiontype', '-p tcp -m multiport --dports ssh -m set --match-set f2b-any src -j REJECT'),
					('multiport', '-p tcp -m multiport --dports ssh -m set --match-set f2b-any src -j REJECT'),
					('protocol', 'tcp'),
					('port', 'ssh'),
					('blocktype', 'REJECT')
				))
			)
			# Cyclic recursion by composite tag creation, tags "create" another tag, that closes cycle:
			self.assertRaises(ValueError, lambda: substituteRecursiveTags( OrderedDict((
					('A', '<<C>>'),
					('B', 'D'), ('C', 'E'),
					('DE', 'cycle <A>'),
			)) ))
			self.assertRaises(ValueError, lambda: substituteRecursiveTags( OrderedDict((
					('DE', 'cycle <A>'),
					('A', '<<C>>'),
					('B', 'D'), ('C', 'E'),
			)) ))
			
		# missing tags are ok
		self.assertEqual(substituteRecursiveTags({'A': '<C>'}), {'A': '<C>'})
		self.assertEqual(substituteRecursiveTags({'A': '<C> <D> <X>','X':'fun'}), {'A': '<C> <D> fun', 'X':'fun'})
		self.assertEqual(substituteRecursiveTags({'A': '<C> ', 'B': 'cool'}), {'A': '<C> cool', 'B': 'cool'})
		# Escaped tags should be ignored
		self.assertEqual(substituteRecursiveTags({'A': '<matches> ', 'B': 'cool'}), {'A': '<matches> cool', 'B': 'cool'})
		# Multiple stuff on same line is ok
		self.assertEqual(substituteRecursiveTags({'failregex': 'to=<honeypot> fromip=<IP> evilperson=<honeypot>', 'honeypot': 'pokie', 'ignoreregex': ''}),
								{ 'failregex': "to=pokie fromip=<IP> evilperson=pokie",
									'honeypot': 'pokie',
									'ignoreregex': '',
								})
		# rest is just cool
		self.assertEqual(substituteRecursiveTags(aInfo),
								{ 'HOST': "192.0.2.0",
									'ABC': '123 192.0.2.0',
									'xyz': '890 123 192.0.2.0',
								})
		# obscure embedded case
		self.assertEqual(substituteRecursiveTags({'A': '<<PREF>HOST>', 'PREF': 'IPV4'}),
						 {'A': '<IPV4HOST>', 'PREF': 'IPV4'})
		self.assertEqual(substituteRecursiveTags({'A': '<<PREF>HOST>', 'PREF': 'IPV4', 'IPV4HOST': '1.2.3.4'}),
						 {'A': '1.2.3.4', 'PREF': 'IPV4', 'IPV4HOST': '1.2.3.4'})
		# more embedded within a string and two interpolations
		self.assertEqual(substituteRecursiveTags({'A': 'A <IP<PREF>HOST> B IP<PREF> C', 'PREF': 'V4', 'IPV4HOST': '1.2.3.4'}),
						 {'A': 'A 1.2.3.4 B IPV4 C', 'PREF': 'V4', 'IPV4HOST': '1.2.3.4'})

def testSubstRec_DontTouchUnusedCallable(self):
		cm = CallingMap({
			'A':0,
			'B':lambda self: '<A><A>',
			'C':'',
			'D':''
		})
		#
		# should raise no exceptions:
		substituteRecursiveTags(cm)
		# add exception tag:
		cm['C'] = lambda self,i=0: 5 // int(self['A']) # raise error by access
		# test direct get of callable (should raise an error):
		self.assertRaises(ZeroDivisionError, lambda: cm['C'])
		# should raise no exceptions (tag "C" still unused):
		substituteRecursiveTags(cm)
		# add reference to "broken" tag:
		cm['D'] = 'test=<C>'
		# should raise an exception (BOOM by replacement of tag "D" recursive):
		self.assertRaises(ZeroDivisionError, lambda: substituteRecursiveTags(cm))
		#
		# should raise no exceptions:
		self.assertEqual(self.__action.replaceTag('test=<A>', cm), "test=0")
		# **Important**: recursive replacement of dynamic data from calling map should be prohibited,
		# otherwise may be vulnerable on foreign user-input:
		self.assertEqual(self.__action.replaceTag('test=<A>----<A>', cm), "test=0--<A><A>--0")
		# should raise an exception (BOOM by replacement of tag "C"):
		self.assertRaises(ZeroDivisionError, lambda: self.__action.replaceTag('test=<C>', cm))
		# should raise no exceptions (replaces tag "D" only):
		self.assertEqual(self.__action.replaceTag('<D>', cm), "test=<C>")

def testReplaceTag(self):
		aInfo = {
			'HOST': "192.0.2.0",
			'ABC': "123",
			'xyz': "890",
		}
		self.assertEqual(
			self.__action.replaceTag("Text text", aInfo),
			"Text\ntext")
		self.assertEqual(
			self.__action.replaceTag("Text <HOST> text", aInfo),
			"Text 192.0.2.0 text")
		self.assertEqual(
			self.__action.replaceTag("Text <xyz> text <ABC> ABC", aInfo),
			"Text 890 text 123 ABC")
		self.assertEqual(
			self.__action.replaceTag("<matches>",
				{'matches': "some >char< should \\< be[ escap}ed&\n"}),
			"some \\>char\\< should \\\\\\< be\\[ escap\\}ed\\&\\n")
		self.assertEqual(
			self.__action.replaceTag("<ipmatches>",
				{'ipmatches': "some >char< should \\< be[ escap}ed&\n"}),
			"some \\>char\\< should \\\\\\< be\\[ escap\\}ed\\&\\n")
		self.assertEqual(
			self.__action.replaceTag("<ipjailmatches>",
				{'ipjailmatches': "some >char< should \\< be[ escap}ed&\r\n"}),
			"some \\>char\\< should \\\\\\< be\\[ escap\\}ed\\&\\r\\n")

# Recursive
		aInfo["ABC"] = "<xyz>"
		self.assertEqual(
			self.__action.replaceTag("Text <xyz> text <ABC> ABC", aInfo),
			"Text 890 text 890 ABC")

# Callable
		self.assertEqual(
			self.__action.replaceTag("09 <matches> 11",
				CallingMap(matches=lambda self: str(10))),
			"09 10 11")

def testReplaceNoTag(self):
		# As tag not present, therefore callable should not be called
		# Will raise ValueError if it is
		self.assertEqual(
			self.__action.replaceTag("abc",
				CallingMap(matches=lambda self: int("a"))), "abc")

def testReplaceTagSelfRecursion(self):
		setattr(self.__action, 'a', "<a")
		setattr(self.__action, 'b', "c>")
		setattr(self.__action, 'b?family=inet6', "b>")
		setattr(self.__action, 'ac', "<a>")
		setattr(self.__action, 'ab', "<ac>")
		setattr(self.__action, 'x?family=inet6', "")
		# produce self-referencing properties except:
		self.assertRaisesRegex(ValueError, r"properties contain self referencing definitions",
			lambda: self.__action.replaceTag("<a>", 
				self.__action._properties, conditional="family=inet4")
		)
		# remore self-referencing in props:
		delattr(self.__action, 'ac')
		# produce self-referencing query except:
		self.assertRaisesRegex(ValueError, r"possible self referencing definitions in query",
			lambda: self.__action.replaceTag("<x"*30+">"*30,
				self.__action._properties, conditional="family=inet6")
		)

def testReplaceTagConditionalCached(self):
		setattr(self.__action, 'abc', "123")
		setattr(self.__action, 'abc?family=inet4', "345")
		setattr(self.__action, 'abc?family=inet6', "567")
		setattr(self.__action, 'xyz', "890-<abc>")
		setattr(self.__action, 'banaction', "Text <xyz> text <abc>")
		# test replacement in sub tags and direct, conditional, cached:
		cache = self.__action._substCache
		for i in range(2):
			self.assertEqual(
				self.__action.replaceTag("<banaction> '<abc>'", self.__action._properties, 
					conditional="", cache=cache),
				"Text 890-123 text 123 '123'")
			self.assertEqual(
				self.__action.replaceTag("<banaction> '<abc>'", self.__action._properties, 
					conditional="family=inet4", cache=cache),
				"Text 890-345 text 345 '345'")
			self.assertEqual(
				self.__action.replaceTag("<banaction> '<abc>'", self.__action._properties, 
					conditional="family=inet6", cache=cache),
				"Text 890-567 text 567 '567'")
		self.assertTrue(len(cache) >= 3)
		# set one parameter - internal properties and cache should be reseted:
		setattr(self.__action, 'xyz', "000-<abc>")
		self.assertEqual(len(cache), 0)
		# test againg, should have 000 instead of 890:
		for i in range(2):
			self.assertEqual(
				self.__action.replaceTag("<banaction> '<abc>'", self.__action._properties, 
					conditional="", cache=cache),
				"Text 000-123 text 123 '123'")
			self.assertEqual(
				self.__action.replaceTag("<banaction> '<abc>'", self.__action._properties, 
					conditional="family=inet4", cache=cache),
				"Text 000-345 text 345 '345'")
			self.assertEqual(
				self.__action.replaceTag("<banaction> '<abc>'", self.__action._properties, 
					conditional="family=inet6", cache=cache),
				"Text 000-567 text 567 '567'")
		self.assertTrue(len(cache) >= 3)

@with_tmpdir
	def testExecuteActionBan(self, tmp):
		tmp += "/fail2ban.test"
		self.__action.actionstart = "touch '%s'" % tmp
		self.__action.actionrepair = self.__action.actionstart
		self.assertEqual(self.__action.actionstart, "touch '%s'" % tmp)
		self.__action.actionstop = "rm -f '%s'" % tmp
		self.assertEqual(self.__action.actionstop, "rm -f '%s'" % tmp)
		self.__action.actionban = "echo -n"
		self.assertEqual(self.__action.actionban, 'echo -n')
		self.__action.actioncheck = "[ -e '%s' ]" % tmp
		self.assertEqual(self.__action.actioncheck, "[ -e '%s' ]" % tmp)
		self.__action.actionunban = "true"
		self.assertEqual(self.__action.actionunban, 'true')
		self.pruneLog()

self.assertNotLogged('returned')
		# no action was actually executed yet

self.__action.ban({'ip': None})
		self.assertLogged('Invariant check failed')
		self.assertLogged('returned successfully')
		self.__action.stop()
		self.assertLogged(self.__action.actionstop)

def testExecuteActionEmptyUnban(self):
		# unban will be executed for actions with banned items only:
		self.__action.actionban = ""
		self.__action.actionunban = ""
		self.__action.actionflush = "echo -n 'flush'"
		self.__action.actionstop = "echo -n 'stop'"
		self.__action.start();
		self.__action.ban({});
		self.pruneLog()
		self.__action.unban({})
		self.assertLogged('Nothing to do', wait=True)
		# same as above but with interim flush, so no unban anymore:
		self.__action.ban({});
		self.pruneLog('[phase 2]')
		self.__action.flush()
		self.__action.unban({})
		self.__action.stop()
		self.assertLogged('stop', wait=True)
		self.assertNotLogged('Nothing to do')

@with_tmpdir
	def testExecuteActionStartCtags(self, tmp):
		tmp += '/fail2ban.test'
		self.__action.HOST = "192.0.2.0"
		self.__action.actionstart = "touch '%s.<HOST>'" % tmp
		self.__action.actionstop = "rm -f '%s.<HOST>'" % tmp
		self.__action.actioncheck = "[ -e '%s.192.0.2.0' ]" % tmp
		self.__action.start()
		self.__action.consistencyCheck()

@with_tmpdir
	def testExecuteActionCheckRestoreEnvironment(self, tmp):
		tmp += '/fail2ban.test'
		self.__action.actionstart = ""
		self.__action.actionstop = "rm -f '%s'" % tmp
		self.__action.actionban = "rm '%s'" % tmp
		self.__action.actioncheck = "[ -e '%s' ]" % tmp
		self.assertRaises(RuntimeError, self.__action.ban, {'ip': None})
		self.assertLogged('Invariant check failed', 'Unable to restore environment', all=True)
		# 2nd time, try to restore with producing error in stop, but succeeded start hereafter:
		self.pruneLog('[phase 2]')
		self.__action.actionstart = "touch '%s'" % tmp
		self.__action.actionstop = "rm '%s'" % tmp
		self.__action.actionban = """printf "%%%%b\n" <ip> >> '%s'""" % tmp
		self.__action.actioncheck = "[ -e '%s' ]" % tmp
		self.__action.ban({'ip': None})
		self.assertLogged('Invariant check failed')
		self.assertNotLogged('Unable to restore environment')

@with_tmpdir
	def testExecuteActionCheckRepairEnvironment(self, tmp):
		tmp += '/fail2ban.test'
		self.__action.actionstart = ""
		self.__action.actionstop = ""
		self.__action.actionban = "rm '%s'" % tmp
		self.__action.actioncheck = "[ -e '%s' ]" % tmp
		self.__action.actionrepair = "echo 'repair ...'; touch '%s'" % tmp
		# 1st time with success repair:
		self.__action.ban({'ip': None})
		self.assertLogged("Invariant check failed. Trying", "echo 'repair ...'", all=True)
		self.pruneLog()
		# 2nd time failed (not really repaired):
		self.__action.actionrepair = "echo 'repair ...'"
		self.assertRaises(RuntimeError, self.__action.ban, {'ip': None})
		self.assertLogged(
			"Invariant check failed. Trying", 
			"echo 'repair ...'", 
			"Unable to restore environment", all=True)

def testExecuteActionChangeCtags(self):
		self.assertRaises(AttributeError, getattr, self.__action, "ROST")
		self.__action.ROST = "192.0.2.0"
		self.assertEqual(self.__action.ROST,"192.0.2.0")

def testExecuteActionUnbanAinfo(self):
		aInfo = CallingMap({
			'ABC': "123",
			'ip': '192.0.2.1',
			'F-*': lambda self: {
		 	'fid': 111,
		 	'fport': 222,
				'user': "tester"
			}
		})
		self.__action.actionban = "echo '<ABC>, failure <F-ID> of <F-USER> -<F-TEST>- from <ip>:<F-PORT>'"
		self.__action.actionunban = "echo '<ABC>, user <F-USER> unbanned'"
		self.__action.ban(aInfo)
		self.__action.unban(aInfo)
		self.assertLogged(
			" -- stdout: '123, failure 111 of tester -- from 192.0.2.1:222'",
			" -- stdout: '123, user tester unbanned'",
			all=True
		)

def testExecuteActionStartEmpty(self):
		self.__action.actionstart = ""
		self.__action.start()
		self.assertTrue(self.__action.executeCmd(""))
		self.assertLogged('Nothing to do')
		self.pruneLog()
		self.assertTrue(self.__action._processCmd(""))
		self.assertLogged('Nothing to do')
		self.pruneLog()

def testExecuteWithVars(self):
		self.assertTrue(self.__action.executeCmd(
			r'''printf %b "foreign input:\n'''
			r''' -- $f2bV_A --\n'''
			r''' -- $f2bV_B --\n'''
			r''' -- $(echo -n $f2bV_C) --''' # echo just replaces \n to test it as single line
			r'''"''', 
			varsDict={
			'f2bV_A': 'I\'m a hacker; && $(echo $f2bV_B)', 
			'f2bV_B': 'I"m very bad hacker', 
			'f2bV_C': '`Very | very\n$(bad & worst hacker)`'
		}))
		self.assertLogged(r"""foreign input:""",
			' -- I\'m a hacker; && $(echo $f2bV_B) --',
			' -- I"m very bad hacker --',
			' -- `Very | very $(bad & worst hacker)` --', all=True)

def testExecuteReplaceEscapeWithVars(self):
		self.__action.actionban = 'echo "** ban <ip>, reason: <reason> ...\\n<matches>"'
		self.__action.actionunban = 'echo "** unban <ip>"'
		self.__action.actionstop = 'echo "** stop monitoring"'
		matches = [
			'<actionunban>',
			'" Hooray! #',
			'`I\'m cool script kiddy',
			'`I`m very cool > /here-is-the-path/to/bin/.x-attempt.sh',
			'<actionstop>',
		]
		aInfo = {
			'ip': '192.0.2.1',
			'reason': 'hacking attempt ( he thought he knows how f2b internally works ;)',
			'matches': '\n'.join(matches)
		}
		self.pruneLog()
		self.__action.ban(aInfo)
		self.assertLogged(
			'** ban %s' % aInfo['ip'], aInfo['reason'], *matches, all=True)
		self.assertNotLogged(
			'** unban %s' % aInfo['ip'], '** stop monitoring', all=True)
		self.pruneLog()
		self.__action.unban(aInfo)
		self.__action.stop()
		self.assertLogged(
			'** unban %s' % aInfo['ip'], '** stop monitoring', all=True)

def testExecuteIncorrectCmd(self):
		CommandAction.executeCmd('/bin/ls >/dev/null\nbogusXXX now 2>/dev/null')
		self.assertLogged('HINT on 127: "Command not found"')

def testExecuteTimeout(self):
		stime = time.time()
		timeout = 1 if not unittest.F2B.fast else 0.01
		# Should take a 30 seconds (so timeout will occur)
		self.assertFalse(CommandAction.executeCmd('sleep 30', timeout=timeout))
		# give a test still 1 second, because system could be too busy
		self.assertTrue(time.time() >= stime + timeout and time.time() <= stime + timeout + 1)
		self.assertLogged('sleep 30', ' -- timed out after', all=True)
		self.assertLogged(' -- killed with SIGTERM', 
		 ' -- killed with SIGKILL')

def testExecuteTimeoutWithNastyChildren(self):
		# temporary file for a nasty kid shell script
		tmpFilename = tempfile.mktemp(".sh", "fail2ban_")
		# Create a nasty script which would hang there for a while
		with open(tmpFilename, 'w') as f:
			f.write("""#!/bin/bash
		trap : HUP EXIT TERM

echo "$$" > %s.pid
		echo "my pid $$ . sleeping lo-o-o-ong"
		sleep 30
		""" % tmpFilename)
		stime = 0

# timeout as long as pid-file was not created, but max 5 seconds
		def getnasty_tout():
			return (
				getnastypid() is not None
				or time.time() - stime > 5
			)

def getnastypid():
			cpid = None
			if os.path.isfile(tmpFilename + '.pid'):
				with open(tmpFilename + '.pid') as f:
					try:
						cpid = int(f.read())
					except ValueError:
						pass
			return cpid

# First test if can kill the bastard
		stime = time.time()
		self.assertFalse(CommandAction.executeCmd(
			'bash %s' % tmpFilename, timeout=getnasty_tout))
		# Wait up to 3 seconds, the child got killed
		cpid = getnastypid()
		# Verify that the process itself got killed
		self.assertTrue(Utils.wait_for(lambda: not pid_exists(cpid), 3))  # process should have been killed
		self.assertLogged('my pid ', 'Resource temporarily unavailable')
		self.assertLogged('timed out')
		self.assertLogged('killed with SIGTERM', 
		                  'killed with SIGKILL')
		os.unlink(tmpFilename + '.pid')

# A bit evolved case even though, previous test already tests killing children processes
		stime = time.time()
		self.assertFalse(CommandAction.executeCmd(
			'out=`bash %s`; echo ALRIGHT' % tmpFilename, timeout=getnasty_tout))
		# Wait up to 3 seconds, the child got killed
		cpid = getnastypid()
		# Verify that the process itself got killed
		self.assertTrue(Utils.wait_for(lambda: not pid_exists(cpid), 3))
		self.assertLogged('my pid ', 'Resource temporarily unavailable')
		self.assertLogged(' -- timed out')
		self.assertLogged(' -- killed with SIGTERM', 
		                  ' -- killed with SIGKILL')
		os.unlink(tmpFilename)
		os.unlink(tmpFilename + '.pid')

def testCaptureStdOutErr(self):
		CommandAction.executeCmd('echo "How now brown cow"')
		self.assertLogged("stdout: 'How now brown cow'\n")
		CommandAction.executeCmd(
			'echo "The rain in Spain stays mainly in the plain" 1>&2')
		self.assertLogged(
			"stderr: 'The rain in Spain stays mainly in the plain'\n")

def testCallingMap(self):
		mymap = CallingMap(callme=lambda self: str(10), error=lambda self: int('a'),
			dontcallme= "string", number=17)

# Should work fine
		self.assertEqual(
			"%(callme)s okay %(dontcallme)s %(number)i" % mymap,
			"10 okay string 17")
		# Error will now trip, demonstrating delayed call
		self.assertRaises(ValueError, lambda x: "%(error)i" % x, mymap)

def testCallingMapModify(self):
		m = CallingMap({
			'a': lambda self: 2 + 3,
			'b': lambda self: self['a'] + 6,
			'c': 'test',
		})
		# test reset (without modifications):
		m.reset()
		# do modifications:
		m['a'] = 4
		del m['c']
		# test set and delete:
		self.assertEqual(len(m), 2)
		self.assertNotIn('c', m)
		self.assertEqual((m['a'], m['b']), (4, 10))
		# reset to original and test again:
		m.reset()
		s = repr(m)
		self.assertEqual(len(m), 3)
		self.assertIn('c', m)
		self.assertEqual((m['a'], m['b'], m['c']), (5, 11, 'test'))
		# immutability of copy:
		m['d'] = 'dddd'
		m2 = m.copy()
		m2['c'] = lambda self: self['a'] + 7
		m2['a'] = 1
		del m2['b']
		del m2['d']
		self.assertTrue('b' in m)
		self.assertTrue('d' in m)
		self.assertFalse('b' in m2)
		self.assertFalse('d' in m2)
		self.assertEqual((m['a'], m['b'], m['c'], m['d']), (5, 11, 'test', 'dddd'))
		self.assertEqual((m2['a'], m2['c']), (1, 8))

def testCallingMapRep(self):
		m = CallingMap({
			'a': lambda self: 2 + 3,
			'b': lambda self: self['a'] + 6,
			'c': ''
		})
		s = repr(m); # only stored values (no calculated)
		self.assertNotIn("'a': ", s)
		self.assertNotIn("'b': ", s)
		self.assertIn("'c': ''", s)

s = m._asrepr(True) # all values (including calculated)
		self.assertIn("'a': 5", s)
		self.assertIn("'b': 11", s)
		self.assertIn("'c': ''", s)
		
		m['c'] = lambda self: self['xxx'] + 7; # unresolvable
		s = m._asrepr(True)
		self.assertIn("'a': 5", s)
		self.assertIn("'b': 11", s)
		self.assertIn("'c': ", s) # presents as callable
		self.assertNotIn("'c': ''", s) # but not empty

def testActionsIdleMode(self):
		a = Actions(DummyJail())
		a.sleeptime = 0.0001;   # don't need to wait long
		# enter idle mode right now (start idle):
		a.idle = True;
		# start:
		a.start()
		# wait for enter/leave of idle mode:
		self.assertLogged("Actions: enter idle mode", wait=10)
		# leave idle mode:
		a.idle = False
		self.assertLogged("Actions: leave idle mode", wait=10)
		# stop it:
		a.active = False
		a.join()

${this.title}

Code Editor : actiontestcase.py