[svn-commits] pabelanger: branch pabelanger/testClass r1120 - in /asterisk/team/pabelanger/...
SVN commits to the Digium repositories
svn-commits at lists.digium.com
Wed Dec 8 20:46:52 CST 2010
Author: pabelanger
Date: Wed Dec 8 20:46:48 2010
New Revision: 1120
URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=1120
Log:
Move AMIFactory into TestCase :D
Modified:
asterisk/team/pabelanger/testClass/lib/python/asterisk/TestCase.py
asterisk/team/pabelanger/testClass/tests/manager/login/run-test
asterisk/team/pabelanger/testClass/tests/udptl/configs/ast2/manager.conf
asterisk/team/pabelanger/testClass/tests/udptl/run-test
Modified: asterisk/team/pabelanger/testClass/lib/python/asterisk/TestCase.py
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/pabelanger/testClass/lib/python/asterisk/TestCase.py?view=diff&rev=1120&r1=1119&r2=1120
==============================================================================
--- asterisk/team/pabelanger/testClass/lib/python/asterisk/TestCase.py (original)
+++ asterisk/team/pabelanger/testClass/lib/python/asterisk/TestCase.py Wed Dec 8 20:46:48 2010
@@ -10,6 +10,7 @@
import sys
from optparse import OptionParser
from twisted.internet import reactor
+from starpy import manager
from asterisk import Asterisk
@@ -38,6 +39,12 @@
reactor.callWhenRunning(self.run)
def create_asterisk(self, count=1):
+ """
+
+ Keywork arguments:
+ count --
+
+ """
for c in range(count):
print "Creating Asterisk instance %d ..." % (c + 1)
self.ast.append(Asterisk(base=self.options.base))
@@ -48,22 +55,55 @@
self.ast[c].install_configs("%s/configs/ast%d" %
(self.options.test_name, c + 1))
+ def create_ami_factory(self, count=1, username=None, secret=None, port=5038):
+ """
+
+ Keywork arguments:
+ count --
+ username --
+ secret --
+ port --
+
+ """
+ for c in range(count):
+ host = "127.0.0.%d" % (c + 1)
+ print "Creating AMIFactory %d ..." % (c + 1)
+ ami_method = getattr(self, "ami_connect%d" % (c + 1))
+ self.ami_factory = manager.AMIFactory(username, secret)
+ self.ami_factory.login(host).addCallbacks(ami_method,
+ self.ami_login_error)
+
def start_asterisk(self):
+ """
+
+ """
for index, item in enumerate(self.ast):
print "Starting Asterisk instance %d ..." % (index + 1)
self.ast[index].start()
def stop_asterisk(self):
+ """
+
+ """
for index, item in enumerate(self.ast):
print "Stopping Asterisk instance %d ..." % (index + 1)
self.ast[index].stop()
def stop_reactor(self):
+ """
+
+ """
print "Stopping Reactor ..."
if reactor.running:
reactor.stop()
def run(self):
+ """
+
+ """
reactor.callLater(self.reactor_timeout, self.stop_reactor)
+ def ami_login_error(self):
+ print "Error logging into AMI"
+ self.stop_reactor()
Modified: asterisk/team/pabelanger/testClass/tests/manager/login/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/pabelanger/testClass/tests/manager/login/run-test?view=diff&rev=1120&r1=1119&r2=1120
==============================================================================
--- asterisk/team/pabelanger/testClass/tests/manager/login/run-test (original)
+++ asterisk/team/pabelanger/testClass/tests/manager/login/run-test Wed Dec 8 20:46:48 2010
@@ -9,7 +9,7 @@
import sys
from twisted.internet import reactor
-from starpy import manager
+
sys.path.append("lib/python")
from asterisk.asterisk import Asterisk
@@ -17,38 +17,20 @@
class AMILoginTest(TestCase):
def __init__(self, argv):
- self.last_step = ""
- self.ami = None
-
TestCase.__init__(self, argv)
self.create_asterisk()
- def log_last_step(self, step):
- print step
- self.last_step = step
-
- def on_error(self, ami):
- print "ERROR, Last Step: %s" % self.last_step
- self.stop_reactor()
-
- def on_logoff(self, ami):
- self.log_last_step("Logoff Successful")
+ def ami_logoff(self, ami):
self.passed = True
self.stop_reactor()
- def on_connect(self, ami):
- self.log_last_step("Connected to the AMI")
- self.ami = ami
- self.ami.logoff().addCallbacks(self.on_logoff, self.on_error)
- self.log_last_step("Logging off")
+ def ami_connect1(self, ami):
+ print "AMI Connect"
+ ami.logoff().addCallbacks(self.ami_logoff, self.ami_login_error)
def run(self):
TestCase.run(self)
-
- self.log_last_step("Logging in to the AMI")
- self.ami_factory = manager.AMIFactory("user", "mysecret")
- self.ami_factory.login('127.0.0.1', 5038).addCallbacks(self.on_connect,
- self.on_error)
+ self.create_ami_factory(1, "user", "mysecret")
def main(argv=None):
if argv is None:
Modified: asterisk/team/pabelanger/testClass/tests/udptl/configs/ast2/manager.conf
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/pabelanger/testClass/tests/udptl/configs/ast2/manager.conf?view=diff&rev=1120&r1=1119&r2=1120
==============================================================================
--- asterisk/team/pabelanger/testClass/tests/udptl/configs/ast2/manager.conf (original)
+++ asterisk/team/pabelanger/testClass/tests/udptl/configs/ast2/manager.conf Wed Dec 8 20:46:48 2010
@@ -1,7 +1,7 @@
[general]
enabled = yes
-port = 5039
-bindaddr = 127.0.0.1
+port = 5038
+bindaddr = 127.0.0.2
[user]
secret = mysecret
Modified: asterisk/team/pabelanger/testClass/tests/udptl/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/pabelanger/testClass/tests/udptl/run-test?view=diff&rev=1120&r1=1119&r2=1120
==============================================================================
--- asterisk/team/pabelanger/testClass/tests/udptl/run-test (original)
+++ asterisk/team/pabelanger/testClass/tests/udptl/run-test Wed Dec 8 20:46:48 2010
@@ -17,21 +17,18 @@
class UdptlTest(TestCase):
def __init__(self, argv):
- self.last_step = ""
self.reactor_timeout = 60
TestCase.__init__(self, argv)
self.create_asterisk(2)
- 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",
+ def ami_connect1(self, ami):
+ print "AMI Connect 1"
+ ami.originate(channel = "SIP/666 at receiver",
application = "SendFax",
data = "tests/udptl/1page.tif")
def check_recv_fax_result(self, ami, event):
- self.log_last_step("Checking Userevent")
if event.get("userevent").lower() != "recvstatus":
return
self.stop_reactor()
@@ -45,25 +42,13 @@
self.passed = False
print "UDPTL test failed!"
- def ami_on_connect2(self, ami):
- self.log_last_step("Connected to AMI 2")
+ def ami_connect2(self, ami):
+ print "AMI Connect 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.stop_reactor)
-
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
+ self.create_ami_factory(2, "user", "mysecret")
def run(self):
TestCase.run(self)
More information about the svn-commits
mailing list