/
test_utils.py
227 lines (183 loc) · 7.76 KB
/
test_utils.py
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
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
# Copyright 2016 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.
"""Common utilities used in unit tests, within this directory."""
import dependency_graph
import devtools_monitor
import loading_trace
import page_track
import request_track
import tracing
import user_satisfied_lens
class FakeRequestTrack(devtools_monitor.Track):
def __init__(self, events):
super(FakeRequestTrack, self).__init__(None)
self._events = events
for e in self._events:
e.timing.request_time = e.timestamp
def Handle(self, _method, _msg):
assert False # Should never be called.
def GetEvents(self):
return self._events
def ToJsonDict(self):
cls = request_track.RequestTrack
return {cls._EVENTS_KEY: [
rq.ToJsonDict() for rq in self.GetEvents()],
cls._METADATA_KEY: {
cls._DUPLICATES_KEY: 0,
cls._INCONSISTENT_INITIATORS_KEY: 0}}
class FakePageTrack(devtools_monitor.Track):
def __init__(self, events):
super(FakePageTrack, self).__init__(None)
self._events = events
def Handle(self, _method, _msg):
assert False # Should never be called.
def GetEvents(self):
return self._events
def GetMainFrameId(self):
event = self._events[0]
# Make sure our laziness is not an issue here.
assert event['method'] == page_track.PageTrack.FRAME_STARTED_LOADING
return event['frame_id']
def ToJsonDict(self):
return {'events': [event for event in self._events]}
def MakeRequestWithTiming(
url, source_url, timing_dict, magic_content_type=False,
initiator_type='other'):
"""Make a dependent request.
Args:
url: a url, or number which will be used as a url.
source_url: a url or number which will be used as the source (initiating)
url. If the source url is not present, then url will be a root. The
convention in tests is to use a source_url of 'null' in this case.
timing_dict: (dict) Suitable to be passed to request_track.Timing().
initiator_type: the initiator type to use.
Returns:
A request_track.Request.
"""
assert initiator_type in ('other', 'parser')
timing = request_track.Timing.FromDevToolsDict(timing_dict)
rq = request_track.Request.FromJsonDict({
'timestamp': timing.request_time,
'request_id': str(MakeRequestWithTiming._next_request_id),
'url': 'http://' + str(url),
'initiator': {'type': initiator_type, 'url': 'http://' + str(source_url)},
'response_headers': {'Content-Type':
'null' if not magic_content_type
else 'magic-debug-content' },
'timing': timing.ToJsonDict()
})
MakeRequestWithTiming._next_request_id += 1
return rq
MakeRequestWithTiming._next_request_id = 0
def MakeRequest(
url, source_url, start_time=None, headers_time=None, end_time=None,
magic_content_type=False, initiator_type='other'):
"""Make a dependent request.
Args:
url: a url, or number which will be used as a url.
source_url: a url or number which will be used as the source (initiating)
url. If the source url is not present, then url will be a root. The
convention in tests is to use a source_url of 'null' in this case.
start_time: The request start time in milliseconds. If None, this is set to
the current request id in seconds. If None, the two other time parameters
below must also be None.
headers_time: The timestamp when resource headers were received, or None.
end_time: The timestamp when the resource was finished, or None.
magic_content_type (bool): if true, set a magic content type that makes url
always be detected as a valid source and destination request.
initiator_type: the initiator type to use.
Returns:
A request_track.Request.
"""
assert ((start_time is None and
headers_time is None and
end_time is None) or
(start_time is not None and
headers_time is not None and
end_time is not None)), \
'Need no time specified or all times specified'
if start_time is None:
# Use the request id in seconds for timestamps. This guarantees increasing
# times which makes request dependencies behave as expected.
start_time = headers_time = end_time = (
MakeRequestWithTiming._next_request_id * 1000)
timing_dict = {
# connectEnd should be ignored.
'connectEnd': (end_time - start_time) / 2,
'receiveHeadersEnd': headers_time - start_time,
'loadingFinished': end_time - start_time,
'requestTime': start_time / 1000.0}
return MakeRequestWithTiming(
url, source_url, timing_dict, magic_content_type, initiator_type)
def LoadingTraceFromEvents(requests, page_events=None, trace_events=None):
"""Returns a LoadingTrace instance from a list of requests and page events."""
request = FakeRequestTrack(requests)
page_event_track = FakePageTrack(page_events if page_events else [])
if trace_events is not None:
tracing_track = tracing.TracingTrack(None)
tracing_track.Handle('Tracing.dataCollected',
{'params': {'value': [e for e in trace_events]}})
else:
tracing_track = None
return loading_trace.LoadingTrace(
None, None, page_event_track, request, tracing_track)
class SimpleLens(object):
"""A simple replacement for RequestDependencyLens.
Uses only the initiator url of a request for determining a dependency.
"""
def __init__(self, trace):
self._trace = trace
def GetRequestDependencies(self):
url_to_rq = {}
deps = []
for rq in self._trace.request_track.GetEvents():
assert rq.url not in url_to_rq
url_to_rq[rq.url] = rq
for rq in self._trace.request_track.GetEvents():
initiating_url = rq.initiator['url']
if initiating_url in url_to_rq:
deps.append((url_to_rq[initiating_url], rq, rq.initiator['type']))
return deps
class TestDependencyGraph(dependency_graph.RequestDependencyGraph):
"""A dependency graph created from requests using a simple lens."""
def __init__(self, requests):
lens = SimpleLens(LoadingTraceFromEvents(requests))
super(TestDependencyGraph, self).__init__(requests, lens)
class MockConnection(object):
"""Mock out connection for testing.
Use Expect* for requests expecting a repsonse. SyncRequestNoResponse puts
requests into no_response_requests_seen.
TODO(mattcary): use a standard mock system (the back-ported python3
unittest.mock? devil.utils.mock_calls?)
"""
def __init__(self, test_case):
# List of (method, params) tuples.
self.no_response_requests_seen = []
self._test_case = test_case
self._expected_responses = {}
def ExpectSyncRequest(self, response, method, params=None):
"""Test method when the connection is expected to make a SyncRequest.
Args:
response: (dict) the response to generate.
method: (str) the expected method in the call.
params: (dict) the expected params in the call.
"""
self._expected_responses.setdefault(method, []).append((params, response))
def AllExpectationsUsed(self):
"""Returns true when all expectations where used."""
return not self._expected_responses
def SyncRequestNoResponse(self, method, params):
"""Mocked method."""
self.no_response_requests_seen.append((method, params))
def SyncRequest(self, method, params=None):
"""Mocked method."""
expected_params, response = self._expected_responses[method].pop(0)
if not self._expected_responses[method]:
del self._expected_responses[method]
self._test_case.assertEqual(expected_params, params)
return response
class MockUserSatisfiedLens(user_satisfied_lens._UserSatisfiedLens):
def _CalculateTimes(self, _):
self._satisfied_msec = float('inf')
self._event_msec = float('inf')