tests Package

test_cooperative Module

class cooperative.tests.test_cooperative.Doer(own_reactor, own_cooperator)

Bases: object

count = 0

run(*args, **kwargs)

Cooperatively iterator over two iterators consecutively and the result of the final one is returned.

Returns:

class cooperative.tests.test_cooperative.TestAccumulate(methodName='runTest')

Bases: twisted.trial._asynctest.TestCase

test_accumulate(*args, **kwargs)

Ensure that within an inline callback function, a accumulate wrapped generator yields the result of the output of the generator.

Returns:

test_failure(*args, **kwargs)

Ensure that within an inline callback function, a accumulate based function yields the result if it’s cooperative generator.

Since and_the_winner_is is designed to always log and error, Ensure one IndexError is logged.

Returns:

test_multi_deux_batched(*args, **kwargs)

Ensure multiple inline callback functions will run cooperatively.

Ensure the result of gatherResults can be chained together in order.

Ensure cooperatively run generators will complete no matter the length.

Ensure the longest one will continue to iterate after the others run out of iterations.

Ensure those called with batch_accumulate will iterate over the generator in batches the size of max_size.

Returns:

test_multi_deux_chain(*args, **kwargs)

Ensure multiple inline callback functions will run cooperatively.

Ensure the result of gatherResults can be chained together in order.

Ensure cooperatively run generators will complete no matter the length.

Ensure the longest one will continue to iterate after the others run out of iterations.

Returns:

test_multi_winner(*args, **kwargs)

Ensure multiple inline callback functions will run cooperatively.

Returns:

test_multi_winner_chain(*args, **kwargs)

Ensure multiple inline callback functions will run cooperatively.

Ensure the result of gatherResults can be chained together in order.

Returns:

test_trice_winner(*args, **kwargs)

Ensure multiple inline callback functions will run cooperatively.

Returns:

class cooperative.tests.test_cooperative.TestHandler(methodName='runTest')

Bases: twisted.trial._asynctest.TestCase

test_accumulation_handler()

Ensure the return value of accumulation_handler is the contents of a Bucket instance with it’s contents drained.

Returns:

class cooperative.tests.test_cooperative.TestOwnCooperator(methodName='runTest')

Bases: twisted.trial._asynctest.TestCase

setUp()

Create a reactor and Cooperator that can be controlled.

Instantiate a Doer with the reactor and cooperator.

Create a Looping Call and set it’s clock to the reactor.

Returns:

tearDown()

test_control_coop()

Ensure control of own cooperator.

Returns:

cooperative.tests.test_cooperative.i_get_tenth_11(value)

Yield the tenth and eleventh item of value.

Parameters:value – Returns:

cooperative.tests.test_cooperative.run_some_with_error(*args, **kwargs)

Cooperatively iterator over two iterators consecutively, but

the second one will always raise an IndexError, which is caught, logged and a message is returned.

Returns:

cooperative.tests.test_cooperative.run_some_without_error(*args, **kwargs)

Cooperatively iterator over two iterators consecutively and the result of the final one is returned.

Parameters:value – Any sequence. Returns: