A Cucumber/Lettuce-like test runner with in-Python "Gherkin" syntax and complete parallelism.

01-rough.py

# encoding: utf-8

"""
Sample output (at no concurrency):

Feature: Test running.
  Scenario: Basic tests.
    Given: We have something to test.
      When: Tests pass.
        Then: No error should be present.
         And: Nothing is logged.
      When: Tests fail.
        Then: An error should be present.
         And: The error is logged.
  Scenario: Advanced tests.
    Given: We have something to test.
      When: Tests pass.
        Then: No error should be present.
         And: Nothing is logged.
      When: Tests fail.
        Then: An error should be present.
         And: The error is logged.

Crappily interleaved output (at 10 concurrency):

Feature: Test running.
  Scenario: Basic tests.
    Given: We have something to test.
  Scenario: Advanced tests.
      When: Tests pass.
    Given: We have something to test.
        Then: No error should be present.
      When: Tests pass.
      When: Tests fail.
         And: Nothing is logged.
        Then: No error should be present.
        Then: An error should be present.
         And: Nothing is logged.
         And: The error is logged.
      When: Tests fail.
        Then: An error should be present.
         And: The error is logged.
"""

from __future__ import unicode_literals, print_function

import copy

from marrow.util.bunch import Bunch

from functools import partial
from concurrent import futures


__all__ = ["Feature"]

log = __import__('logging').getLogger(__name__)



class Step(object):
    def __init__(self, fn, description, kind=None, executor=None):
        super(Step, self).__init__()
        
        self.registry = Bunch(
                scenario = [],
                given = [],
                when = [],
                then = []
            )
        
        self.fn = fn
        self.description = description
        self.executor = executor
        
        self.kind = kind if kind else self.__class__.__name__
    
    def __call__(self, environ=None):
        log.debug("%s%s: %s%s.", '  ' * self.indentation, self.kind, self.description[0].upper(), self.description[1:])
        
        if environ is None:
            environ = Bunch()
        
        if self.fn:
            result = self.fn(environ)
            if result:
                environ = result
        
        # Execute steps.
        jobs = []
        executor = self.executor()
        for stage in ('scenario', 'given', 'when', 'then'):
            for next in self.registry[stage]:
                job = executor.submit(next, environ)
                # job.add_done_callback(...) # error handling
                jobs.append(job)
        
        futures.wait(jobs)


class Feature(Step):
    indentation = 0
    
    def __init__(self, name, description=None, executor=None):
        super(Feature, self).__init__(
                None,
                name,
                executor=executor if executor else partial(futures.ThreadPoolExecutor, max_workers=1)
            )
    
    def scenario(self, description):
        kind = '     And' if self.__class__ is Scenario else None
        
        def decorator(fn):
            obj = Scenario(fn, description, kind, self.executor)
            self.registry.scenario.append(obj)
            return obj
        
        return decorator


class Then(Step):
    indentation = 4
    
    def then(self, description):
        kind = ' And' if self.__class__ is Then else None
        
        def decorator(fn):
            obj = Then(fn, description, kind, self.executor)
            self.registry.then.append(obj)
            return obj
        
        return decorator


class When(Then):
    indentation = 3
    
    def when(self, description):
        kind = ' And' if self.__class__ is When else None
        
        def decorator(fn):
            obj = When(fn, description, kind, self.executor)
            self.registry.when.append(obj)
            return obj

        return decorator


class Given(When):
    indentation = 2
    
    def given(self, description):
        kind = '  And' if self.__class__ is Given else None
        
        def decorator(fn):
            obj = Given(fn, description, kind, self.executor)
            self.registry.given.append(obj)
            return obj
        
        return decorator


class Scenario(Given):
    indentation = 1



def main():
    import logging
    logging.basicConfig(level=logging.DEBUG, format="%(message)s")
    
    tests = Feature(
            "test running",
            """In order to write tests that work.
            We need to know that tests can be run.""",
            executor = partial(futures.ThreadPoolExecutor, max_workers=10)
        )
    
    @tests.scenario("basic tests")
    def basic(environ):
        # Configure the environment here.
        # log.info("Configuring the environment.")
        
        environ.foo = 27
        
        return environ
    
    @basic.given("we have something to test")
    def given_tests(environ):
        # log.info("has_testable: %r", environ)
        environ.foo += 1
        return environ
    
    @given_tests.when("tests pass")
    def tests_pass(environ):
        # log.info("tests_pass: %r", environ)
        environ.foo += 1
        return environ
    
    @tests_pass.then("no error should be present")
    def no_errors(environ):
        # log.info("no_errors: %r", environ)
        pass
    
    @no_errors.then("nothing is logged")
    def no_errors(environ):
        # log.info("no_errors: %r", environ)
        pass
    
    @given_tests.when("tests fail")
    def tests_fail(environ):
        # log.info("tests_fail: %r", environ)
        environ.foo += 1
        return environ
    
    @tests_fail.then("an error should be present")
    def has_errors(environ):
        # log.info("has_errors: %r", environ)
        pass
    
    @has_errors.then("the error is logged")
    def has_errors(environ):
        pass
    
    ###
    
    @tests.scenario("advanced tests")
    def advanced(environ):
        # Configure the environment here.
        # log.info("Configuring the environment.")
        
        environ.foo = 27
        
        return environ
    
    @advanced.given("we have something to test")
    def given_tests(environ):
        # log.info("has_testable: %r", environ)
        environ.foo += 1
        return environ
    
    @given_tests.when("tests pass")
    def tests_pass(environ):
        # log.info("tests_pass: %r", environ)
        environ.foo += 1
        return environ
    
    @tests_pass.then("no error should be present")
    def no_errors(environ):
        # log.info("no_errors: %r", environ)
        pass
    
    @no_errors.then("nothing is logged")
    def no_errors(environ):
        # log.info("no_errors: %r", environ)
        pass
    
    @given_tests.when("tests fail")
    def tests_fail(environ):
        # log.info("tests_fail: %r", environ)
        environ.foo += 1
        return environ
    
    @tests_fail.then("an error should be present")
    def has_errors(environ):
        # log.info("has_errors: %r", environ)
        pass
    
    @has_errors.then("the error is logged")
    def has_errors(environ):
        pass
    
    ###
    
    tests()


if __name__ == '__main__':
    main()

Here's the overall idea, back-linked for referential integrity: https://gist.github.com/amcgregor/1338661