asterisk-16.7-regression-ari

ari-bug.py

import ari
import time                                                                                           
                                                   
from threading import Thread                                                                          
import logging      

logging.basicConfig(level=logging.INFO, format='%(asctime)s :: %(levelname)s :: %(message)s')
logger = logging.getLogger(__name__)

URL = 'http://localhost:5039'
USERNAME = 'username'
PASSWORD = 'password'     
APPLICATION = 'MY_APPLICATION'
                                                                                                      
client = ari.connect(base_url=URL, username=USERNAME, password=PASSWORD)
thread = None

def run():
    try:
        client.run(apps=APPLICATION)
    except Exception:
        logger.info('=== Websockets closed')


def start_websocket():
    global thread
    thread = Thread(target=run)
    thread.start()
    while not client.websockets:
        time.sleep(0.1)


def stop_websocket():
    client.close()
    thread.join()


def trigger_bug():
    try:
        client.applications.get(applicationName=APPLICATION)
    except Exception:
        logger.exception('=== BUG FOUND')
        return
    logger.info('=== No bug found')


def main():
    logger.info('=== Creating Bridge...')
    bridge = client.bridges.createWithId(bridgeId=APPLICATION, name=APPLICATION, type='holding')
    eventSource = 'bridge:{}'.format(bridge.id)

    logger.info('=== Creating Application and subscribe bridge...')
    start_websocket()
    client.applications.subscribe(applicationName=APPLICATION, eventSource=eventSource)
    stop_websocket()

    logger.info('=== Removing created bridge...')
    bridge.destroy()

    start_websocket()
    trigger_bug()
    stop_websocket()


if __name__ == '__main__':
    main()