lava-submit.py

#!/usr/bin/python

import argparse
import collections
import json
import os
import re
import string
import sys
import xmlrpclib

SERVER='validation.linaro.org/lava-server/RPC2/'

DEVICE_STREAM = {
    'panda': 'leb-panda-4430',
    'origen': 'leb-origen',
    'snowball_sd': 'leb-snowball',
    'beaglexm': 'beaglexm',
    'vexpress': 'vexpress',
    'mx53loco': 'mx53loco',
}

tests_nano = [
    'gatortests',
    'pwrmgmt',
    'perf',
]

tests_alip = [
    'bootchart',
]

tests_desktop = [
    'e2eaudiotest',
    'bluetooth-enablement',
    'wifi-enablement',
    'leb-basic-graphics',
]

def get_install_action(tests):
    return {
       'command': 'lava_test_install',
       'parameters': {
            'tests': tests
        }
    }

def get_test_actions(tests):
    actions = []
    for t in tests:
        actions.append( {
            'command': 'lava_test_run',
            'parameters': { 'test_name': t }
        })
    return actions

def get_job(jobname, device_type, img_url, tests):
    stream = '/private/team/linaro/pre-built-%s/' % DEVICE_STREAM[device_type]

    job = collections.OrderedDict()
    job['job_name'] = jobname
    job['device_type'] = device_type
    job['timeout'] = 18000
    job['actions'] = [
        {
            'command': "deploy_linaro_image",
            'parameters': {
                'image': img_url
            }
        },
            {
                'command': 'boot_linaro_image'
            }
    ]
    job['actions'].append(get_install_action(tests))
    job['actions'].extend(get_test_actions(tests))
    job['actions'].append(
        {
            'command': 'submit_results',
            'parameters': {
                'stream': stream,
                'server': 'http://validation.linaro.org/lava-server/RPC2/'
            }
        }
    )
    return json.dumps(job, indent=4)

def obfuscate_credentials(s):
    return re.sub(r"([^ ]:).+?(@)", r"\1xxx\2", s)

def main():
    p = argparse.ArgumentParser(description='submits a benchmark job to lava')
    p.add_argument('-j', dest='job_name', required=True,
                   help='name of job for LAVA')
    p.add_argument('-u', dest='img_url', required=True,
                   help='URL of image')
    p.add_argument('-d', dest='device_type', required=True,
                   help='The device type to execute on. ie "panda"')
    p.add_argument('-t', dest='image_type', default='nano',
                   help='The image type to execute on, nano (default) or ubuntu-desktop')
    args = p.parse_args()

    # test sets specific to images and boards
    tests = tests_nano

    # if ubuntu-desktop, cover more test cases (LEB)
    if args.image_type == 'ubuntu-desktop':
        tests += tests_desktop

    # if alip, specific tests like bootchart (which should be first)
    if args.image_type == 'alip':
        tests = tests_alip + tests

    # removing bluetooth and wifi for devices that don't support it
    if args.device_type in ['beaglexm', 'vexpress', 'mx53loco']:
        tests.remove('bluetooth-enablement')
        tests.remove('wifi-enablement')

    # vexpress doesn't support PM, so disable pwrmgmt
    if args.device_type in ['vexpress']:
        tests.remove('pwrmgmt')

    job = get_job(args.job_name, args.device_type, args.img_url, tests)

    user = os.environ.get('LAVA_USER')
    token = os.environ.get('LAVA_TOKEN')
    if user is None or token is None:
        print "ERROR: LAVA_USER and LAVA_TOKEN environment variables are required"
        sys.exit(1)
    server = xmlrpclib.ServerProxy("https://%s:%s@%s" % (user, token, SERVER))
    try:
        lava_job_id = server.scheduler.submit_job(job)
        print lava_job_id
    except xmlrpclib.ProtocolError, e:
        print "ERROR: Error making a LAVA request:", obfuscate_credentials(str(e))
        sys.exit(1)

if __name__ == "__main__":
    main()