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
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
|
# Copyright 2020 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import base64
import json
import os
import shutil
import sys
import common
BLINK_TOOLS_DIR = os.path.join(common.SRC_DIR, 'third_party', 'blink', 'tools')
WEB_TESTS_DIR = os.path.join(BLINK_TOOLS_DIR, os.pardir, 'web_tests')
if BLINK_TOOLS_DIR not in sys.path:
sys.path.append(BLINK_TOOLS_DIR)
from blinkpy.common.host import Host
from blinkpy.web_tests.models import test_failures
class BaseWptScriptAdapter(common.BaseIsolatedScriptArgsAdapter):
"""The base class for script adapters that use wptrunner to execute web
platform tests. This contains any code shared between these scripts, such
as integrating output with the results viewer. Subclasses contain other
(usually platform-specific) logic."""
def __init__(self):
super(BaseWptScriptAdapter, self).__init__()
host = Host()
self.port = host.port_factory.get()
def generate_test_output_args(self, output):
return ['--log-chromium', output]
def generate_sharding_args(self, total_shards, shard_index):
return ['--total-chunks=%d' % total_shards,
# shard_index is 0-based but WPT's this-chunk to be 1-based
'--this-chunk=%d' % (shard_index + 1)]
def do_post_test_run_tasks(self):
# Move json results into layout-test-results directory
results_dir = os.path.dirname(self.options.isolated_script_test_output)
layout_test_results = os.path.join(results_dir, 'layout-test-results')
if os.path.exists(layout_test_results):
shutil.rmtree(layout_test_results)
os.mkdir(layout_test_results)
# Perform post-processing of wptrunner output
self.process_wptrunner_output()
shutil.copyfile(self.options.isolated_script_test_output,
os.path.join(layout_test_results, 'full_results.json'))
# create full_results_jsonp.js file which is used to
# load results into the results viewer
with open(self.options.isolated_script_test_output, 'r') \
as full_results, \
open(os.path.join(
layout_test_results, 'full_results_jsonp.js'), 'w') \
as json_js:
json_js.write('ADD_FULL_RESULTS(%s);' % full_results.read())
# copy layout test results viewer to layout-test-results directory
shutil.copyfile(
os.path.join(WEB_TESTS_DIR, 'fast', 'harness', 'results.html'),
os.path.join(layout_test_results, 'results.html'))
def process_wptrunner_output(self):
"""Post-process the output generated by wptrunner.
This output contains a single large json file containing the raw content
or artifacts which need to be extracted into their own files and removed
from the json file (to avoid duplication)."""
output_json = json.load(
open(self.options.isolated_script_test_output, "r"))
test_json = output_json["tests"]
results_dir = os.path.dirname(self.options.isolated_script_test_output)
self._process_test_leaves(results_dir, output_json["path_delimiter"],
test_json, "")
# Write output_json back to the same file after modifying it in memory
with open(self.options.isolated_script_test_output, "w") as output_file:
json.dump(output_json, output_file)
def _process_test_leaves(self, results_dir, delim, root_node, path_so_far):
"""Finds and processes each test leaf below the specified root.
This will recursively traverse the trie of results in the json output,
keeping track of the path to each test and identifying leaves by the
presence of certain attributes.
Args:
results_dir: str path to the dir that results are stored
delim: str delimiter to be used for test names
root_node: dict representing the root of the trie we're currently
looking at
path_so_far: str the path to the current root_node in the trie
"""
if "actual" in root_node:
# Found a leaf, process it
if "artifacts" not in root_node:
return
log_artifact = root_node["artifacts"].pop("log", None)
if log_artifact:
artifact_subpath = self._write_log_artifact(
results_dir, path_so_far, log_artifact)
root_node["artifacts"]["actual_text"] = [artifact_subpath]
screenshot_artifact = root_node["artifacts"].pop("screenshots",
None)
if screenshot_artifact:
screenshot_paths_dict = self._write_screenshot_artifact(
results_dir, path_so_far, screenshot_artifact)
for screenshot_key, path in screenshot_paths_dict.items():
root_node["artifacts"][screenshot_key] = [path]
return
# We're not at a leaf node, continue traversing the trie.
for key in root_node:
# Append the key to the path, separated by the delimiter. However if
# the path is empty, skip the delimiter to avoid a leading slash in
# the path.
new_path = path_so_far + delim + key if path_so_far else key
self._process_test_leaves(results_dir, delim, root_node[key],
new_path)
def _write_log_artifact(self, results_dir, test_name, log_artifact):
"""Writes a log artifact to disk.
The log artifact contains all the output of a test. It gets written to
the -actual.txt file for the test.
Args:
results_dir: str path to the directory that results live in
test_name: str name of the test that this artifact is for
log_artifact: list of strings, the log entries for this test from
the json output.
Returns:
string path to the artifact file that the log was written to,
relative to the directory that the original output is located.
"""
log_artifact_sub_path = (
os.path.join("layout-test-results",
self.port.output_filename(
test_name, test_failures.FILENAME_SUFFIX_ACTUAL,
".txt"))
)
log_artifact_full_path = os.path.join(results_dir,
log_artifact_sub_path)
if not os.path.exists(os.path.dirname(log_artifact_full_path)):
os.makedirs(os.path.dirname(log_artifact_full_path))
with open(log_artifact_full_path, "w") as artifact_file:
artifact_file.write("\n".join(log_artifact).encode("utf-8"))
return log_artifact_sub_path
def _write_screenshot_artifact(self, results_dir, test_name,
screenshot_artifact):
"""Write screenshot artifact to disk.
The screenshot artifact is a list of strings, each of which has the
format <url>:<base64-encoded PNG>. Each url-png pair is a screenshot of
either the test, or one of its refs. We can identify which screenshot is
for the test by comparing the url piece to the test name.
Args:
results_dir: str path to the directory that results live in
test:name str name of the test that this artifact is for
screenshot_artifact: list of strings, each being a url-png pair as
described above.
Returns:
A dict mapping the screenshot key (ie: actual, expected) to the
path of the file for that screenshot
"""
result={}
for screenshot_pair in screenshot_artifact:
screenshot_split = screenshot_pair.split(":")
url = screenshot_split[0]
# The url produced by wptrunner will have a leading / which we trim
# away for easier comparison to the test_name below.
if url.startswith("/"):
url = url[1:]
image_bytes = base64.b64decode(screenshot_split[1].strip())
screenshot_key = "expected_image"
file_suffix = test_failures.FILENAME_SUFFIX_EXPECTED
if test_name == url:
screenshot_key = "actual_image"
file_suffix = test_failures.FILENAME_SUFFIX_ACTUAL
screenshot_sub_path = (
os.path.join("layout-test-results",
self.port.output_filename(
test_name, file_suffix, ".png"))
)
result[screenshot_key] = screenshot_sub_path
screenshot_full_path = os.path.join(results_dir,screenshot_sub_path)
if not os.path.exists(os.path.dirname(screenshot_full_path)):
os.makedirs(os.path.dirname(screenshot_full_path))
# Note: we are writing raw bytes to this file
with open(screenshot_full_path, "wb") as artifact_file:
artifact_file.write(image_bytes)
return result
|
