Add indication testing framework

A small test framework for testing LMI Indications.  Oriented on
data-driven testing approach and capable of "provoking" events
and tracking relevant indications.

Classes should be extended on provider-level to include provider-
-specific actions and knowledge.

1 files changed, 411 insertions, 0 deletions

diff --git a/src/python/lmi/test/ind.py b/src/python/lmi/test/ind.py
new file mode 100644
index 0000000..e722336
--- /dev/null
+++ b/src/python/lmi/test/ind.py
@@ -0,0 +1,411 @@
+# Copyright (C) 2012-2014 Red Hat, Inc.  All rights reserved.
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
+#
+# Authors: Alois Mahdal <amahdal@redhat.com>
+#
+"""
+LMI indication testing framework
+"""
+
+import os
+import string
+import time
+
+import lmi.shell
+import lmi.test.util
+import lmi.test.lmibase
+
+
+class IndicationStreamTestCase(lmi.test.lmibase.LmiTestCase):
+    """
+    Base class for indication test cases with helper methods.
+
+    This class is intended to be used as base for new test
+    cases that are testing indications.  It provides utility
+    methods to simplify definifion of test cases and reduce
+    code repetition.
+
+    Also provides mechanism to set defaults for driver options.
+    Set cls.default_driver_options for hardcoded defaults per
+    test case class.  These can be overrided on runtime by
+    environment variables as defined in import_driver_options,
+    and furthermore in each test case, by updating
+    driver_options property.
+
+    driver_default_argset can be set in similar manner (save
+    for ENV), although it's not recommended to make extensive
+    use of these; most of the time it's better to be explicit
+    or use driver options instead.
+    """
+
+    # what to use as default options
+    # override this as self.driver_options in your
+    # subclassed testcase or as ENV variables
+    default_driver_options = {
+        'listener_host': 'localhost',
+        'listener_port': 12000,
+        'delay_action': 0.1,
+        'delay_chillout': 5,
+        'gen_prefix': 'lmi_test_ind_',
+        'gen_strength': 10,
+    }
+
+    # what to import, where in options and what handler to use
+    # for conversion to proper format
+    import_driver_options = [
+        ("LMI_IND_DELAY_CHILLOUT", 'delay_chillout', float),
+        ("LMI_IND_DELAY_ACTION", 'delay_action', float),
+        ("LMI_IND_GEN_PREFIX", 'gen_prefix', str),
+        ("LMI_IND_GEN_STRENGTH", 'gen_strength', int),
+        ("LMI_IND_LISTENER_HOST", 'listener_host', str),
+        ("LMI_IND_LISTENER_PORT", 'listener_port', int),
+    ]
+
+    def setUp(self):
+        self.set_driver_options()
+        self.set_driver_default_argset()
+        self.driver_options['conn'] = self.conn
+        self.driver_default_argset['name'] = self._testMethodName
+
+    def assertExpectedStream(self, probe, case):
+        """
+        Assert correct sequence of class names was delivered.
+
+        Uses probe interface to pull list of class names (or
+        more  spefifically, their aliases as registered in
+        IndicationTestProbe.classname_aliases) delivered and
+        compares that to value under 'expected_stream' key of
+        case argument.
+
+        Both of these values are strings as used by
+        lmi.test.util.PackedSequence.
+        """
+        es = case['expected_stream']
+        oracle = str(lmi.test.util.PackedSequence(es))
+        result = probe.get_cns()
+        msg = ("Expected classes not delivered:\n"
+               "..subscriptions: %r\n"
+               "..triggered actions: %r\n"
+               "..expected stream: %r\n"
+               "..got: %r\n"
+               % (case['subscriptions'], probe.get_ans(),
+                  oracle, result))
+        assert result == oracle, msg
+
+    def execute(self, case):
+        """
+        Just instantiate new driver and run it with case data.
+
+        Instantiate self.driver_class with self.driver_options
+        and self.driver_default_argset, and call its run method,
+        returning the result untouched.
+
+        This constitutes preferred scenario of how the drivers
+        should be used, i.e. one throw-away instance per each
+        test case.
+        """
+        drv = self.driver_class(self.driver_options,
+                                self.driver_default_argset)
+        return drv.run(case)
+
+    def mknumcase(self, numcase, action):
+        """
+        Construct case based on numbers of events/suscriptions.
+
+        This is to support scenarios when you want to create
+        test cases automatically, and you are only interested
+        in number (possibly zero) of events, or relevant
+        subscriptions and handlers.
+
+        For example, you want to check that with M handlers,
+        N subscriptions and P actions, Q indications are
+        generated.  I.e. you only want to define the numbers,
+        not whole cases.
+        """
+        counts, expect = numcase
+        return {
+            'handlers': '%dpr' % counts[0],
+            'subscriptions': '%d%s' % (counts[1], action),
+            'actions': '%d%s' % (counts[2], action),
+            'expected_stream': "%d%s" % (expect, action),
+        }
+
+    def set_driver_default_argset(self):
+        """
+        Set driver default argset from class defaults
+        """
+        self.driver_default_argset = {}
+        if hasattr(self, "default_driver_default_argset"):
+            self.driver_default_argset.update(
+                self.default_driver_default_argset.copy())
+
+    def set_driver_options(self):
+        """
+        Set driver options from class defaults and ENV
+        """
+        self.driver_options = {}
+        # 1. class default
+        if hasattr(self, "default_driver_options"):
+            self.driver_options.update(self.default_driver_options.copy())
+        # 2. import from ENV
+        if hasattr(self, "import_driver_options"):
+            for envkey, optkey, convert in self.import_driver_options:
+                if envkey in os.environ:
+                    self.driver_options[optkey] = convert(os.environ[envkey])
+
+
+class IndicationTestStub(object):
+    """
+    Test stub - imitation of a system called by OpenLMI
+    """
+
+    @staticmethod
+    def probe_report(ind, probe, *args, **kwargs):
+        probe.report_indication(ind, args, kwargs)
+
+
+class IndicationTestProbe(object):
+    """
+    Hold details about actions trigerred and indications seen.
+
+    Provides methods to be called from listener handler, to
+    allow for tracking what indications were seen and when, and
+    by action trigger, to track when event was triggered.
+
+    These reports can then be queried by get_* methods, to help
+    make assertions and report discrepancies.
+    """
+
+    class _ActionReport(object):
+
+        def __init__(self, action, rv):
+            self.time = time.time()
+            self.action = action
+            self.rv = rv
+
+        def __cmp__(self, other):
+            return cmp(self.time, other.time)
+
+    class _IndicationReport(object):
+
+        def __init__(self, ind, args, kwargs):
+            self.time = time.time()
+            self.ind = ind
+            self.args = args
+            self.kwargs = kwargs
+
+        def __cmp__(self, other):
+            a = self.ind.exported_objects()[0]['IndicationTime'].value
+            b = other.ind.exported_objects()[0]['IndicationTime'].value
+            return cmp(a, b)
+
+    def __init__(self):
+        self.indication_reports = []
+        self.action_reports = []
+        self.classname_aliases = {}
+
+    def get_cns(self):
+        """
+        Get sequence of class names aliases delivered to probe
+
+        Return string describing sequence of class names found
+        in probe reports.  The string follows notation used by
+        lmi.test.util.PackedSequence.
+
+        The string does not use full names, but their "aliases"
+        as defined in classname_aliases dictionary.  This is to
+        enforce use of simple, data-driven and and easy to
+        understand case definitions.
+        """
+        ps = lmi.test.util.PackedSequence()
+        for rep in self.indication_reports:
+            cn = rep.ind.exported_objects()[0].classname
+            try:
+                ps.append(self.classname_aliases[cn])
+            except KeyError:
+                raise ValueError("unknown CIM class name: " + cn)
+        return str(ps)
+
+    def get_ans(self):
+        """
+        Get sequence of action names delivered to probe.
+
+        Return string describing sequence of action names found
+        in probe reports.  The string follows notation used by
+        lmi.test.util.PackedSequence.
+        """
+        ps = lmi.test.util.PackedSequence()
+        [ps.append(r.action) for r in self.action_reports]
+        return str(ps)
+
+    def report_indication(self, ind, args, kwargs):
+        r = self._IndicationReport(ind, args, kwargs)
+        self.indication_reports.append(r)
+
+    def report_action(self, action, rv):
+        r = self._ActionReport(action, rv)
+        self.action_reports.append(r)
+
+
+class IndicationTestDriver(lmi.test.util.BaseTestDriver):
+    """
+    Test Driver for OpenLMI Indications.
+
+    This driver wraps common "correct" scenario to register
+    indications and listener and triggering actions of interest.
+    It then allows running the scenario with arbitrary
+    variations of what sequence of events is triggered and what
+    is expected.
+
+    To make use of the driver, subclass it and in __init__, add
+    your queries, handlers and actions to respective
+    self.known_* dictionaries.  Keys under which you add them
+    become "aliases", which you then use when describing the
+    test case.
+
+    Behavior inherited by lmi.test.util.BaseTestDriver is that
+    test case is described by a dict argument passed to run().
+    Eventually _do_run() is called, where your implementation
+    refers to above as self.argset.
+
+    This implementation of _do_run expects following values
+    passed as options when instantiating:
+
+    *   'listener_host' and 'listener_port', hostname/ip/port
+         where listener should be started; this is also where
+         subscriptions are directed.
+
+    *   'delay_action'/'delay_chillout', delay between each
+        triggered action, and after last action, respectively.
+
+    *   'gen_prefix'/'gen_strength', whenever driver needs to
+        "make up" own name e.g. for username, it will use this
+        prefix, and add this number of random ASCII characters.
+
+    and following in the argset dictionary when calling run():
+
+    *   'name' - a short descriptive name to be used to
+        construct handler and subsctiption names; aim is
+        basically to help in case of debugging
+
+    rest of them have special format to be used with
+    `PackedSequence`:
+
+    *   'subscriptions' describes list of subscriptions to
+        register; these are currently only WQL queries; the
+        rest of subscription parameters is inferred from
+        options or left to default
+
+    *   'handlers' describes list of handlers to pair with
+        subscriptions.
+
+        This is optional. By default each subscription will be
+        paired with simple reporting handler that only passes
+        indication to the probe, so it should  suffice in most
+        cases.  However, if you specify this  argument, you
+        need to ensure mapping to subscription works as you
+        want, i.e. you normally want to specify same amount of
+        handlers as subscriptions.
+
+    *   'actions' describes sequence of events to be triggered,
+        such as creation of account or installation of a
+        software package or whatever you have provided in
+        known_actions.
+
+    Return value of run() is a IndicationTestProbe instance that
+    contains records of all important events and contains helper
+    methods, to query these, so that assertions can be made to
+    complete the test.
+
+    """
+
+    def __init__(self, *args, **kwargs):
+        super(IndicationTestDriver, self).__init__(*args, **kwargs)
+
+        # create essential objects
+        self.probe = IndicationTestProbe()
+        self.listener = lmi.shell.LMIIndicationListener(
+            self.options['listener_host'],
+            self.options['listener_port'])
+        self.known_queries = {}
+        self.known_handlers = {}
+        self.known_actions = {}
+
+        # register cleanup handlers
+        uai = self.options['conn'].unsubscribe_all_indications
+        self.cleanup_handlers.append(uai)
+        self.cleanup_handlers.append(self.listener.stop)
+
+        # add basic handler
+        self.known_handlers['pr'] = IndicationTestStub.probe_report
+
+        # add basic actions
+        self.known_actions['SL1'] = lambda: time.sleep(1)
+        self.known_actions['SL2'] = lambda: time.sleep(2)
+        self.known_actions['SL5'] = lambda: time.sleep(5)
+
+    def _prepare_handlers(self):
+        """Create listener, add handlers and start listener"""
+        hnames = []
+        prefix = self.argset['name'] + "_"
+        chars = filter(lambda c: c != 'X',
+                       string.ascii_letters + string.digits)
+        for h in lmi.test.util.PackedSequence(self.argset['handlers']):
+            name = lmi.test.util.random_string(prefix=prefix, chars=chars)
+            fn = self.known_handlers[h]
+            hname = self.listener.add_handler(name, fn, self.probe, name)
+            hnames.append(hname)
+        self.hnames = hnames
+
+    def _prepare_subscriptions(self):
+        """Register all subscriptions"""
+
+        dest = "http://%(listener_host)s:%(listener_port)s" % self.options
+
+        def getname():
+            """Pop from handler names or make up new"""
+            try:
+                return self.hnames.pop()
+            except IndexError:
+                prefix = "_orphan_%s_" % sub
+                return lmi.test.util.random_string(prefix=prefix)
+
+        for sub in lmi.test.util.PackedSequence(self.argset['subscriptions']):
+            rval = self.options['conn'].subscribe_indication(
+                Query=self.known_queries[sub],
+                Name=getname(),
+                Destination=dest
+            )
+            assert rval.rval, "subscribing indication failed"
+
+    def _trigger_actions(self):
+
+        actions = lmi.test.util.PackedSequence(self.argset['actions'])
+
+        for action in actions:
+            if action not in self.known_actions:
+                raise ValueError("unknown action in sequence: " + action)
+            rv = self.known_actions[action]()
+            self.probe.report_action(action, rv)
+            time.sleep(self.options['delay_action'])
+
+    def _do_run(self):
+        self._prepare_handlers()
+        self._prepare_subscriptions()
+        self.listener.start()
+        self._trigger_actions()
+        time.sleep(self.options['delay_chillout'])
+        return self.probe