[asterisk-commits] mjordan: branch mjordan/twisted_process r3100 - in /asterisk/team/mjordan/twi...
SVN commits to the Asterisk project
asterisk-commits at lists.digium.com
Mon Mar 19 09:06:10 CDT 2012
Author: mjordan
Date: Mon Mar 19 09:06:06 2012
New Revision: 3100
URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=3100
Log:
Upload fixes for sip_incomplete, udptl_v6 for process migration
Removed:
asterisk/team/mjordan/twisted_process/tests/udptl_v6/configs/ast1/manager.conf
asterisk/team/mjordan/twisted_process/tests/udptl_v6/configs/ast2/manager.conf
Modified:
asterisk/team/mjordan/twisted_process/tests/apps/incomplete/sip_incomplete/run-test
asterisk/team/mjordan/twisted_process/tests/udptl_v6/run-test
Modified: asterisk/team/mjordan/twisted_process/tests/apps/incomplete/sip_incomplete/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/mjordan/twisted_process/tests/apps/incomplete/sip_incomplete/run-test?view=diff&rev=3100&r1=3099&r2=3100
==============================================================================
--- asterisk/team/mjordan/twisted_process/tests/apps/incomplete/sip_incomplete/run-test (original)
+++ asterisk/team/mjordan/twisted_process/tests/apps/incomplete/sip_incomplete/run-test Mon Mar 19 09:06:06 2012
@@ -40,12 +40,19 @@
def ami_connect(self, ami):
TestCase.ami_connect(self, ami)
+ def __check_sipp_results(self, result):
+ logger.debug("Checking results for %s" % str(result))
+ if not result.passed:
+ self.passed = False
+
self.ast[ami.id].cli_exec("sip set debug on")
ami.registerEvent('UserEvent', self.user_event)
logger.debug("Starting SIP scenarios")
- self.receiverTest.run()
- self.senderTest.run()
+ dr = self.receiverTest.run()
+ ds = self.senderTest.run()
+ dr.addCallback(__check_sipp_results)
+ ds.addCallback(__check_sipp_results)
def user_event(self, ami, event):
if event['userevent'] != 'TestResult':
@@ -60,17 +67,7 @@
logger.warn("status: %s" % (event['status'],))
self.passed = False
- self.stop_reactor()
-
- def stop_asterisk(self):
- TestCase.stop_asterisk(self)
-
- """ Wait for SIPp to exit """
- logger.debug("Waiting for SIPp to end")
- result = self.senderTest.waitAndEvaluate()
- result = result and self.receiverTest.waitAndEvaluate()
- if not result:
- self.passed = False
+ reactor.callLater(5, self.stop_reactor)
def run(self):
TestCase.run(self)
Modified: asterisk/team/mjordan/twisted_process/tests/udptl_v6/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/mjordan/twisted_process/tests/udptl_v6/run-test?view=diff&rev=3100&r1=3099&r2=3100
==============================================================================
--- asterisk/team/mjordan/twisted_process/tests/udptl_v6/run-test (original)
+++ asterisk/team/mjordan/twisted_process/tests/udptl_v6/run-test Mon Mar 19 09:06:06 2012
@@ -6,108 +6,65 @@
This program is free software, distributed under the terms of
the GNU General Public License Version 2.
'''
+
import sys
-import os
-import signal
-import subprocess
-from twisted.application import service, internet
-from twisted.internet import reactor, defer
+from twisted.internet import reactor
from starpy import manager
+import logging
sys.path.append("lib/python")
from asterisk.asterisk import Asterisk
+from asterisk.TestCase import TestCase
-workingdir = "udptl_v6"
-testdir = "tests/%s" % workingdir
+logger = logging.getLogger(__name__)
-class UdptlTest:
+class UdptlTest(TestCase):
def __init__(self):
- self.last_step = ""
- self.passed = False
+ TestCase.__init__(self)
- reactor.callWhenRunning(self.run)
+ self.reactor_timeout = 60
+ self.create_asterisk(2)
- self.ast1 = Asterisk(base=workingdir)
- self.ast1.install_configs("%s/configs/ast1" % (testdir))
-
- self.ast2 = Asterisk(base=workingdir)
- self.ast2.install_configs("%s/configs/ast2" % (testdir))
-
- def start_asterisk(self):
- self.log_last_step("Starting Asterisk")
- self.ast1.start()
- self.ast1.cli_exec("fax set debug on")
- self.ast2.start()
- self.ast2.cli_exec("fax set debug on")
-
- def ami_on_connect1(self, ami):
- self.log_last_step("Connected to AMI 1")
- self.ami1 = ami
- self.ami1.originate(channel = "SIP/666 at receiver",
- application = "SendFax",
- data = "tests/udptl_v6/1page.tif")
+ def ami_connect(self, ami):
+ if ami.id == 0:
+ self.ami[0].originate(channel = "SIP/666 at receiver",
+ application = "SendFax",
+ data = "tests/udptl_v6/1page.tif")
+ elif ami.id == 1:
+ self.ami[1].registerEvent("UserEvent", self.check_recv_fax_result)
def check_recv_fax_result(self, ami, event):
- self.log_last_step("Checking Userevent")
if event.get("userevent").lower() != "recvstatus":
return
status = event.get("status")
- print ("Status of FAX is %s" % (status))
+ logger.debug("Status of FAX is %s" % (status))
if status == "SUCCESS":
self.passed = True
+ logger.debug("UDPTL test passed!")
else:
self.passed = False
-
- if self.shutdowncall.active():
- self.shutdowncall.reset(1)
-
- def ami_on_connect2(self, ami):
- self.log_last_step("Connected to AMI 2")
- self.ami2 = ami
- self.ami2.registerEvent("UserEvent", self.check_recv_fax_result)
-
- def ami_login_error(self, ami):
- self.log_last_step("AMI login failed")
- reactor.callLater(1, self.shut_down)
-
- def ami_login(self):
- self.log_last_step("Logging in to the AMI")
- self.ami_factory1 = manager.AMIFactory("user", "mysecret")
- self.ami_factory2 = manager.AMIFactory("user", "mysecret")
- self.ami_factory1.login('127.0.0.1', 5038).addCallbacks(self.ami_on_connect1, self.ami_login_error)
- self.ami_factory2.login('127.0.0.1', 5039).addCallbacks(self.ami_on_connect2, self.ami_login_error)
-
- def log_last_step(self, step):
- print step
- self.lastStep = step
-
- def shut_down(self):
- self.ast1.stop()
- self.ast2.stop()
-
- if self.passed:
- print "UDPTL test passed!"
- else:
- print "UDPTL test failed!"
-
- if reactor.running:
- print "Stopping reactor"
- reactor.stop()
+ logger.warn("UDPTL test failed!")
+ self.stop_reactor()
def run(self):
- self.start_asterisk()
- self.ami_login()
- self.shutdowncall = reactor.callLater(90, self.shut_down)
-
+ TestCase.run(self)
+ self.ast[0].cli_exec("fax set debug on")
+ self.ast[1].cli_exec("fax set debug on")
+ self.create_ami_factory(2)
def main():
# Run UDPTL test
- udptl_test = UdptlTest()
+ test = UdptlTest()
+ test.start_asterisk()
reactor.run()
- if not udptl_test.passed:
+ test.stop_asterisk()
+
+ if not test.passed:
return 1
return 0
if __name__ == "__main__":
sys.exit(main() or 0)
+
+# vim:sw=4:ts=4:expandtab:textwidth=79
More information about the asterisk-commits
mailing list