[asterisk-commits] twilson: testsuite/asterisk/trunk r196 - in /asterisk/trunk: lib/python/aster...

SVN commits to the Asterisk project asterisk-commits at lists.digium.com
Fri Apr 2 18:52:23 CDT 2010


Author: twilson
Date: Fri Apr  2 18:52:20 2010
New Revision: 196

URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=196
Log:
Add CDR handling to the library

This is a library that adds the ability to test the CDRs that are generated
during a test.  It can read a list of CDRs in from cdr-csv file, and test that
the fields match some given input.

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

Added:
    asterisk/trunk/lib/python/asterisk/cdr.py   (with props)
    asterisk/trunk/lib/python/asterisk/self_test/
    asterisk/trunk/lib/python/asterisk/self_test/Master.csv   (with props)
    asterisk/trunk/lib/python/asterisk/self_test/Master2.csv   (with props)
    asterisk/trunk/tests/cdr/
Modified:
    asterisk/trunk/lib/python/asterisk/__init__.py
    asterisk/trunk/tests/iax-call-basic/run-test

Modified: asterisk/trunk/lib/python/asterisk/__init__.py
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/lib/python/asterisk/__init__.py?view=diff&rev=196&r1=195&r2=196
==============================================================================
--- asterisk/trunk/lib/python/asterisk/__init__.py (original)
+++ asterisk/trunk/lib/python/asterisk/__init__.py Fri Apr  2 18:52:20 2010
@@ -1,1 +1,1 @@
-__all__ = [ "asterisk", "config", "version" ]
+__all__ = [ "asterisk", "config", "version", "cdr" ]

Added: asterisk/trunk/lib/python/asterisk/cdr.py
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/lib/python/asterisk/cdr.py?view=auto&rev=196
==============================================================================
--- asterisk/trunk/lib/python/asterisk/cdr.py (added)
+++ asterisk/trunk/lib/python/asterisk/cdr.py Fri Apr  2 18:52:20 2010
@@ -1,0 +1,137 @@
+#!/usr/bin/env python
+"""Asterisk call detail record testing
+
+This module implements an Asterisk CDR parser.
+
+Copyright (C) 2010, Digium, Inc.
+Terry Wilson<twilson at digium.com>
+
+This program is free software, distributed under the terms of
+the GNU General Public License Version 2.
+"""
+
+import unittest
+import sys
+
+class AsteriskCSVCDRLine:
+    "A single Asterisk call detail record"
+
+    def __init__(self, accountcode=None, source=None, destination=None,
+            dcontext=None, callerid=None, channel=None, dchannel=None,
+            lastapp=None, lastarg=None, start=None, answer=None, end=None,
+            duration=None, billsec=None, disposition=None, amaflags=None,
+            uniqueid=None, userfield=None):
+        """Construct an Asterisk CSV CDR.
+
+        The arguments list definition must be in the same order that the
+        arguments appear in the CSV file. They can, of course, be passed to
+        __init__ in any order. AsteriskCSVCDR will pass the arguments via a
+        **dict.
+        """
+
+        # make all arguments passed available as instance variables
+        tmp = locals()
+        del tmp['self']
+        self.__dict__.update(tmp)
+
+    def match(self, other):
+        """Matches if the subset of fields that exist in both records match.
+
+        It is important to make sure that if you want to test against an empty
+        field, that it is passed as an empty string and not left out of the
+        initialization or passed as None.
+        """
+
+        for k,v in self.iteritems():
+            if None not in (v, other.get(k)) and str(v).lower() != str(other.get(k)).lower():
+                print "CDR MATCH FAILED, Expected: %s:%s Got: %s:%s" % (k, v,
+                        k, other.get(k))
+                return False
+        return True
+
+    def get(self, k):
+        return self.__dict__.get(k)
+
+    def iteritems(self):
+        return self.__dict__.iteritems()
+
+    __fields = __init__.func_code.co_varnames[1:]
+
+    @classmethod
+    def get_fields(self):
+        return self.__fields
+
+    @classmethod
+    def get_field(self, i):
+        return self.__fields[i]
+
+
+class AsteriskCSVCDR:
+    """A representation of an Asterisk CSV CDR file"""
+
+    def __init__(self, fn=None, records=None):
+        """Initialize CDR records from an Asterisk cdr-csv file"""
+
+        if records:
+            self.__records = records
+            return
+
+        self.__records = []
+        try:
+            f = open(fn, "r")
+            lines = f.read().splitlines()
+            f.close()
+        except IOError:
+            print "Failed to open CDR file '%s'" %s (fn)
+            return
+        except:
+            print "Unexpected error: %s" % (sys.exc_info()[0])
+            return
+
+        for line in lines:
+            r = dict([(AsteriskCSVCDRLine.get_field(i), x.strip('"')) for i,x in enumerate(line.split(','))])
+            record = AsteriskCSVCDRLine(**r)
+            self.__records.append(record)
+
+    def __len__(self):
+        return len(self.__records)
+
+    def __getitem__(self, key):
+        return self.__records.__getitem__(key)
+
+    def __iter__(self):
+        return self.__records.__iter__()
+
+    def match(self, other):
+        """Compares the length of self and other AsteriskCSVCDRs and then compares
+        each record"""
+
+        if len(self) != len(other):
+            print "CDR MATCH FAILED, different number of records"
+            return False
+        for i,x in enumerate(self):
+            if not x.match(other[i]):
+                return False
+        return True
+
+
+class AsteriskCSVCDRTests(unittest.TestCase):
+    def test_cdr(self):
+        c = AsteriskCSVCDR("self_test/Master.csv")
+        self.assertEqual(len(c), 2)
+        self.assertTrue(AsteriskCSVCDRLine(duration=7,lastapp="hangup").match(c[0]))
+        self.assertTrue(c[0].match(AsteriskCSVCDRLine(duration=7,lastapp="hangup")))
+
+        self.assertFalse(c[1].match(c[0]))
+        self.assertFalse(c[0].match(c[1]))
+        self.assertEqual(c[0].billsec, "7")
+
+        self.assertTrue(c.match(c))
+        c2 = AsteriskCSVCDR("self_test/Master2.csv")
+        self.assertFalse(c.match(c2))
+
+
+if __name__ == '__main__':
+    unittest.main()
+
+# vim:sw=4:ts=4:expandtab:textwidth=79

Propchange: asterisk/trunk/lib/python/asterisk/cdr.py
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: asterisk/trunk/lib/python/asterisk/cdr.py
------------------------------------------------------------------------------
    svn:keywords = Author Date Id Revision

Propchange: asterisk/trunk/lib/python/asterisk/cdr.py
------------------------------------------------------------------------------
    svn:mime-type = text/plain

Added: asterisk/trunk/lib/python/asterisk/self_test/Master.csv
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/lib/python/asterisk/self_test/Master.csv?view=auto&rev=196
==============================================================================
--- asterisk/trunk/lib/python/asterisk/self_test/Master.csv (added)
+++ asterisk/trunk/lib/python/asterisk/self_test/Master.csv Fri Apr  2 18:52:20 2010
@@ -1,0 +1,2 @@
+"","","1","default","","Console/dsp","","Hangup","","2010-04-02 00:26:32","2010-04-02 00:26:32","2010-04-02 00:26:39",7,7,"ANSWERED","DOCUMENTATION","1270167992.0",""
+"","","1","default","","Console/dsp","","Hangup","","2010-04-02 00:34:02","2010-04-02 00:34:02","2010-04-02 00:34:09",7,7,"ANSWERED","DOCUMENTATION","1270168442.0",""

Propchange: asterisk/trunk/lib/python/asterisk/self_test/Master.csv
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: asterisk/trunk/lib/python/asterisk/self_test/Master.csv
------------------------------------------------------------------------------
    svn:keywords = Author Date Id Revision

Propchange: asterisk/trunk/lib/python/asterisk/self_test/Master.csv
------------------------------------------------------------------------------
    svn:mime-type = text/plain

Added: asterisk/trunk/lib/python/asterisk/self_test/Master2.csv
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/lib/python/asterisk/self_test/Master2.csv?view=auto&rev=196
==============================================================================
--- asterisk/trunk/lib/python/asterisk/self_test/Master2.csv (added)
+++ asterisk/trunk/lib/python/asterisk/self_test/Master2.csv Fri Apr  2 18:52:20 2010
@@ -1,0 +1,1 @@
+"","","1","default","","Console/dsp","","Hangup","","2010-04-02 00:26:32","2010-04-02 00:26:32","2010-04-02 00:26:39",7,7,"ANSWERED","DOCUMENTATION","1270167992.0",""

Propchange: asterisk/trunk/lib/python/asterisk/self_test/Master2.csv
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: asterisk/trunk/lib/python/asterisk/self_test/Master2.csv
------------------------------------------------------------------------------
    svn:keywords = Author Date Id Revision

Propchange: asterisk/trunk/lib/python/asterisk/self_test/Master2.csv
------------------------------------------------------------------------------
    svn:mime-type = text/plain

Modified: asterisk/trunk/tests/iax-call-basic/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/trunk/tests/iax-call-basic/run-test?view=diff&rev=196&r1=195&r2=196
==============================================================================
--- asterisk/trunk/tests/iax-call-basic/run-test (original)
+++ asterisk/trunk/tests/iax-call-basic/run-test Fri Apr  2 18:52:20 2010
@@ -18,12 +18,14 @@
 sys.path.append("lib/python")
 from asterisk.asterisk import Asterisk
 from asterisk.version import AsteriskVersion
+from asterisk.cdr import AsteriskCSVCDR, AsteriskCSVCDRLine
 
 
 class IAXCallTest:
     def __init__(self, argv):
         self.chan1_connected = False
         self.chan2_connected = False
+        self.cdr_match = False
         self.f = fastagi.FastAGIFactory(self.fastagi_func)
         reactor.listenTCP(4573, self.f, 50, '127.0.0.1')
         reactor.callWhenRunning(self.run)
@@ -78,6 +80,15 @@
         print "Stopping Asterisk instances ..."
         self.asterisk.stop()
         self.asterisk2.stop()
+        cdr = AsteriskCSVCDR(fn="%s/var/log/asterisk/cdr-csv/Master.csv" %
+                self.asterisk2.base)
+        cdr_test = AsteriskCSVCDR(
+                records=[AsteriskCSVCDRLine(disposition="ANSWERED", dchannel="",
+                dcontext="iaxtest", amaflags="DOCUMENTATION", accountcode="",
+                callerid="", userfield="", source="", destination="1000",
+                lastapp="AGI", lastarg="GET")])
+        if cdr_test.match(cdr):
+            self.cdr_match = True
 
     def run(self):
         self.start_asterisk()
@@ -102,7 +113,7 @@
     test = IAXCallTest(argv)
     reactor.run()
     test.stop_asterisk()
-    if test.chan1_connected and test.chan2_connected:
+    if test.chan1_connected and test.chan2_connected and test.cdr_match:
         return 0
     return 1
 




More information about the asterisk-commits mailing list