[svn-commits] mjordan: testsuite/asterisk/trunk r3336 - /asterisk/trunk/lib/python/asterisk/

[SVN commits to the Digium repositories]

Author: mjordan
Date: Thu Jul 19 09:46:06 2012
New Revision: 3336

URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=3336
Log:
Add CELModule to the Asterisk Test Suite

This adds a pluggable module for CEL verification to the Asterisk Test Suite.
Tests that are configuration driven can use this module to verify CEL records
after a test has completed.

Review: https://reviewboard.asterisk.org/r/2050


Modified:
    asterisk/trunk/lib/python/asterisk/cdr.py
    asterisk/trunk/lib/python/asterisk/cel.py

Modified: asterisk/trunk/lib/python/asterisk/cdr.py
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/lib/python/asterisk/cdr.py?view=diff&rev=3336&r1=3335&r2=3336
==============================================================================
--- asterisk/trunk/lib/python/asterisk/cdr.py (original)
+++ asterisk/trunk/lib/python/asterisk/cdr.py Thu Jul 19 09:46:06 2012
@@ -47,16 +47,7 @@
                 dict_record = dict((k, None) for k in AsteriskCSVCDRLine.fields)
                 dict_record.update(csv_line)
 
-                self.cdr_records[file_name].append(AsteriskCSVCDRLine(
-                    accountcode=dict_record['accountcode'], source=dict_record['source'],
-                    destination=dict_record['destination'], dcontext=dict_record['dcontext'],
-                    callerid=dict_record['callerid'], channel=dict_record['channel'],
-                    dchannel=dict_record['dchannel'], lastapp=dict_record['lastapp'],
-                    lastarg=dict_record['lastarg'], start=dict_record['start'],
-                    answer=dict_record['answer'], end=dict_record['end'],
-                    duration=dict_record['duration'], billsec=dict_record['billsec'],
-                    disposition=dict_record['disposition'], amaflags=dict_record['amaflags'],
-                    uniqueid=dict_record['uniqueid'], userfield=dict_record['userfield']))
+                self.cdr_records[file_name].append(AsteriskCSVCDRLine(**dict_record))
 
         # Hook ourselves onto the test object
         test_object.register_stop_observer(self._check_cdr_records)

Modified: asterisk/trunk/lib/python/asterisk/cel.py
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/lib/python/asterisk/cel.py?view=diff&rev=3336&r1=3335&r2=3336
==============================================================================
--- asterisk/trunk/lib/python/asterisk/cel.py (original)
+++ asterisk/trunk/lib/python/asterisk/cel.py Thu Jul 19 09:46:06 2012
@@ -18,7 +18,70 @@
 import logging
 import time
 
-logger = logging.getLogger(__name__)
+LOGGER = logging.getLogger(__name__)
+
+class CELModule(object):
+    ''' A module that checks a test for expected CEL results '''
+
+
+    def __init__(self, module_config, test_object):
+        ''' Constructor
+
+        Parameters:
+        module_config The yaml loaded configuration for the CEL Module
+        test_object A concrete implementation of TestClass
+        '''
+        self.test_object = test_object
+
+        # Build our expected CEL records
+        self.cel_records = {}
+        for record in module_config:
+            file_name = record['file']
+            if file_name not in self.cel_records:
+                self.cel_records[file_name] = []
+            for csv_line in record['lines']:
+                # Set the record to the default fields, then update with what
+                # was passed in to us
+                dict_record = dict((k, None) for k in AsteriskCSVCELLine.fields)
+                dict_record.update(csv_line)
+
+                self.cel_records[file_name].append(AsteriskCSVCELLine(**dict_record))
+
+        # Hook ourselves onto the test object
+        test_object.register_stop_observer(self._check_cel_records)
+
+    def _check_cel_records(self, callback_param):
+        ''' A deferred callback method that is called by the TestCase
+        derived object when all Asterisk instances have stopped
+
+        Parameters:
+        callback_param
+        '''
+        LOGGER.debug("Checking CEL records...")
+        self.match_cels()
+        return callback_param
+
+
+    def match_cels(self):
+        ''' Called when all instances of Asterisk have exited.  Derived
+        classes can override this to provide their own behavior for CEL
+        matching.
+        '''
+        expectations_met = True
+        for key in self.cel_records:
+            cel_expect = AsteriskCSVCEL(records=self.cel_records[key])
+            cel_file = AsteriskCSVCEL(fn="%s/%s/cel-custom/%s.csv" %
+                (self.test_object.ast[0].base,
+                 self.test_object.ast[0].directories['astlogdir'],
+                 key))
+            if cel_expect.match(cel_file):
+                LOGGER.debug("%s.csv - CEL results met expectations" % key)
+            else:
+                LOGGER.error("%s.csv - CEL results did not meet expectations.  Test Failed." % key)
+                expectations_met = False
+
+        self.test_object.set_passed(expectations_met)
+
 
 class AsteriskCSVCELLine(astcsv.AsteriskCSVLine):
     "A single Asterisk Call Event Log record"