OrchestrationLibrary.py

""" test.robot
*** Settings ***
Library    OrchestrationLibrary

*** Test Cases ***
Configuration
    [Tags]    configuration
    Trigger on    Some condition    Test 1
    Trigger on    Some condition     test 2
    Trigger on    Other condition    Test 2

Test 1
    No operation

Test 2
    No operation


*** Keywords ***
Some condition
    Log to console    some cond
    [Return]     ${True}

Other condition
    Log to console     other cond
    [Return]    ${False}
"""

import sys

from collections import defaultdict
from StringIO import StringIO
from tempfile import mkdtemp
from time import sleep

from robot.libraries.BuiltIn import BuiltIn


class RPA(object):
    ROBOT_LISTENER_API_VERSION = 3

    def __init__(self):
        self.tasks = defaultdict(list)

    def start_suite(self, suite, result):
        if not suite.tests:
            raise Exception('invalid invoke')

        configuration, tests = [], []
        for t in suite.tests:
            (configuration if 'configuration' in [tag.lower() for tag in t.tags] else tests).append(t)

        assert len(configuration) == 1, 'More than one conf'

        tasks = defaultdict(list)
        s = TestSuite('Running configuration')
        s.resource.imports.library('OrchestrationLibrary')
        configuration[0].keywords.create(name='Do magic', args=[tasks])
        s.tests = configuration
        output = StringIO()
        res = s.run(outputdir=mkdtemp(), stdout=output)
        #self.out(output.getvalue())

        assert len(tasks) > 0, 'Zero tasks D:'
        self.out(tasks)
        for trigger, test_names in tasks.iteritems():
            actual_tests = [t for t in tests if t.name.lower() in test_names]
            assert len(actual_tests) == len(test_names), 'not all matched D:'
            self.tasks[trigger] = actual_tests

        while True:
            self.process(self.tasks)
            sleep(2)
            self.out('DING')

    def process(self, tasks):
        for trigger, tasks in tasks.iteritems():
            self.out('DONG')
            if not trigger.run_condition():
                continue
            suite = TestSuite('whooo')
            ##### IMPORTS
            suite.tests = tasks
            suite.run(outputdir=mkdtemp())

    def _on_message(self, message):
        self.out('msg: %s' % message)

    def out(self, thing):
        sys.stdout.write('%s\n' % thing)

class OrchestrationLibrary(object):
    ROBOT_LIBRARY_SCOPE = 'TEST SUITE'

    def __init__(self):
        self.tasks = defaultdict(list)
        self.builtin = BuiltIn()

    def trigger_on(self, condition_kw, tc_name):
        trigger = Trigger(condition_kw, self.builtin)
        self.tasks[trigger].append(tc_name.lower())

    def do_magic(self, d):
        for k, v in self.tasks.iteritems():
            d[k] = v

class Trigger(object):
    def __init__(self, kw, builtin):
        self.kw = kw.lower().replace(' ', '').replace('_', '')
        self.builtin = builtin

    def __hash__(self):
        return hash(self.kw)

    def __cmp__(self, other):
        return cmp(self.kw, other.kw)

    def run_condition(self):
        try:
            return self.builtin.run_keyword(self.kw)
        except:
            return False