# -*- coding: utf-8 -*-
#
# Copyright (c) 2011 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

# DESCRIPTION :
# This factory test runs at the start of a test sequence to verify the DUT has
# been setup correctly.
#
# The start provides several settings (set via darg):
# 'require_external_power': Prompts and waits for external power to be applied.
# 'require_shop_floor': Prompts and waits for serial number as input.  The
#       server is default to the host running mini-omaha, unless you specify an
#       URL by 'shop_floor_server_url' darg.
# 'press_to_continue': Prompts and waits for a key press (SPACE) to continue.

import glob
import logging
import os
import socket
import sys

import gobject
import gtk
import pango

from autotest_lib.client.bin import test, utils
from autotest_lib.client.common_lib import error
from autotest_lib.client.cros import factory_setup_modules
from cros.factory.test import factory
from cros.factory.test import shopfloor
from cros.factory.test import task
from cros.factory.test import ui
from cros.factory.test.event import Event, EventClient
from cros.factory.event_log import EventLog


# Messages for tasks
_MSG_TASK_POWER = (
        'Plug in external power to continue.\n'
        '请插上外接电源以继续。')
_MSG_TASK_SERIAL = (
        'Enter valid serial number:\n'
        '请输入有效的序号:')
_MSG_TASK_SPACE = (
        'Hit SPACE to start testing...\n'
        '按 "空白键" 开始测试...')
_MSG_TASK_SHOP_FLOOR = (
        'Preparing to connect to shop floor server.\n'
        'Please plug in network cable and hit SPACE\n'
        'to start testing...\n'
        '准备连线 shop floor 伺服器。 \n'
        '请插上网路线后，按下"空白键"开始测试...')

_MSG_NO_SHOP_FLOOR_SERVER_URL = (
        'No shop floor server URL. Auto-testing stopped.\n\n'
        'Please install the factory test image using the mini-Omaha server\n'
        'rather than booting from a USB drive.\n\n'
        'For debugging or development, use the listed hot-keys to start\n'
        'individual tests.\n\n'
        '未指定 Shop Floor 服务器位址，停止自动测试。\n\n'
        '请使用完整的 mini-Omaha 服务器安装测试程式，\n'
        '不要直接从 USB 碟开机执行。\n\n'
        '若想除错或执行部份测试，请直接按下对应热键。')

_LABEL_FONT = pango.FontDescription('courier new condensed 24')


class PressSpaceTask(task.FactoryTask):
    def __init__(self, pop_up_message):
        self.pop_up_message = pop_up_message

    def start(self):
        self.add_widget(
                ui.make_label(self.pop_up_message, font=ui.LABEL_LARGE_FONT))
        self.connect_window('key-press-event', self.window_key_press)

    def window_key_press(self, window, event):
        if event.keyval == gtk.keysyms.space:
            self.stop()
        else:
            factory.log('PressSpaceTask: non-space hit: %d' % event.keyval)
        return True

class ExternalPowerTask(task.FactoryTask):

    AC_CONNECTED = 1
    AC_DISCONNECTED = 2
    AC_CHECK_PERIOD = 500

    def start(self):
        self.active = True
        widget = ui.make_label(_MSG_TASK_POWER, font=ui.LABEL_LARGE_FONT)
        self.add_widget(widget)
        self.add_timeout(self.AC_CHECK_PERIOD, self.check_event, widget)

    def stop(self):
        self.active = False

    def check_event(self, label):
        if not self.active:
            return True
        state = self.get_external_power_state()
        if state == self.AC_CONNECTED:
            self.stop()
        return True

    def get_external_power_state(self):
        for type_file in glob.glob('/sys/class/power_supply/*/type'):
            type_value = utils.read_one_line(type_file).strip()
            if type_value == 'Mains':
                status_file = os.path.join(os.path.dirname(type_file), 'online')
                try:
                    status = int(utils.read_one_line(status_file).strip())
                except ValueError as details:
                    raise ValueError('Invalid external power state in %s: %s' %
                                     (status_file, details))
                if status == 0:
                    return self.AC_DISCONNECTED
                elif status == 1:
                    return self.AC_CONNECTED
                else:
                    raise ValueError('Invalid external power state "%s" in %s' %
                                     (status, status_file))
        raise IOError('Unable to determine external power state.')


class ShopFloorTask(task.FactoryTask):
    def __init__(self, server_url, serial_number):
        self.server_url = server_url or shopfloor.detect_default_server_url()
        self.serial_number = serial_number

    def start(self):
        # Many developers will try to run factory test image directly without
        # mini-omaha server, so we should either alert and fail, or ask for
        # server address.
        if not self.server_url:
            self.add_widget(ui.make_label(_MSG_NO_SHOP_FLOOR_SERVER_URL,
                                          fg=ui.RED))
            return

        shopfloor.set_server_url(self.server_url)

        # If no partner-specific serial number, pop-up a make_input window.
        if self.serial_number is None:
            self.add_widget(ui.make_input_window(
                    prompt=_MSG_TASK_SERIAL,
                    on_validate=self.validate_serial_number,
                    on_complete=self.complete_serial_task))
        else:
            # Use partner-specific serial number.
            if self.validate_serial_number(self.serial_number):
                self.complete_serial_task(self.serial_number)

    def validate_serial_number(self, serial):
        # This is a callback function for widgets created by make_input_window.
        # When the input is not valid (or temporary network failure), either
        # return False or raise a ValueError with message to be displayed in
        # bottom status line of input window.
        try:
            # All exceptions
            shopfloor.check_serial_number(serial.strip())
            return True
        except shopfloor.ServerFault as e:
            raise ui.InputError("Server error:\n%s" % e)
        except ValueError as e:
            logging.exception("ValueError:")
            raise ui.InputError(e.message)
        except socket.gaierror as e:
            raise ui.InputError("Network failure (address error).")
        except socket.error as e:
            raise ui.InputError("Network failure:\n%s" % e[1])
        except:
            logging.exception("UnknownException:")
            raise ui.InputError(sys.exc_info()[1])
        return False

    def complete_serial_task(self, serial):
        serial = serial.strip()
        EventLog.ForAutoTest().Log('mlb_serial_number',
                                   serial_number=serial)
        factory.log('Serial number: %s' % serial)
        shopfloor.set_serial_number(serial)

        EventClient().post_event(Event(Event.Type.UPDATE_SYSTEM_INFO))
        self.stop()
        return True


class factory_Start(test.test):
    version = 2

    def run_once(self,
                 press_to_continue=True,
                 require_external_power=False,
                 require_shop_floor=None,
                 shop_floor_server_url=None,
                 serial_number=None,
                 press_to_start_shop_floor=False):
        factory.log('%s run_once' % self.__class__)

        self._task_list = []

        # Reset shop floor data only if require_shop_floor is explicitly
        # defined, for test lists using factory_Start multiple times between
        # groups (ex, to prompt for space or check power adapter).

        if require_shop_floor is not None:
            shopfloor.reset()
            shopfloor.set_enabled(require_shop_floor)
            if press_to_start_shop_floor:
                self._task_list.append(PressSpaceTask(_MSG_TASK_SHOP_FLOOR))

        if require_shop_floor:
            self._task_list.append(
                ShopFloorTask(shop_floor_server_url, serial_number))
        if require_external_power:
            self._task_list.append(ExternalPowerTask())
        if press_to_continue:
            self._task_list.append(PressSpaceTask(_MSG_TASK_SPACE))

        if self._task_list:
            task.run_factory_tasks(self.job, self._task_list)

        factory.log('%s run_once finished' % repr(self.__class__))