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__)))


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
    from devices import board_decider, debian, logstash, elasticlogger

    # 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:
            # 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', None),
                                                 tftp_username=config.board.get('wan_username', 'root'),
                                                 tftp_password=config.board.get('wan_password', 'bigfoot1'),
                                                 connection_type=config.board.get('connection_type', 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'))
            # spawn wan and lan consoles if present
            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"))

        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()