Internals

funkload.TestRunner – TestRunner

FunkLoad Test runner.

Similar to unittest.TestProgram but: * you can pass the python module to load * able to override funkload configuration file using command line options * cool color output * support doctest with python2.4

$Id: TestRunner.py 24758 2005-08-31 12:33:00Z bdelbosc $

class funkload.TestRunner.TestLoader

Override to add options when instanciating test case.

loadTestsFromModule(module)

Return a suite of all tests cases contained in the given module

loadTestsFromName(name, module=None)

Return a suite of all tests cases given a string specifier.

The name may resolve either to a module, a test case class, a test method within a test case class, or a callable object which returns a TestCase or TestSuite instance.

The method optionally resolves the names relative to a given module.

loadTestsFromTestCase(testCaseClass)

Return a suite of all tests cases contained in testCaseClass

class funkload.TestRunner.TestProgram(module=None, defaultTest=None, argv=None, testRunner=None, testLoader=<unittest.loader.TestLoader object>)

Override to add a python module and more options.

loadTests()

Load unit and doc tests from modules or names.

parseArgs(argv)

Parse programs args.

runTests()

Launch the tests.

funkload.TestRunner.display_testcases(suite)

Display test cases of the suite.

funkload.TestRunner.filter_testcases(suite, cpattern, negative_pattern=False)

Filter a suite with test names that match the compiled regex pattern.

funkload.TestRunner.main()

Default main.

funkload.BenchRunner – BenchRunner

FunkLoad Bench runner.

$Id: BenchRunner.py 24746 2005-08-31 09:59:27Z bdelbosc $

class funkload.BenchRunner.BenchRunner(module_name, class_name, method_name, options)

Run a unit test in bench mode.

addThreads(number_of_threads)

Adds new threads to existing list. Used to dynamically add new threads during a debug bench run.

createThreads(cycle, number_of_threads)

Creates number_of_threads threads and returns as a list.

NOTE: This method is not thread safe. Thread safety must be handled by the caller.

deleteThreads(number_of_threads)

Stops given number of threads and deletes from thread list.

NOTE: This method is not thread safe. Thread safety must be handled by the caller.

dumpThreads()

Display all different traceback of Threads for debugging.

Require threadframe module.

getMonitorsConfig()

Get monitors configuration from hosts

logging(cycle, cvus)

Log activity during duration.

logr(message)

Log to the test result file.

logr_close()

Stop logging tag.

logr_open()

Start logging tag.

removeThreads(number_of_threads)

Removes threads. Used to dynamically remove threads during a debug bench run.

run()

Run all the cycles.

return 0 on success, 1 if there were some failures and -1 on errors.

startMonitors(monitor_key)

Start monitoring on hosts list.

startThreads(cycle, number_of_threads)

Starts threads.

stopMonitors(monitor_key)

Stop monitoring and save xml result.

stopThreads()

Stops all running threads.

class funkload.BenchRunner.LoopTestRunner(test_module, test_class, test_name, options, cycle, cvus, thread_id, thread_signaller, sleep_time, debug=False, feedback=None)

Run a unit test in loop.

run()

Run a test in loop.

class funkload.BenchRunner.ThreadData(thread, thread_id, thread_signaller)

Container for thread related data.

class funkload.BenchRunner.ThreadSignaller

A simple class to signal whether a thread should continue running or stop.

funkload.BenchRunner.add_cycle_result(status)

Count number of result.

funkload.BenchRunner.get_cycle_results()

Return counters.

funkload.BenchRunner.get_shared_OptionParser()

Make an OptionParser that can be used in both normal mode and in discover mode.

funkload.BenchRunner.get_status(success, failures, errors, color=False)

Return a status and an exit code.

funkload.BenchRunner.load_unittest(test_module, test_class, test_name, options)

instantiate a unittest.

funkload.BenchRunner.main(sys_args=['-b', 'html', '-d', 'build/doctrees', 'source', 'build/html'])

Default main.

funkload.BenchRunner.reset_cycle_results()

Clear the previous results.

funkload.Distributed – Distributed

class funkload.Distributed.DistributionMgr(module_name, class_name, method_name, options, cmd_args)

Interface for use by funkload.TestRunner to distribute the bench over multiple machines.

prepare_workers(allow_errors=False)

Initialize the sandboxes in each worker node to prepare for a bench run. The additional parameter allow_errors will essentially make the distinction between ignoring unresponsive/inappropriate nodes - or raising an error and failing the entire bench.

run()

startMonitors()

Start monitoring on hosts list.

stopMonitors()

Stop monitoring and save xml result.

write_statistics(successful_results)

Write the distributed stats to a file in the output dir

class funkload.Distributed.DistributorBase(host, username, password)

base class for any XXXDistributor objects that can be used to distribute benches accross multiple machines.

class funkload.Distributed.SSHDistributor(name, host, username=None, password=None, key_filename=None, channel_timeout=None)

Provides commands to perform distirbuted actions using an ssh connection (depends on paramiko). Essentially used by DistributionMgr.

die()

kills the ssh connection

execute(*args, **kwargs)

evaluated the command specified by cmd_string in the context of cwdir if it is specified. The optional shell_interpreter parameter allows overloading the default bash.

get(*args, **kwargs)

performs a copy from remote_path to local_path. For performing the inverse operation, use the put()

isdir(*args, **kwargs)

test to see if the path pointing to remote_dir exists as a directory.

isfile(*args, **kwargs)

test to see if the path pointing to remote_path exists as a file.

put(*args, **kwargs)

performs a copy from local_path to remote_path For performing the inverse operation, use the get()

threaded_execute(*args, **kwargs)

basically the same as execute() execept that it returns a started threading.Thread object instead of the output.

funkload.Distributed.load_unittest(test_module, test_class, test_name, options)

instantiate a unittest.

funkload.Distributed.requiresconnection(fn)

decorator for SSHDistributor object that raises a runtime exception upon calling methods if the object hasn’t been connected properly.