test_mediaworker.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''
Criticality: High
Checks that MediaWorker can be reached using SSH
'''
import os
import imp
import sys
import subprocess

GREEN = '\033[92m'
RED = '\033[91m'
DEF = '\033[0m'

os.chdir(os.path.dirname(__file__))
if not os.path.isfile('../utils.py'):
    print('conf.sh not found')
    sys.exit(1)

es_utils = imp.load_source('es_utils', '../utils.py')
conf = es_utils.load_conf()

def check_ssh(ip):
    status, output = subprocess.getstatusoutput('timeout 2 ssh -o StrictHostKeyChecking=no %s ls /tmp' % ip)
    if status != 0:
        print('%sFailed to login using SSH%s' % (RED, DEF))
        return False
    return True

all_ok = True

worker_ip = conf.get('CELERITY_WORKER_IP')
if worker_ip != '127.0.1.1':
    if not check_ssh(worker_ip):
        all_ok = False