1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
#
# 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
|
