KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use quit_application method in robotframework-appiumlibrary

Best Python code snippet using robotframework-appiumlibrary_python

caller.py

Source:caller.py

1#!/usr/bin/env python2# $Id: registration.py 2171 2008-07-24 09:01:33Z bennylp $3#4# SIP account and registration sample. In this sample, the program5# will block to wait until registration is complete6#7# Copyright (C) 2003-2008 Benny Prijono <benny@prijono.org>8#9# This program is free software; you can redistribute it and/or modify10# it under the terms of the GNU General Public License as published by11# the Free Software Foundation; either version 2 of the License, or12# (at your option) any later version.13#14# This program is distributed in the hope that it will be useful,15# but WITHOUT ANY WARRANTY; without even the implied warranty of16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the17# GNU General Public License for more details.18#19# You should have received a copy of the GNU General Public License20# along with this program; if not, write to the Free Software21# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA22#23import sys24import pjsua as pj25import threading26import signal27import time28import ConfigParser29import argparse30current_call = None31quit_application = False32player_ringing = None33sip_extension = None34sip_password = None35sip_domain = None36sip_reception_extension = None37sip_peer = None38audio_micro_id = None39audio_speaker_id = None40default_config_file = "config.ini"41def _get_config(path):42    global sip_extension43    global sip_password44    global sip_domain45    global sip_reception_extension46    global audio_micro_id47    global audio_speaker_id48    config = ConfigParser.ConfigParser()49    config.read(path)50    # get sip config51    sip_extension = config.get('sip', 'extension')52    sip_password = config.get('sip', 'password')53    sip_domain = config.get('sip', 'domain')54    sip_reception_extension = config.get('sip', 'reception_extension')55    # get audio config56    audio_micro_id = config.get('audio', 'micro_id')57    audio_speaker_id = config.get('audio', 'speaker_id')58def _args_parser(args):59    global sip_peer60    global sip_reception_extension61    if hasattr(args, "config") and args.config:62        config_file_path = args.config63    else:64        print("missing configuration file")65        lib.destroy()66        sys.exit(1)67    try:68        _get_config(config_file_path)69    except Exception:70        print("configuration incorrect")71        lib.destroy()72        sys.exit(1)73    if hasattr(args, "reception") and args.reception:74        sip_peer = sip_reception_extension75    elif hasattr(args, "extension") and args.extension:76        sip_peer = args.extension77    else:78        print("missing parameters")79def log_cb(level, str, len):80    print str,81def _sig_handler(signum, frame):82    global quit_application83    if current_call:84        current_call.hangup()85        quit_application = True86class MyAccountCallback(pj.AccountCallback):87    sem = None88    def __init__(self, account):89        pj.AccountCallback.__init__(self, account)90    def wait(self):91        self.sem = threading.Semaphore(0)92        self.sem.acquire()93    def on_reg_state(self):94        if self.sem:95            if self.account.info().reg_status >= 200:96                self.sem.release()97# Callback to receive events from Call98class MyCallCallback(pj.CallCallback):99    def __init__(self, call=None):100        pj.CallCallback.__init__(self, call)101    # Notification when call state has changed102    def on_state(self):103        global current_call104        global cv105        global quit_application106        global player_ringing107        if self.call.info().state == pj.CallState.DISCONNECTED and current_call:108            # we get disconnected109            current_call = None110            quit_application = True111        elif (self.call.info().state == pj.CallState.CONFIRMED and112                not player_ringing):113            # stop the ringing tone114            lib.conf_disconnect(lib.player_get_slot(player_ringing), 0)115    # Notification when call's media state has changed.116    def on_media_state(self):117        if self.call.info().media_state == pj.MediaState.ACTIVE:118            # Connect the call to sound device119            call_slot = self.call.info().conf_slot120            pj.Lib.instance().conf_connect(call_slot, 0)121            pj.Lib.instance().conf_connect(0, call_slot)122            print "Media is now active"123        else:124            print "Media is inactive"125lib = pj.Lib()126signal.signal(signal.SIGINT, _sig_handler)127signal.signal(signal.SIGTERM, _sig_handler)128parser = argparse.ArgumentParser()129parser.add_argument('-c', '--config', help='use a configuration file')130parser.add_argument('-e', '--extension', help='call a sip extension')131parser.add_argument('-r', '--reception', help='contact the reception', action='store_true')132_args_parser(parser.parse_args())133try:134    lib.init(log_cfg = pj.LogConfig(level=1, callback=log_cb))135    lib.create_transport(pj.TransportType.UDP, pj.TransportConfig(5080))136    # set built-in pieso speaker to output and usb micro to input137    lib.set_snd_dev(int(audio_micro_id), int(audio_speaker_id))138    lib.start()139    acc = lib.create_account(140            pj.AccountConfig(sip_domain, sip_extension, sip_password))141    acc_cb = MyAccountCallback(acc)142    acc.set_callback(acc_cb)143    acc_cb.wait()144    print "\n"145    print "Registration complete, status=", acc.info().reg_status, \146          "(" + acc.info().reg_reason + ")"147    if sip_peer:148        # play ringing tone until the call is confirmed149        player_ringing = lib.create_player("ringing.wav", loop=True)150        lib.conf_connect(lib.player_get_slot(player_ringing), 0)151        current_call = acc.make_call("sip:"+sip_peer+"@"+sip_domain, cb=MyCallCallback())152        while True:153            # a main loop checking every second if we shall quit or not154            time.sleep(1)155            if quit_application:156                break157    else:158        print "please provide an extension to call"159except pj.Error, e:160    print "Exception: " + str(e)...

modelhandler.py

Source:modelhandler.py

...61        else:62            self.controller_queue.put(["user_authenticated", False, None, user_id])63    def admin_get_users_data(self, *args):64        self.controller_queue.put(["admin_users_data", self.validator.get_users_data()])65    def quit_application(self):66        # print("quit application request")67        self.flag = False68class ModelHandlerTesting:69    def model_proc(self, v_q, c_q, m_q):70        m_handler = ModelHandler(v_q, c_q, m_q)71    def controller_proc(self, v_q, c_q, m_q):72        # print("putting data on queue")73        m_q.put(["admin_get_users_data", "alex", "admin"])74        m_q.put(["validate_credentials", "11", "alex"])75        m_q.put(["validate_credentials", "12", "adam"])76        m_q.put(["validate_credentials", "13", "jack"])77        m_q.put(["validate_credentials", "14", "john"])78        m_q.put(["validate_credentials", "15", "zed"])79        while True:...

controllerhandler.py

Source:controllerhandler.py

...61        # print("load complete")62        self.view_queue.put(["load_complete"])63    def admin_get_users_data(self, *args):64        self.model_queue.put(["admin_get_users_data", *args])65    def quit_application(self):66        # print("quit application request")67        self.flag = False68class ControllerHandlerTesting:69    def controller_proc(self, v_q, c_q, m_q):70        c_handler = ControllerHandler(v_q, c_q, m_q)71    def other_proc(self, v_q, c_q, m_q):72        # print("putting data on queue")73        c_q.put(["admin_get_users_data", "alex", "admin", "root"])74        c_q.put(["quit_application"])75    def test(self):76        v_q = mp.Queue()77        c_q = mp.Queue()78        m_q = mp.Queue()79        c_proc = mp.Process(...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.