bft

#!/usr/bin/env python

# Copyright (c) 2015
#
# All rights reserved.
#
# This file is distributed under the Clear BSD license.
# The full text can be found in LICENSE in the root directory.

import datetime
import inspect
import os
import random
import sys
import unittest2
import junitxml
import json

# Put this directory into the python path, so
# that devices may be imported.
import site
site.addsitedir(os.path.dirname(os.path.realpath(__file__)))

from devices import board_decider, debian, logstash, elasticlogger

def setup_legacy_devices(config):
    '''This will setup legacy hard coded devices to into the config'''

    config.wan = None
    config.lan = None
    config.wlan = None
    config.wlan5g = None
    config.wlan2g = None

    if config.board.get('wan_device'):
        config.wan = debian.DebianBox(config.board.get('wan_device'),
                                      color='cyan', reboot=config.reboot_vms,
                                      location=config.board.get('location'),
                                      username=config.board.get('wan_username', "root"),
                                      password=config.board.get('wan_password', "bigfoot1"),
                                      port=config.board.get('wan_port', "22"))
    if config.board.get('lan_device'):
        config.lan = debian.DebianBox(config.board.get('lan_device'), color='blue', reboot=config.reboot_vms,
                                    username=config.board.get('lan_username', "root"),
                                    password=config.board.get('lan_password', "bigfoot1"),
                                    port=config.board.get('lan_port', "22"))
    if config.board.get('wlan_device'):
        config.wlan = debian.DebianBox(config.board.get('wlan_device'), color='green', reboot=config.reboot_vms,
                                    username=config.board.get('wlan_username', "root"),
                                    password=config.board.get('wlan_password', "bigfoot1"),
                                    port=config.board.get('wlan_port', "22"))
    if config.board.get('5g_device'):
        config.wlan5g = debian.DebianBox(config.board.get('5g_device'), color='grey', reboot=config.reboot_vms,
                                        username=config.board.get('5g_username', "root"),
                                        password=config.board.get('5g_password', "bigfoot1"),
                                        port=config.board.get('5g_port', "22"))
    if config.board.get('2g_device'):
        config.wlan2g = debian.DebianBox(config.board.get('2g_device'), color='magenta', reboot=config.reboot_vms,
                                        username=config.board.get('2g_username', "root"),
                                        password=config.board.get('2g_password', "bigfoot1"),
                                        port=config.board.get('2g_port', "22"))


def setup_dynamic_devices(config):
    '''Sets up dynamic devices from devices node in JSON config file'''
    config.devices = []
    for device in config.board['devices']:
        if device['type'] == 'debian':
            d = debian.DebianBox(name=device.get('ipaddr', None),
                color=device.get('color', 'black'),
                reboot=config.reboot_vms,
                username=device.get('username', "root"),
                password=device.get('password', "bigfoot1"),
                port=device.get('port', "22"),
                pre_cmd_host=device.get('pre_cmd_host', None),
                cmd=device.get('cmd', None),
                post_cmd_host=device.get('post_cmd_host', None),
                post_cmd=device.get('post_cmd', None),
                cleanup_cmd=device.get('cleanup_cmd', None))
            setattr(config, device['name'], d)
            config.devices.append(device['name'])
        else:
            print("Unknown device type for %d", device)

def main():
    '''Connect to devices, run tests, record results.'''

    # Read command-line arguments
    import arguments
    config = arguments.parse()

    import library
    import devices
    from termcolor import colored
    from library import print_bold

    # Connect to any board in list
    connected_to_board = False
    random.shuffle(config.BOARD_NAMES)
    for name in config.BOARD_NAMES:
        try:
            config.board = config.boardfarm_config[name]
        except Exception as e:
            print(e)
            print("Error reading info about board %s from board farm configuration." % name)
            break

        print_bold("Connecting to board named = %s, type = %s ..." % (name, config.board['board_type']))
        try:
            # If we find a dynamic device with the tftpd-server option let's use that for the server
            # otherwise we use the device with the 'wan' name... finally if we don't find an ip address let's
            # manually figure it out later
            tftp_server = None
            tftp_port = "22"
            try:
                if 'devices' in config.board:
                    tftp_dev = [ x for x in config.board['devices'] if 'tftpd-server' in x.get('options', "") ]
                    if len(tftp_dev) == 1:
                        tftp_server = tftp_dev[0]['ipaddr']
                        tftp_port = tftp_dev[0].get('port', '22')
                    else:
                        print("WARN: more than one TFTP server is configured, please pick one")
                elif tftp_server is None:
                    if len(tftp_dev) == 1:
                        tftp_dev = [ x for x in config.board['devices'] if x['name'] == 'wan' ]
                        tftp_server = tftp_dev[0]['ipaddr']
                        tftp_port = tftp_dev[0].get('port', '22')
                    else:
                        print("WARN: more than one TFTP server is configured, please pick one")
            except:
                # if not by ipaddr let's connect to the lan device later and update
                pass

            # Connect to board
            config.console = board_decider.board(config.board['board_type'],
                                                 conn_cmd=config.board['conn_cmd'],
                                                 power_ip=config.board.get('powerip', None),
                                                 power_outlet=config.board.get('powerport', None),
                                                 web_proxy=config.board.get('lan_device', None),
                                                 tftp_server=config.board.get('wan_device', tftp_server),
                                                 tftp_username=config.board.get('wan_username', 'root'),
                                                 tftp_password=config.board.get('wan_password', 'bigfoot1'),
                                                 tftp_port=config.board.get('wan_port', tftp_port),
                                                 connection_type=config.board.get('connection_type', None),
                                                 ssh_password=config.board.get('ssh_password', None),
                                                 power_username=config.board.get('power_username', None),
                                                 power_password=config.board.get('power_password', None))
            print_bold("dut device console = %s" % colored("black", 'grey'))

            setup_legacy_devices(config)

            if 'devices' in config.board:
                setup_dynamic_devices(config)

            # legacy wan device
            if tftp_server is None and hasattr(config, 'wan'):
                saved = config.console.logfile_read
                config.console.logfile_read = None
                config.console.tftp_server = config.wan.get_ip_addr("eth0")
                config.console.tftp_username = "root"
                config.console.tftp_password = "bigfoot1"
                config.console.tftp_port = "22"
                config.console.logfile_read = saved

        except Exception as e:
            print(e)
            connected_to_board = False
            continue
        connected_to_board = True
        break
    if not connected_to_board:
        print_bold("Failed to connect to any board")
        sys.exit(2)

    try:
        print_bold("Using Board %s, User %s" % (name, os.environ['BUILD_USER_ID']))
    except:
        print_bold("Using Board %s, User %s" % (name, os.environ['USER']))

    # Make devices (board, lan, wan, available to tests easily)
    devices.initialize_devices(config)

    # Write board info to json file and stdout
    config.board['station'] = name
    print_bold('\n==========')
    library.print_board_info(config.board)

    # Run tests
    os.environ['TEST_START_TIME'] = datetime.datetime.now().strftime("%s")
    result_name = os.path.join(config.output_dir + "test_results.xml")
    result_file = open(result_name, "w")
    result = junitxml.JUnitXmlResult(result_file)
    result.startTestRun()
    tests_to_run = []
    suite = unittest2.TestSuite()
    # Add tests from specified suite
    print_bold('==========\nTest suite "%s" has been specified, will attempt to run tests:' % config.TEST_SUITE)
    import tests
    import testsuites
    for i, name in enumerate(testsuites.list_tests[config.TEST_SUITE]):
        if isinstance(name, str):
            test = getattr(tests, name)
        else:
            test = name
        print_bold("  %s %s from %s" % (i+1, test.__name__, inspect.getfile(test)))
        tests_to_run.append(test(config))
    if hasattr(config, 'EXTRA_TESTS') and config.EXTRA_TESTS:
        if tests_to_run[-1].__class__.__name__ == "Interact":
            print_bold("Last test is interact in testsuite, removing")
            tests_to_run.pop()

        print_bold("Extra tests specified on command line:")
        try:
            for t in [getattr(tests, name) for name in config.EXTRA_TESTS]:
                print_bold("  %s" % t)
                tests_to_run.append(t(config))
        except:
            print_bold("Unable to find specified extra tests, aborting...")
            sys.exit(1)

    for x in tests_to_run:
        suite.addTest(x)

    print_bold('==========')
    try:
        print_bold(suite.run(result))
    except KeyboardInterrupt:
        print_bold("Run interrupted. Wrapping up...")
    result.stopTestRun()
    try:
        config.console.close()
        config.lan.close()
        config.wan.close()
    except Exception as e:
        print(e)
        print_bold("For some reason, could not close a connection.")
    print_bold("Wrote %s" % result_name)
    library.print_board_info(config.board)
    result_file.close()

    with open(os.path.join(config.output_dir, 'console.log'), 'w') as clog:
        clog.write(config.console.log)

    os.environ['TEST_END_TIME'] = datetime.datetime.now().strftime("%s")

    # Write test result messages to a file
    full_results = library.process_test_results(tests_to_run)
    json.dump(full_results,
              open(os.path.join(config.output_dir + 'test_results.json'), 'w'),
              indent=4,
              sort_keys=True)

    # run all analysis classes (post processing)
    # also, never fail so we don't block automation
    try:
        import analysis
        for cstr in dir(analysis):
            c = getattr(analysis, cstr)
            if inspect.isclass(c) and issubclass(c, analysis.Analysis):
                c().analyze(config.console.log, config.output_dir)
    except Exception as e:
        if not issubclass(type(e), (StopIteration)):
            print("Failed to run anaylsis:")
            print(e)

    # Try to remotely log information about this run
    info_for_remote_log = dict(config.board)
    info_for_remote_log.update(full_results)
    try:
        info_for_remote_log['duration'] = int(os.environ['TEST_END_TIME'])-int(os.environ['TEST_START_TIME'])
    except:
        pass
    if hasattr(config, 'TEST_SUITE'):
        info_for_remote_log['test_suite'] = str(config.TEST_SUITE)
    # logstash cannot handle multi-level json, remove full test results
    info_for_remote_log.pop('test_results', None)
    # but we will add back specific test results data
    for t in tests_to_run:
        if hasattr(t, 'override_kibana_name'):
            n = t.override_kibana_name
        else:
            n = t.__class__.__name__
        for k, v in t.logged.items():
            info_for_remote_log[n + '-' + k] = v
        if hasattr(t, 'result_grade'):
            info_for_remote_log[n + "-result"] = t.result_grade

    try:
        if config.logging_server is not None:
            logstash.RemoteLogger(config.logging_server).log(info_for_remote_log)
    except Exception as e:
        print(e)
        print("Unable to access logging_server specified in config. "
              "Results stored only locally.")

    try:
        if config.elasticsearch_server is not None:
            elasticlogger.ElasticsearchLogger(config.elasticsearch_server).log(info_for_remote_log)
        else:
            print("No elasticsearch_server specified in config. Results stored locally")
    except Exception as e:
        print(e)
        print("Unable to store results to elasticsearch_server specified in config. "
              "Results stored locally.")

    # Create Pretty HTML output
    import make_human_readable
    try:
        title_str = make_human_readable.get_title()
        make_human_readable.xmlresults_to_html(full_results['test_results'], title=title_str,
                                output_name=os.path.join(config.output_dir, "results.html"),
                                board_info=config.board)
    except Exception as e:
        print(e)
        print("Unable to create HTML results")

    # Send url of pretty html results to MySQL build database
    try:
        library.send_results_to_myqsl(config.TEST_SUITE, config.output_dir)
    except Exception as e:
        print(e)
        print("Unable to log results to mysql database.")


if __name__ == '__main__':
    main()