[svn-commits] pabelanger: branch pabelanger/DirectedPickupTest r2022 - in /asterisk/team/pa...
SVN commits to the Digium repositories
svn-commits at lists.digium.com
Tue Aug 30 07:52:20 CDT 2011
Author: pabelanger
Date: Tue Aug 30 07:52:15 2011
New Revision: 2022
URL: http://svnview.digium.com/svn/testsuite?view=rev&rev=2022
Log:
Rewrite test
Modified:
asterisk/team/pabelanger/DirectedPickupTest/tests/directed_pickup/configs/ast1/extensions.conf
asterisk/team/pabelanger/DirectedPickupTest/tests/directed_pickup/run-test
Modified: asterisk/team/pabelanger/DirectedPickupTest/tests/directed_pickup/configs/ast1/extensions.conf
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/pabelanger/DirectedPickupTest/tests/directed_pickup/configs/ast1/extensions.conf?view=diff&rev=2022&r1=2021&r2=2022
==============================================================================
--- asterisk/team/pabelanger/DirectedPickupTest/tests/directed_pickup/configs/ast1/extensions.conf (original)
+++ asterisk/team/pabelanger/DirectedPickupTest/tests/directed_pickup/configs/ast1/extensions.conf Tue Aug 30 07:52:15 2011
@@ -3,33 +3,14 @@
[globals]
[pickuptest]
-exten => h,1,GoToIf($[${RINGING_EXTEN}=1]?is_ringing:bye)
-; this is the ringing channel, see if it got picked up correctly or not.
-exten => h,n(is_ringing),GoToIf($[${RINGING_EXTEN_PASS}=1]?is_ringing_passed:is_ringing_failed)
-exten => h,n(is_ringing_failed),AGI(agi://127.0.0.1:4574) ;Tell the test script pickup failed
-exten => h,n,GoTo(pickuptest,h,bye) ; finish
-exten => h,n(is_ringing_passed),AGI(agi://127.0.0.1:4573) ;Tell the test script pickup passed
-exten => h,n(bye),NoOp(Bye ${CHANNEL})
+exten => 1001,1,NoOp()
+ same => n,Answer()
+ same => n,Dial(IAX2/iax_pickup_b/ringing_exten)
-exten => 1001,1,Dial(IAX2/iax_pickup_b at 127.0.0.1/ringing_exten)
+exten => ringing_exten,1,NoOp()
+ same => n,Ringing()
+ same => n,UserEvent(TestStatus,status:RINGING)
+ same => n,Background(demo-congrats,noanswer)
-;test 1 with no options
-exten => 1002,1,PickupChan(IAX2/iax_pickup_b)
-exten => 1002,n,Hangup()
-
-;test 2 with 'p' option
-exten => 1003,1,PickupChan(IAX2/iax_pickup,p)
-exten => 1003,n,Hangup()
-
-; this exten Answers and waits for the other side to answer
-exten => waiting_exten,1,Answer()
-exten => waiting_exten,n,Wait(10)
-
-; this exten never answers and expects another channel to pick it up during its wait period
-exten => ringing_exten,1,Set(RINGING_EXTEN=1)
-exten => ringing_exten,n,Set(RINGING_EXTEN_PASS=1)
-exten => ringing_exten,n,Wait(10)
-exten => ringing_exten,n,Verbose(This should not get called)
-; this exten should be hungup on before reaching this
-exten => ringing_exten,n,Set(RINGING_EXTEN_PASS=0)
-exten => ringing_exten,n,Hangup()
+exten => 1002,1,NoOp()
+ same => n,PickupChan(IAX2/iax_pickup_b)
Modified: asterisk/team/pabelanger/DirectedPickupTest/tests/directed_pickup/run-test
URL: http://svnview.digium.com/svn/testsuite/asterisk/team/pabelanger/DirectedPickupTest/tests/directed_pickup/run-test?view=diff&rev=2022&r1=2021&r2=2022
==============================================================================
--- asterisk/team/pabelanger/DirectedPickupTest/tests/directed_pickup/run-test (original)
+++ asterisk/team/pabelanger/DirectedPickupTest/tests/directed_pickup/run-test Tue Aug 30 07:52:15 2011
@@ -9,138 +9,79 @@
import sys
import os
-import time
-from twisted.application import service, internet
+import logging
+
from twisted.internet import reactor
-from starpy import fastagi
sys.path.append("lib/python")
from asterisk.asterisk import Asterisk
from asterisk.version import AsteriskVersion
+from asterisk.TestCase import TestCase
-workingdir = "directed_pickup"
-testdir = "tests/%s" % workingdir
+logger = logging.getLogger(__name__)
-NOT_RUN = 0
-PASS = 1
-FAIL = 2
-SKIP = 4
-class PickupTest:
+class PickupTest(TestCase):
+ firstChannel = None
+ secondChannel = None
+ bridgeCount = 0
+
def __init__(self):
+ TestCase.__init__(self)
+ self.create_asterisk()
- # Test timeout in seconds
- self.test_to = 15
- self.last_step = ""
- self.test1_res = NOT_RUN
- self.test2_res = NOT_RUN
- self.passed = FAIL
-
- # FastAGI, listen for results from dialplan
- # PASS on port 4573
- # FAIL on port 4574
- self.agi_pass = fastagi.FastAGIFactory(self.report_test_passed)
- self.agi_fail = fastagi.FastAGIFactory(self.report_test_failed)
- reactor.listenTCP(4573, self.agi_pass, self.test_to, '127.0.0.1')
- reactor.listenTCP(4574, self.agi_fail, self.test_to, '127.0.0.1')
-
- reactor.callWhenRunning(self.run)
-
- self.ast1 = Asterisk(base=workingdir)
- self.ast1.install_configs("%s/configs/ast1" % (testdir))
-
- def shut_down(self):
- self.log_last_step("Reading results")
-
+ def hangup(self, ami):
self.stop_reactor()
- if self.test1_res != FAIL and self.test2_res != FAIL:
+ def check_Bridge(self, ami, event):
+ if self.bridgeCount != 0:
+ return
+ self.bridgeCount += 1
+
+ channel1 = event.get('channel1')
+ channel2 = event.get('channel2')
+
+ if channel1 == self.firstChannel and channel2 == self.secondChannel:
self.passed = True
- print "Tests Passed"
- else:
- if self.test1_res != PASS:
- print "Test 1 failed, Test 2 did not run"
- else:
- print "Test 1 Passed, Test 2 failed"
- print "Tests Failed"
- # Close out the agi
- def handle_agi_result(self, agi):
- sequence = fastagi.InSequence()
- sequence.append(agi.finish)
- return sequence()
+ self.ami[0].hangup(self.firstChannel).addCallback(self.hangup)
- # set the result, schedule next step
- def handle_result(self, result):
- if self.test1_res == NOT_RUN:
- self.log_last_step("Pickup Test 1 results came in.")
- self.test1_res = result
- elif self.test2_res == NOT_RUN:
- self.log_last_step("Pickup Test 2 results came in.")
- self.test2_res = result
+ def check_Pickup(self, ami, event):
+ self.secondChannel = event.get('channel')
+ self.ami[0].registerEvent('Bridge', self.check_Bridge)
- if self.test1_res == FAIL or self.test2_res == FAIL:
- # schedule early shutdown, a test failed
- self.shut_down()
- elif self.test1_res != NOT_RUN and self.test2_res != NOT_RUN:
- # schedule early shutdown, all tests ran
- self.shut_down()
- elif self.test2_res == NOT_RUN:
- # schedule test2 if test1 passed
- reactor.callLater(2, self.launch_test2)
+ def check_UserEvent(self, ami, event):
+ if event.get("userevent") != "TestStatus":
+ return
- # This gets invoked by the dialplan when pickup test passes
- def report_test_passed(self, agi):
- self.handle_result(PASS)
- return self.handle_agi_result(agi)
+ status = event.get('status')
+ if status != 'RINGING':
+ return
- # This gets invoked by the dialplan when pickup test fails
- def report_test_failed(self, agi):
- self.handle_result(FAIL)
- return self.handle_agi_result(agi)
+ self.ami[0].registerEvent('Pickup', self.check_Pickup)
+ self.ami[0].originate(
+ channel = "Local/1002 at pickuptest",
+ application = "Echo",
+ )
- def test1_pickup(self):
- self.log_last_step("Issuing the pickup")
- self.ast1.cli_originate("Local/1002 at pickuptest application echo")
+ def check_Dial(self, ami, event):
+ dialstring = event.get('dialstring')
+ if dialstring != 'iax_pickup_b/ringing_exten':
+ return
- def launch_test1(self):
- self.log_last_step("Starting Test 1")
- self.ast1.cli_originate("IAX2/iax_pickup_b at 127.0.0.1/waiting_exten extension 1001 at pickuptest")
- self.test1_pickup()
+ self.firstChannel = event.get('channel')
- def test2_pickup(self):
- self.log_last_step("Issuing the pickup")
- self.ast1.cli_originate("Local/1003 at pickuptest application echo")
-
- def launch_test2(self):
- self.log_last_step("Starting Test 2")
- if self.ast1.ast_version < AsteriskVersion('1.8'):
- self.test2 = SKIP
- self.shut_down()
- else:
- self.ast1.cli_originate("IAX2/iax_pickup_b at 127.0.0.1/waiting_exten extension 1001 at pickuptest")
- self.test2_pickup()
-
- def start_asterisk(self):
- self.log_last_step("Starting Asterisk")
- self.ast1.start()
-
- def stop_asterisk(self):
- self.ast1.stop()
-
- def log_last_step(self, step):
- print step
- self.last_step = step
-
- def stop_reactor(self):
- print "Stopping Reactor ..."
- if reactor.running:
- reactor.stop()
+ def ami_connect(self, ami):
+ TestCase.ami_connect(self, ami)
+ self.ami[0].registerEvent("Dial", self.check_Dial)
+ self.ami[0].registerEvent("UserEvent", self.check_UserEvent)
+ self.ami[0].originate(
+ channel = "IAX2/iax_pickup_b/1001",
+ application = "Echo",
+ )
def run(self):
- self.launch_test1()
-
- # stop reactor after timeout
- reactor.callLater(self.test_to, self.stop_reactor)
+ TestCase.run(self)
+ self.create_ami_factory()
def main():
@@ -148,10 +89,11 @@
test.start_asterisk()
reactor.run()
test.stop_asterisk()
- if test.passed != PASS:
+ if test.passed != True:
return 1
return 0
if __name__ == "__main__":
sys.exit(main() or 0)
+# vim:sw=4:ts=4:expandtab:textwidth=79
More information about the svn-commits
mailing list