harnesses/host_controller/tfc/command_attempt.py

#
# Copyright (C) 2017 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

import time

from host_controller.tfc import api_message


class EventType(object):
    """The types of command events."""
    ALLOCATION_FAILED = "AllocationFailed"
    CONFIGURATION_ERROR = "ConfigurationError"
    EXECUTE_FAILED = "ExecuteFailed"
    FETCH_FAILED = "FetchFailed"
    INVOCATION_COMPLETED = "InvocationCompleted"
    INVOCATION_STARTED = "InvocationStarted"
    TEST_RUN_IN_PROGRESS = "TestRunInProgress"


class CommandAttempt(api_message.ApiMessage):
    """The command attempt defined by TFC API.

    Attributes:
        _COMMAND_EVENT: The parameters of command_events.submit.
        _COMMAND_EVENT_DATA: The fields in "data" parameter of command_events.
        _LIST_ATTEMPT: The fields returned by commandAttempts.list.
    """
    _COMMAND_EVENT = {
            "attempt_id",
            "data",
            "device_serial",
            "hostname",
            "task_id",
            "time",
            "type"}
    _COMMAND_EVENT_DATA = {
            "error",
            "failed_test_count",
            "summary",
            "test_run_name",
            "total_test_count"}
    _LIST_ATTEMPT = {
            "attempt_id",
            "command_id",
            "create_time",
            "end_time",
            "error",
            "device_serial",
            "failed_test_count",
            "hostname",
            "request_id",
            "start_time",
            "state",
            "status",
            "summary",
            "task_id",
            "total_test_count",
            "update_time"}

    def __init__(self, task_id, attempt_id, hostname, device_serial, **kwargs):
        """Initializes the attributes.

        Args:
            task_id: A string, the task id assigned by the server.
            attempt_id: A string or UUID, the attempt id generated by the host.
            hostname: The name of the TradeFed host.
            device_serial: The serial number of the device.
            **kwargs: The optional attributes.
        """
        super(CommandAttempt, self).__init__(self._LIST_ATTEMPT,
                                             task_id=task_id,
                                             attempt_id=str(attempt_id),
                                             hostname=hostname,
                                             device_serial=device_serial,
                                             **kwargs)

    def CreateCommandEvent(self, event_type, error=None, event_time=None):
        """Creates an event defined by command_events.submit.

        Args:
            event_type: A string in EventType.
            error: A string, the error message for *Failed, *Error, and
                   *Completed events.
            event_time: A float, Unix timestamp of the event in seconds.

        Returns:
            A JSON object.
        """
        obj = self.ToJson(self._COMMAND_EVENT)
        obj["type"] = event_type
        obj["time"] = int(event_time if event_time is not None else time.time())
        data_obj = self.ToJson(self._COMMAND_EVENT_DATA)
        if error is not None:
            data_obj["error"] = error
        if data_obj:
            obj["data"] = data_obj
        return obj

    def CreateInvocationCompletedEvent(self,
                                       summary,
                                       total_test_count,
                                       failed_test_count,
                                       error=None,
                                       event_time=None):
        """Creates an InvocationCompleted event.

        Args:
            summary: A string, the result of the command.
            total_test_count: Number of test cases.
            failed_test_count: Number of failed test cases.
            error: A string, the error message.
            event_time: A float, Unix timestamp of the event in seconds.

        Returns:
            A JSON object.
        """
        obj = self.CreateCommandEvent(EventType.INVOCATION_COMPLETED,
                                      error, event_time)
        if "data" not in obj:
            obj["data"] = dict()
        obj["data"].update({"summary": summary,
                            "total_test_count": total_test_count,
                            "failed_test_count": failed_test_count})
        return obj