The Whimsy Library

Main Modules

Modules which complete most of the hard work and the majority of execution time is spent in.

whimsy.main module

The main source for whimsy. Discovers and loads sources using the TestLoader objects and runs tests using the Runner object passing the runner ResultLogger instances which will stream output data to the terminal and into various result files.

There are three commands which this program handles:

  • run - By default will search for and run all tests in the current
    and children directories reporting the results through the terminal, saving them to a pickle file, and saving them to a junit file.
  • rerun - Load all tests and then rerun the tests which failed in the previous
    run.
  • list - List tests with various querying options.
whimsy.main.doclient()[source]

Handle the client command.

whimsy.main.dolist()[source]

Handle the list command.

whimsy.main.dorerun()[source]

Handle the rerun command.

whimsy.main.dorun()[source]

Handle the run command.

whimsy.main.load_tests()[source]

Create a TestLoader and load tests for the directory given by the config.

whimsy.main.main()[source]

whimsy.loader module

Contains the Loader which is responsible for discovering and loading tests.

Loading typically follows the following stages.

  1. Recurse down a given directory looking for tests which match a given regex.

The default regex used will match any python file (ending in .py) that has a name starting or ending in test(s). If there are any additional components of the name they must be connected with ‘-‘ or ‘_’. Lastly, file names that begin with ‘.’ will be ignored.

The following names would match:

  • tests.py
  • test.py
  • test-this.py
  • tests-that.py
  • these-test.py

These would not match:

  • .test.py - ‘hidden’ files are ignored.
  • test - Must end in ‘.py’
  • test-.py - Needs a character after the hypen.
  • testthis.py - Needs a hypen or underscore to separate ‘test’ and ‘this’
  1. With all files discovered execute each file gathering its test items we care about collecting. (TestCase, TestSuite and Fixture objects.)

In order to collect these objects the TestLoader must find a way to capture objects as they are instantiated. To do this, before calling execfile() on the file about to be loaded, TestLoader will monkey patch the __new__() method of test item classes. The monkey patched __new__ will insert the new instance of the test method as it is created into an OrderedSet saved in the TestLoader.

In addition to adding a callback to the __new__() function of these test objects the loader adds a __no_collect__() method. This method can be used by test writers to prevent a test item from being collected. This functionallity is exposed to test writers with the no_collect() function in this module. Users might create an item that fits almost all their needs butbut wish to make copy of it and modify a single attribute without the original version being collected.

>>> from __future__ import print_function
>>> from testlib import *
>>> from copy import copy
>>> hello_func = lambda : print('Hello')
>>> test = TestFunction(hello_func)
>>> # Create a copy, don't collect the original.
>>> test_copy = copy(no_collect(test))
>>> test_copy.name = 'hello-test'

As a final note, TestCase instances which are not put into a TestSuite by the test writer will be placed into a TestSuite named after the module.

See also

load_file()

exception whimsy.loader.DuplicateTestItemError[source]

Bases: exceptions.Exception

class whimsy.loader.TestLoader(filepath_filter=<function default_filepath_filter>, quiet=False)[source]

Bases: object

Base class for discovering tests.

To simply discover and load all tests using the default filter create an instance and load_root.

>>> import os
>>> tl = TestLoader()
>>> tl.load_root(os.getcwd())

Note

If tests are not manually placed in a TestSuite, they will automatically be placed into one for the module.

discover_files(root)[source]

Recurse down from the given root directory returning a list of directories which contain a list of files matching self.filepath_filter.

drop_caches()[source]

Drop our internal tag cache.

fixtures
get_uid(uid)[source]

Return the test item with the given uid.

load_dir(directory)[source]
load_file(path, collection=None)[source]

Load the given path for tests collecting all created instances of TestSuite, TestCase, and Fixture objects.

TestSuites are placed into self._suites (a private SuiteList). TestCases which are not explicitly stored within a TestSuite are placed into a TestSuite which is generated from the filepath name.

There are couple of things that might be unexpected to the user in the loading namespace.

First, the current working directory will change to the directory path of the file being loaded.

Second, in order to collect tests we wrap __new__ calls of collected objects before calling execfile. If a user wishes to prevent an instantiated object from being collected (possibly to create a copy with a modified attribute) they should use no_collect() to do so. (Internally a __no_collect__ method is added to objects we plan to collect. This method will then remove the object from those collected from the file.)

Note

Automatically drop_caches

Warning

There isn’t a way to prevent reloading of test modules that are imported by other test modules. It’s up to users to never import a test module from a test module, otherwise those tests enumerated during the importer module’s load.

load_root(root)[source]

Load files from the given root directory which match self.filepath_filter.

static load_uid(uid)[source]

Attempt to load the given UID.

Parameters:uid – The uid to attempt to load a test item for.
Returns:A whimsy.suite.TestSuite or whimsy.test.TestCase if the given UID can be found, else None.
suites
suites_with_tag(tag)[source]
tag_index(tag)[source]

Return a list of test items with the given tag.

tags
tests
whimsy.loader.default_filepath_filter(filepath)[source]

The default filter applied to filepaths to marks as test sources.

whimsy.loader.no_collect(test_item)[source]

Prevent the collection of the test item by the TestLoader.

Returns:the item given so this can be easily used in comprehensions.
whimsy.loader.path_as_modulename(filepath)[source]

Return the given filepath as a module name.

whimsy.loader.path_as_testsuite(filepath, *args, **kwargs)[source]

Return the given filepath as a testsuite.

The testsuite will be named after the containing directory of the file.

whimsy.runner module

Test Items

Modules which implement the abstraction of different testing concepts.

whimsy.suite module

class whimsy.suite.SuiteList(suites=[])[source]

Bases: object

Container class for test suites which provides some utility functions.

append(item)[source]
extend(items)[source]
iter_fixtures()[source]

Return an iterable of all fixtures of all TestSuites and their contained TestCases in this collection.

Note

Will return duplicates if fixtures are duplicated across test cases.

class whimsy.suite.TestList(items=[], fail_fast=False)[source]

Bases: object

Container class for whimsy.test.TestCase objects which provides some utility functions for iteration as well as the fail_fast variable. A TestList can be heirarchical, in which case iteration yields tests in in-order traversal.

append(item)[source]

Add the item to the end of this TestList.

extend(items)[source]

Add the items to the end of this TestList.

iter_keep_containers()[source]

Iterate through tests yielding a tuple of the containing TestList and the TestCase.

Returns:iterator of (testlist, testcase) objects.
iter_tests()[source]

Iterate in-order over all contained tests.

class whimsy.suite.TestSuite(name, tests=(), tags=None, fixtures=None, fail_fast=True)[source]

Bases: object

An object containing a collection of tests, represents a completely self-contained testing unit. That is, if it should be able to be ran without relying on any other suites running.

Note

In order for TestSuite objects to be enumerated by the test system this class’ __new__ method must be called. The loader class will monkey patch this method in order to enumerate suites.

append(item)[source]

Adds the given TestCase or TestList to this TestSuite

extend(items)[source]

Adds the given TestCases or TestLists to this TestSuite

iter_testlists()[source]

Iterate through tests yielding a tuple of the containing TestList and the TestCase.

Returns:iterator of (testlist, testcase) objects.
name
path
testcases

Return a tuple containing all the test cases this TestSuite holds.

uid

whimsy.test module

class whimsy.test.TestCase(name, tags=None, fixtures=None, path=None)[source]

Bases: object

Abstract Base Class for test cases. All that’s missing is a __call__() implementation.

Represents a single major item of assertion. (Yes that’s vague.) Some examples are: Asserting that gem5 actually started, asserting that output from gem5 standard out matched a gold standard diff file. (See whimsy.gem5.verifier for some concrete examples.)

Warning

Although this class is abstract its __new__ method must be called by subclasses in order for them to be discovered by the whimsy.loader.TestLoader.

name
path
uid
exception whimsy.test.TestFailException[source]

Bases: whimsy.test.TestingException

Signals that a test has failed.

class whimsy.test.TestFunction(test, name=None, *args, **kwargs)[source]

Bases: whimsy.test.TestCase

A concrete implementation of the abc TestCase. Uses a function as a test.

exception whimsy.test.TestSkipException[source]

Bases: whimsy.test.TestingException

Signals that a test has been skipped.

exception whimsy.test.TestingException[source]

Bases: exceptions.Exception

Common ancestor for manual Testing Exceptions.

whimsy.test.fail(message)[source]

Cause the current test to fail with the given message.

whimsy.test.skip(message)[source]

Cause the current test to skip with the given message.

whimsy.test.testfunction(function=None, name=None, tag=None, tags=None, fixtures=None)[source]

A decorator used to wrap a function as a TestFunction.

whimsy.fixture module

Exposes the Fixture class.

class whimsy.fixture.Fixture(name, build_once=False, lazy_init=True)[source]

Bases: object

Base Class for a test Fixture.

Fixtures are items which possibly require setup and/or tearing down after a TestCase or a TestSuite has run.

Fixtures are the prefered method of carrying incremental results or variables between TestCases in TestSuites. (Rather than using globals.) The test system assumes that TestSuites are self contained so paralellization is possible in the future. Using global variables will likely cause things to break unexpectedly.

Note

no_collect function is provided by the loader to remove a fixture that should not be collected.

Note

In order for Fixtures to be enumerated by the test system this class’ __new__ method must be called. The loader class will monkey patch (modify at runtime) this method in order to enmerate tests.

built
require(other_fixture)[source]

Require that the other_fixture be built before us. This is particularly useful for build targets.

See also

SConsTarget

setup()[source]

Call setup of fixtures we require.

teardown()[source]

Empty method, meant to be overriden if fixture requires teardown.

Support Modules

Modules which act has helpers or support items throughout the framwework.

whimsy.result module

Module which contains streaming result loggers. The main goal of these loggers is to support large amounts of results and not create large standing pools of strings.

class whimsy.result.ConsoleLogger[source]

Bases: whimsy.result.ResultLogger

A logger implementing the streaming ResultLogger interface. This logger is used to stream testing result output to a user terminal.

begin(item)[source]
begin_testing()[source]
color = <whimsy.terminal.ColorStrings object>
colormap = {<whimsy._util._EnumVal object>: '\x1b[36m', <whimsy._util._EnumVal object>: '\x1b[32m', <whimsy._util._EnumVal object>: '\x1b[31m', <whimsy._util._EnumVal object>: '\x1b[36m', <whimsy._util._EnumVal object>: '\x1b[31m'}
end_testing()[source]
insert_results(internal_results)[source]

Insert the given results from an whimsy.result.InternalLogger into the console logger. (Display them.)

reset = '\x1b(B\x1b[m'
sep_fmtkey = 'separator'
sep_fmtstr = '{separator}'
set_outcome(item, outcome, **kwargs)[source]
class whimsy.result.InternalLogger(filestream)[source]

Bases: whimsy.result.ResultLogger

An internal logger which writes streaming pickle items on completion of TestSuite items.

This logger also offers some metadata methods to and can load back out previous results.

See also

load() suites()

begin_testing()[source]
end_testing()[source]
insert_results(internal_results)[source]
static load(filestream)[source]

Load results out of a dumped file replacing our own results.

set_outcome(item, **kwargs)[source]
suites

Return an iterator over all the test suite results loaded or collected.

testcases
exception whimsy.result.InvalidResultException[source]

Bases: exceptions.Exception

class whimsy.result.JUnitFormatter(internal_results, translate_names=True)[source]

Bases: object

Formats TestResults into the JUnit XML format.

dump(dumpfile)[source]
dump_testcase(fstream, testcase)[source]
dump_testsuite(fstream, suite, idx)[source]
error_tag = '<error message="{message}"></error>\n'
fail_tag = '<failure message="{message}"></error>\n'
generic_closing = '</{tag}>\n'
passing_results = set([<whimsy._util._EnumVal object>, <whimsy._util._EnumVal object>])
skipped_tag = '<skipped/>'
system_err_opening = '<system-err>'
system_out_opening = '<system-out>'
testcase_opening = '<testcase name="{name}" classname="{classname}" status="{status}" time="{time}">\n'
testsuite_opening = '<testsuite name="{name}" tests="{numtests}" errors="{errors}" failures="{failures}" skipped="{skipped}" id={suitenum} time="{time}">\n'
testsuites_opening = '<testsuites errors="{errors}" failures="{failures}" tests="{tests}" time="{time}">\n'
xml_header = '<?xml version="1.0" encoding="UTF-8"?>\n'
class whimsy.result.JUnitLogger(junit_fstream, internal_fstream)[source]

Bases: whimsy.result.InternalLogger

Logger which uses the internal logger to collect streaming results to the internal_fstream, and on completion of testing writes the results out to a junit_fstream.

Parameters:
  • junit_fstream – File stream to write junit formatted results to.
  • internal_fstream – File stream to write internal formatted results to.

See also

InternalLogger

end_testing()[source]

Signal the end of writing to the file stream. We will write all our results to our junit_fstream.

class whimsy.result.ResultLogger[source]

Bases: object

Interface which allows writing of streaming results to a file stream.

bad_item = 'Result formatter can only handle test cases and test suites'
begin(item)[source]

Signal the beginning of the given item. :param item: The test item which is about to begin running

begin_testing()[source]

Signal the beginning of writing to the file stream. Indicates that results are about to be logged.

This is garunteed to be called before any results are added.

delegate_instance(function, instance, *args, **kwargs)[source]

Helper function to delegate a method for the given function based on the type of the given instance.

Effectively is shorthand for:

>>> if isinstance(instance, TestSuite):
>>>     self._call_this_testsuite(*args, **kwargs)
>>> elif isinstance(instance, TestCase):
>>>     self._call_this_testsuite(*args, **kwargs)
>>> elif __debug__:
>>>     raise AssertionError
end(item)[source]

Signal the end of the current item.

Parameters:item – The test item which is finished running
end_testing()[source]

Indicates that results are done being collected.

This is guaranteed to be called after all results are added.

not_supplied = <object object>

A sentinel value indicating that the kwarg wasn’t supplied.

set_outcome(item, outcome, **kwargs)[source]

Set the outcome of the given item.

Parameters:
  • item – The test item which we are setting the outcome of
  • outcome – The outcome the test item will be set to

TestCase Only kwargs:

Parameters:
  • reason – Reason for the test case outcome.
  • fstdout_name – Name of the file stdout is available at.
  • fstderr_name – Name of the file stdout is available at.
  • ff_skipped – Indicates that the test was skipped due to a fail_fast condition.
class whimsy.result.TestCaseResult(fstdout_name=None, fstderr_name=None, reason=None, ff_skipped=None, **kwargs)[source]

Bases: whimsy.result.TestResult

class whimsy.result.TestResult(item, outcome, runtime=0)[source]

Bases: object

class whimsy.result.TestSuiteResult(test_case_results, **kwargs)[source]

Bases: whimsy.result.TestResult

whimsy.result.test_results_output_path(test_case)[source]

Return the path which results for a specific test case should be stored.

whimsy.logger module

Provides a common logging system. With ability to add additional logging levels.

class whimsy.logger.ConsoleLogFormatter[source]

Bases: object

Formats output to be sent to an interactive terminal. Colors may be added based on logging level.

color = <whimsy.terminal.ColorStrings object>
format(record)[source]
level_colormap = {1000: '\x1b[37m\x1b[1m', 50: '\x1b[31m', 30: '\x1b[33m'}
reset = '\x1b(B\x1b[m'
whimsy.logger.add_logging_level(name, val)[source]
whimsy.logger.set_logging_verbosity(verbosity)[source]

Set the logging level based on the number of verbosity flags given.

whimsy.config module

Global configuration module which exposes two types of configuration variables:

  1. config
  2. constants (Also attached to the config variable as an attribute)

The main motivation for this module is to have a centralized location for defaults and configuration by command line and files for the test framework.

A secondary goal is to reduce programming errors by providing common constant strings and values as python attributes to simplify detection of typos. A simple typo in a string can take a lot of debugging to uncover the issue, attribute errors are easier to notice and most autocompletion systems detect them.

The config variable is initialzed by callling initialize_config(). Before this point only constants will be availaible. This is to ensure that library function writers never accidentally get stale config attributes.

Program arguments/flag arguments are available from the config as attributes. If a attribute was not set by the command line or the optional config file, then it will fallback to the _defaults value, if still the value is not found an AttributeError will be raised.

func define_defaults:
 Provided by the config if the attribute is not found in the config or commandline. For instance, if we are using the list command fixtures might not be able to count on the build_dir being provided since we aren’t going to build anything.
var constants:Values not directly exposed by the config, but are attached to the object for centralized access. I.E. you can reach them with config.constants.attribute. These should be used for setting common string names used across the test framework. _defaults.build_dir = None Once this module has been imported constants should not be modified and their base attributes are frozen.
class whimsy.config.ArgParser(parser)[source]

Bases: object

class whimsy.config.Argument(*flags, **kwargs)[source]

Bases: object

Class represents a cli argument/flag for a argparse parser.

Attr name:The long name of this object that will be stored in the arg output by the final parser.
add_to(parser)[source]

Add this argument to the given parser.

copy()[source]

Copy this argument so you might modify any of its kwargs.

class whimsy.config.ClientParser(subparser)[source]

Bases: whimsy.config.ArgParser

Parser for the ‘client’ command.

class whimsy.config.CommandParser[source]

Bases: whimsy.config.ArgParser

Main parser which parses command strings and uses those to direct to a subparser.

class whimsy.config.ListParser(subparser)[source]

Bases: whimsy.config.ArgParser

Parser for the ‘list’ command.

class whimsy.config.RerunParser(subparser)[source]

Bases: whimsy.config.ArgParser

class whimsy.config.RunParser(subparser)[source]

Bases: whimsy.config.ArgParser

Parser for the ‘run’ command.

exception whimsy.config.UninitializedConfigException[source]

Bases: exceptions.Exception

Signals that the config was not initialized before trying to access an attribute.

exception whimsy.config.UninitialzedAttributeException[source]

Bases: exceptions.Exception

Signals that an attribute in the config file was not initialized.

whimsy.config.config
whimsy.config.constants = <whimsy._util.FrozenAttrDict object>

This config object is the singleton config object available throughtout the framework.

whimsy.config.define_common_args(config)[source]

Common args are arguments which are likely to be simular between different subcommands, so they are available to all by placing their definitions here.

whimsy.config.define_constants(constants)[source]

‘constants’ are values not directly exposed by the config, but are attached to the object for centralized access. These should be used for setting common string names used across the test framework. A simple typo in a string can take a lot of debugging to uncover the issue, attribute errors are easier to notice and most autocompletion systems detect them.

whimsy.config.define_defaults(defaults)[source]

Defaults are provided by the config if the attribute is not found in the config or commandline. For instance, if we are using the list command fixtures might not be able to count on the build_dir being provided since we aren’t going to build anything.

whimsy.config.define_post_processors(config)[source]

post_processors are used to do final configuration of variables. This is useful if there is a dynamically set default, or some function that needs to be applied after parsing in order to set a configration value.

Post processors must accept a single argument that will either be a tuple containing the already set config value or None if the config value has not been set to anything. They must return the modified value in the same format.

whimsy.config.initialize_config()[source]

Parse the commandline arguments and setup the config varibles.

whimsy.helper module

Helper classes for writing tests with this test library.

  • log_call()
    A wrappper around Popen which behaves like subprocess.check_call() but will pipe output to the log at a low verbosity level.
  • cacheresult()
    A function decorator which will cache results for a function given the same arguments. (A poor man’s python3 lru_cache.)
  • OrderedSet
    A set which maintains object insertion order.
  • absdirpath()
    dirname(abspath())
  • joinpath()
    os.path.join()
  • mkdir_p()
    Same thing as mkdir -p
whimsy.helper.log_call(command, *popenargs, **kwargs)[source]

Calls the given process and automatically logs the command and output.

If stdout or stderr are provided output will also be piped into those streams as well.

Params stdout:Iterable of items to write to as we read from the subprocess.
Params stderr:Iterable of items to write to as we read from the subprocess.
exception whimsy.helper.CalledProcessError(returncode, cmd, output=None)[source]

Bases: exceptions.Exception

This exception is raised when a process run by check_call() or check_output() returns a non-zero exit status.

Attributes:
cmd, returncode, output
whimsy.helper.mkdir_p(path)[source]

Same thing as mkdir -p

https://stackoverflow.com/a/600612

whimsy.helper.cacheresult(function, typed=False)[source]
Parameters:typed – If typed is True, arguments of different types will be cached separately. I.e. f(3.0) and f(3) will be treated as distinct calls with distinct results.

Note

From cpython 3.7

class whimsy.helper.OrderedSet(iterable=None)[source]

Bases: _abcoll.MutableSet

Maintain ordering of insertion in items to the set with quick iteration.

http://code.activestate.com/recipes/576694/

add(key)[source]
discard(key)[source]
pop(last=True)[source]
update(keys)[source]
whimsy.helper.absdirpath(path)[source]

Return the directory component of the absolute path of the given path.

whimsy.helper.joinpath(a, *p)

Join two or more pathname components, inserting ‘/’ as needed. If any component is an absolute path, all previous path components will be discarded. An empty last part will result in a path that ends with a separator.

class whimsy.helper.OrderedDict(*args, **kwds)[source]

Bases: dict

Dictionary that remembers insertion order

clear() → None. Remove all items from od.[source]
copy() → a shallow copy of od[source]
classmethod fromkeys(S[, v]) → New ordered dictionary with keys from S.[source]

If not specified, the value defaults to None.

items() → list of (key, value) pairs in od[source]
iteritems()[source]

od.iteritems -> an iterator over the (key, value) pairs in od

iterkeys() → an iterator over the keys in od[source]
itervalues()[source]

od.itervalues -> an iterator over the values in od

keys() → list of keys in od[source]
pop(k[, d]) → v, remove specified key and return the corresponding[source]

value. If key is not found, d is returned if given, otherwise KeyError is raised.

popitem() → (k, v), return and remove a (key, value) pair.[source]

Pairs are returned in LIFO order if last is true or FIFO order if false.

setdefault(k[, d]) → od.get(k,d), also set od[k]=d if k not in od[source]
update([E, ]**F) → None. Update D from mapping/iterable E and F.

If E present and has a .keys() method, does: for k in E: D[k] = E[k] If E present and lacks .keys() method, does: for (k, v) in E: D[k] = v In either case, this is followed by: for k, v in F.items(): D[k] = v

values() → list of values in od[source]
viewitems() → a set-like object providing a view on od's items[source]
viewkeys() → a set-like object providing a view on od's keys[source]
viewvalues() → an object providing a view on od's values[source]

whimsy.query module

File which implements querying and display logic for metadata about loaded items.

whimsy.query.list_fixtures(loader)[source]
whimsy.query.list_suites(loader)[source]
whimsy.query.list_tags(loader)[source]
whimsy.query.list_tests(loader)[source]
whimsy.query.list_tests_with_tags(loader, tags)[source]

whimsy.runner.parallel module

Implements a base WorkerPool which will execute a serial or parallel implementation of imap_unordered depending on the number of threads requested on subclass __init__. A single instance of imap_unordered may be active at one time. Additional executions will require that the previous imap have finished.

class whimsy.runner.parallel.ComplexMulticorePool(threads=None)[source]

Bases: whimsy.runner.parallel.WorkerPool

Class implements the server container for a multi-client remote and local multiprocessing pool. Exposes an imap_unordered() method which will either execute serially if only initialized with a single thread, or will exececute in parallel if more given.

Additionally, if more than one thread is given will allow remote clients to connect and join the pool.

pool
class whimsy.runner.parallel.MulticoreWorkerPool(threads=None)[source]

Bases: whimsy.runner.parallel.WorkerPool

A worker takes jobs of its queue used to initalize it and sends them to the process which it wraps to execute.

pool
exception whimsy.runner.parallel.SubprocessException[source]

Bases: exceptions.Exception

Exception represents an exception that occured in a child python process.

class whimsy.runner.parallel.WorkClient(hostname, port, passkey, as_client=True)[source]

Bases: multiprocessing.process.Process

static imap_task(wq, rq)[source]
run()[source]

Connects to the server and gathers its work and result queues. Also the majority of the config of the server process. (Allowing flags like fail_fast to be passed to us.)

Note

Exectued on the client process.

class whimsy.runner.parallel.WorkQueueClient(hostname, port, passkey)[source]

Bases: multiprocessing.managers.SyncManager

class whimsy.runner.parallel.WorkQueueServer(hostname, port, passkey)[source]

Bases: multiprocessing.managers.SyncManager

Implements a server which clients can connect to get work and result queues as well as the server’s config setup.

class whimsy.runner.parallel.WorkServer(hostname, port, passkey)[source]

Bases: object

Implements a server object which creates work queues and result queues which clients can connect to to assist in work. Additionally in order to implement an imap_unordered function that does not block, a separate client is spawned with the server.

imap_unordered(function, args)[source]

Provides functional equivalence of:

>>> return (function(arg) for arg in args)

Note

This will not block since we also spawn a work_client to assist this process.

shutdown()[source]

Shutdown the server, the client will close asynchronosly since it will block in its own process waiting for the socket to close.

start()[source]

Start the server and the helping work_client.

class whimsy.runner.parallel.WorkerPool(threads)[source]

Bases: object

Will execute a serial or parallel implementation of imap_unordered depending on the number of threads requested on subclass __init__. A single instance of imap_unordered may be active at one time. Additional executions will require that the previous imap have finished.

imap_unordered(map_function, args)[source]
Parameters:
  • function – A module level function to supply jobs to. (Note: Must be exposed globaly by a module.
  • args – An iterable containing arguments provided to members of the pool which the function will take.

Effectively this function performs:

>>> return (map_function(arg) for arg in args)
pool
whimsy.runner.parallel.log_if_client(callback, *args, **kwargs)[source]

Execute the given callback if the client command was given at program startup.

whimsy.runner.parallel.subprocess_exception_wrapper(args)[source]

Wraps a python child process with a function that will enable tracebacks to be printed from child python processes.

whimsy.tee module

Contains two implementations of the classic unix tee command to tee output from this python process (and all of its subprocesses) into a file and keep directing output to stdout and stderr.

whimsy.tee.python_tee(filepath, append=False, stdout=True, stderr=True)[source]

Python implementation of tee.

Spawns a multiprocessing python process, a full flegged python process to avoid the GIL problem since the subprocess will likely spend a lot of time spinning while reading single char bytes.

whimsy.tee.system_tee(filepath, append=False, stdout=True, stderr=True)[source]

An implementation of tee using the system available program tee as a subprocess.

whimsy.tee.tee(*args, **kwds)[source]

A context manager for the tee command. Tries to default to the tee program for the system for performance reasons, but if it is unavailable, will use a pure python implementaion.

An example of usage:

>>> with tee('stdout', stdout=True, stderr=False):
>>>     print ('This is going to both the file and stdout')

whimsy.terminal module

class whimsy.terminal.ColorStrings(cap_string)[source]

Bases: object

whimsy.terminal.cap_string(s, *args)[source]
whimsy.terminal.get_termcap(use_colors=None)[source]
whimsy.terminal.insert_separator(inside, char='=', min_barrier=3, color=None)[source]

Place the given string inside of the separator. If it does not fit inside, expand the separator to fit it with at least min_barrier.

See also

separator()

whimsy.terminal.null_cap_string(s, *args)[source]
whimsy.terminal.separator(char='=', color=None)[source]

Return a separator of the given character that is the length of the full width of the terminal screen.

whimsy.terminal.terminal_size()[source]

Return the (width, heigth) of the terminal screen.