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
|
#!/usr/bin/env python
import argparse
import datetime
import decimal
import json
import logging
import os
import requests
from collections import OrderedDict
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument('-r', '--result-file', dest='result_file',
required=True, default='./result.json',
help='Specify test result file.')
parser.add_argument('-a', '--attachment', dest='attachment',
action='append', help='Specify attachment file.')
parser.add_argument('-t', '--team', dest='team', required=True,
help='Team identifier. Defaults to "erp"')
parser.add_argument('-p', '--project', dest='project',
help='Project identifier. Defaults to the name of the Linux distribution.')
parser.add_argument('-b', '--build', dest='build', required=True,
help='Build identifier.')
parser.add_argument('-e', '--test-env', dest='test_env',
help='Environment identifier. Defaults to board name.')
parser.add_argument('-u', '--url', dest='url',
default='https://qa-reports.linaro.org',
help='Dashboard URL. Defaults to https://qa-reports.linaro.org.')
parser.add_argument('-v', '--verbose', action='store_true', dest='verbose',
default=True, help='Set log level.')
args = parser.parse_args()
return args
def squad_result(results):
squad_tests = OrderedDict()
squad_metrics = OrderedDict()
for result in results:
for metric in result['metrics']:
key = '%s/%s' % (result['name'], metric['test_case_id'])
if not metric['measurement']:
# Collect pass/fail test results.
squad_tests[key] = metric['result']
else:
# Collect performance test results.
try:
measurement = decimal.Decimal(metric['measurement'])
squad_metrics[key] = float(measurement)
except decimal.InvalidOperation:
logger.info('Invalid measurement: %s' % metric['measurement'])
logger.info('Skipped adding: %s' % metric)
assert squad_tests or squad_metrics, 'No valid result found!'
return (squad_tests, squad_metrics)
def squad_metadata(results):
test_plan = list(set(i['test_plan'] for i in results))
test_version = list(set(i['version'] for i in results))
assert len(test_plan) == 1, 'More then one test plan found!'
assert len(test_version) == 1, 'More then one test version found!'
squad_metadata = OrderedDict()
test_plan = test_plan[0]
test_plan_name = os.path.splitext(os.path.basename(test_plan))[0]
squad_metadata['job_id'] = '{}_{}'.format(test_plan_name, datetime.datetime.utcnow().isoformat())
squad_metadata['test_plan'] = test_plan
squad_metadata['test_version'] = test_version[0]
for key, value in results[-1]['environment'].items():
if key != 'packages':
squad_metadata[key] = value
return squad_metadata
def main():
auth_token = os.environ.get("SQUAD_AUTH_TOKEN")
assert auth_token, "SQUAD_AUTH_TOKEN not provided in environment"
with open(args.result_file, 'r') as f:
results = json.load(f)
metadata = squad_metadata(results)
tests, metrics = squad_result(results)
files = [('metadata', json.dumps(metadata)),
('tests', json.dumps(tests)),
('metrics', json.dumps(metrics)),
('attachment', open(args.result_file, 'rb'))]
if args.attachment is not None:
for item in args.attachment:
if os.path.exists(item):
logger.info('Adding {} to attachment list...'.format(item))
files.append(tuple(['attachment', open(item, 'rb')]))
else:
logger.info('Attachment %s Not found' % args.attachment)
logger.info('Skipped uploading %s' % args.attachment)
logger.debug('Data to post: %s' % files)
project = args.project or metadata['linux_distribution']
test_env = args.test_env or metadata['board_name']
url = '{}/api/submit/{}/{}/{}/{}'.format(args.url, args.team, project, args.build, test_env)
logger.info('Posting to {}'.format(url))
headers = {'Auth-Token': auth_token}
r = requests.post(url, headers=headers, files=files)
print(r.text)
if __name__ == "__main__":
args = parse_args()
logger = logging.getLogger('post-to-squad')
logger.setLevel(logging.INFO)
if args.verbose:
logger.setLevel(logging.DEBUG)
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
ch.setFormatter(formatter)
logger.addHandler(ch)
main()
|
