• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2022 The Chromium Authors
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5import pathlib
6
7from crossbench.probes.probe import Probe
8from crossbench.probes.results import (BrowserProbeResult,
9                                       DuplicateProbeResult, EmptyProbeResult,
10                                       LocalProbeResult, ProbeResultDict)
11from crossbench.runner.run import Run
12from tests import test_helper
13from tests.crossbench.base import (BaseCrossbenchTestCase,
14                                   CrossbenchFakeFsTestCase)
15
16
17class ProbeResultTestCase(CrossbenchFakeFsTestCase):
18
19  def test_is_empty(self):
20    empty = EmptyProbeResult()
21    self.assertTrue(empty.is_empty)
22    self.assertFalse(empty)
23    combined = empty.merge(EmptyProbeResult())
24    self.assertTrue(combined.is_empty)
25    url = LocalProbeResult(url=("http://test.com",))
26    self.assertFalse(url.is_empty)
27    self.assertTrue(url)
28    combined = empty.merge(url)
29    self.assertFalse(combined.is_empty)
30    self.assertTrue(combined)
31
32  def test_equal_empty(self):
33    empty = EmptyProbeResult()
34    self.assertEqual(empty, empty)
35    self.assertEqual(EmptyProbeResult(), EmptyProbeResult())
36    local_empty = LocalProbeResult()
37    self.assertEqual(local_empty, EmptyProbeResult())
38    self.assertEqual(local_empty, local_empty)
39    self.assertNotEqual(LocalProbeResult(), None)
40    self.assertNotEqual(None, LocalProbeResult())
41
42  def test_is_remote(self):
43    empty = EmptyProbeResult()
44    self.assertFalse(empty.is_remote)
45    local_empty = LocalProbeResult()
46    self.assertFalse(local_empty.is_remote)
47
48  def test_equal_single(self):
49    url = "http://test.com"
50    self.assertEqual(LocalProbeResult(url=(url,)), LocalProbeResult(url=(url,)))
51    url_b = "http://foo.test.com"
52    self.assertNotEqual(
53        LocalProbeResult(url=(url,)), LocalProbeResult(url=(url_b,)))
54
55
56  def test_invalid_files(self):
57    with self.assertRaises(ValueError):
58      LocalProbeResult(file=[pathlib.Path("foo.json")])
59    with self.assertRaises(ValueError):
60      LocalProbeResult(file=[pathlib.Path("foo.csv")])
61    with self.assertRaises(ValueError):
62      LocalProbeResult(json=[pathlib.Path("foo.csv")])
63    with self.assertRaises(ValueError):
64      LocalProbeResult(csv=[pathlib.Path("foo.json")])
65
66  def test_inexistent_files(self):
67    with self.assertRaises(ValueError):
68      LocalProbeResult(file=[pathlib.Path("not_there.txt")])
69    with self.assertRaises(ValueError):
70      LocalProbeResult(csv=[pathlib.Path("not_there.csv")])
71    with self.assertRaises(ValueError):
72      LocalProbeResult(json=[pathlib.Path("not_there.json")])
73
74  def test_result_url(self):
75    url = "http://foo.bar.com"
76    with self.assertRaises(DuplicateProbeResult):
77      _ = LocalProbeResult(url=[url, url])
78    result = LocalProbeResult(url=[url])
79    self.assertFalse(result.is_empty)
80    self.assertEqual(result.url, url)
81    self.assertListEqual(result.url_list, [url])
82    self.assertListEqual(list(result.all_files()), [])
83    failed = None
84    with self.assertRaises(ValueError):
85      failed = result.file
86    with self.assertRaises(ValueError):
87      failed = result.json
88    with self.assertRaises(ValueError):
89      failed = result.csv
90    self.assertIsNone(failed)
91    json_data = result.to_json()
92    self.assertDictEqual(json_data, {"url": (url,)})
93
94  def test_result_any_file(self):
95    path = self.create_file("result.txt")
96    with self.assertRaises(DuplicateProbeResult):
97      _ = LocalProbeResult(file=[path, path])
98    result = LocalProbeResult(file=[path])
99    self.assertFalse(result.is_empty)
100    self.assertNotEqual(
101        result, LocalProbeResult(file=[
102            self.create_file("result2.txt"),
103        ]))
104    self.assertEqual(result.file, path)
105    self.assertListEqual(result.file_list, [path])
106    self.assertListEqual(list(result.all_files()), [path])
107    failed = None
108    with self.assertRaises(ValueError):
109      failed = result.url
110    with self.assertRaises(ValueError):
111      failed = result.json
112    with self.assertRaises(ValueError):
113      failed = result.csv
114    self.assertIsNone(failed)
115
116  def test_result_csv(self):
117    path = self.create_file("result.csv")
118    with self.assertRaises(DuplicateProbeResult):
119      _ = LocalProbeResult(csv=[path, path])
120    result = LocalProbeResult(csv=[path])
121    self.assertFalse(result.is_empty)
122    self.assertNotEqual(
123        result, LocalProbeResult(csv=[
124            self.create_file("result2.csv"),
125        ]))
126    self.assertEqual(result.csv, path)
127    self.assertEqual(result.get("csv"), path)
128    self.assertListEqual(result.csv_list, [path])
129    self.assertListEqual(result.get_all("csv"), [path])
130    self.assertListEqual(list(result.all_files()), [path])
131    self.assertEqual(result.file, path)
132    failed = None
133    with self.assertRaises(ValueError):
134      failed = result.url
135    with self.assertRaises(ValueError):
136      failed = result.json
137    self.assertIsNone(failed)
138
139  def test_result_json(self):
140    path = self.create_file("result.json")
141    with self.assertRaises(DuplicateProbeResult):
142      _ = LocalProbeResult(json=[path, path])
143    with self.assertRaises(DuplicateProbeResult):
144      _ = LocalProbeResult(file=[path, path])
145    result = LocalProbeResult(json=[path])
146    self.assertFalse(result.is_empty)
147    self.assertNotEqual(
148        result, LocalProbeResult(json=[
149            self.create_file("result2.json"),
150        ]))
151    self.assertEqual(result.json, path)
152    self.assertEqual(result.get("json"), path)
153    self.assertListEqual(result.json_list, [path])
154    self.assertListEqual(result.get_all("json"), [path])
155    self.assertListEqual(list(result.all_files()), [path])
156    self.assertEqual(result.file, path)
157    failed = None
158    with self.assertRaises(ValueError):
159      failed = result.url
160    with self.assertRaises(ValueError):
161      failed = result.csv
162    self.assertIsNone(failed)
163
164  def test_multiple_urls(self):
165    url1 = "http://one.com"
166    url2 = "http://two.com"
167    result = LocalProbeResult(url=(url1, url2))
168    self.assertFalse(result.is_empty)
169    self.assertFalse(result.has_files)
170    with self.assertRaises(ValueError):
171      _ = result.file
172    with self.assertRaises(ValueError):
173      _ = result.url
174    self.assertSequenceEqual(result.url_list, (url1, url2))
175
176  def test_multiple_files(self):
177    json1 = self.create_file("result_1.json")
178    json2 = self.create_file("result_2.json")
179    zip1 = self.create_file("result.zip")
180    result = LocalProbeResult(file=(json1, json2, zip1))
181    self.assertFalse(result.is_empty)
182    self.assertTrue(result.has_files)
183    with self.assertRaises(ValueError):
184      _ = result.file
185    with self.assertRaises(ValueError):
186      _ = result.json
187    self.assertSequenceEqual(result.file_list, (json1, json2, zip1))
188    self.assertSequenceEqual(result.json_list, (json1, json2))
189    self.assertSequenceEqual(result.get_all("json"), (json1, json2))
190    self.assertEqual(result.get("zip"), zip1)
191    self.assertEqual(result.get_all("zip"), [zip1])
192    with self.assertRaises(ValueError):
193      _ = result.get("other")
194
195  def test_merge(self):
196    file = self.create_file("result.custom")
197    json = self.create_file("result.json")
198    csv = self.create_file("result.csv")
199    url = "http://foo.bar.com"
200    trace = self.create_file("trace.pb")
201
202    result = LocalProbeResult(
203        url=(url,), file=(file,), json=(json,), csv=(csv,), trace=(trace,))
204    self.assertFalse(result.is_empty)
205    self.assertListEqual(list(result.all_files()), [file, json, csv, trace])
206    self.assertListEqual(result.url_list, [url])
207    self.assertListEqual(result.trace_list, [trace])
208
209    merged = result.merge(EmptyProbeResult())
210    self.assertFalse(merged.is_empty)
211    self.assertListEqual(list(merged.all_files()), [file, json, csv, trace])
212    self.assertListEqual(merged.url_list, [url])
213    self.assertListEqual(merged.trace_list, [trace])
214
215    file_2 = self.create_file("result.2.custom")
216    json_2 = self.create_file("result.2.json")
217    csv_2 = self.create_file("result.2.csv")
218    url_2 = "http://foo.bar.com/2"
219    trace_2 = self.create_file("trace.2.pb")
220    other = LocalProbeResult(
221        url=(url_2,),
222        file=(file_2,),
223        json=(json_2,),
224        csv=(csv_2,),
225        trace=(trace_2,))
226    merged = result.merge(other)
227    self.assertFalse(merged.is_empty)
228    self.assertListEqual(
229        list(merged.all_files()),
230        [file, file_2, json, json_2, csv, csv_2, trace, trace_2])
231    self.assertListEqual(merged.url_list, [url, url_2])
232    # result is unchanged:
233    self.assertFalse(result.is_empty)
234    self.assertListEqual(list(result.all_files()), [file, json, csv, trace])
235    self.assertListEqual(result.url_list, [url])
236    self.assertListEqual(result.trace_list, [trace])
237    # other is unchanged:
238    self.assertFalse(other.is_empty)
239    self.assertListEqual(
240        list(other.all_files()), [file_2, json_2, csv_2, trace_2])
241    self.assertListEqual(other.url_list, [url_2])
242    self.assertListEqual(other.trace_list, [trace_2])
243
244  def test_merge_duplicate_files(self):
245    path = self.create_file("result.custom")
246    result_1 = LocalProbeResult(file=(path,))
247    result_2 = LocalProbeResult(file=(path,))
248    with self.assertRaises(DuplicateProbeResult):
249      result_1.merge(result_1)
250    with self.assertRaises(DuplicateProbeResult):
251      result_1.merge(result_2)
252    with self.assertRaises(DuplicateProbeResult):
253      result_2.merge(result_1)
254
255  def test_traces_are_files(self):
256    trace = self.create_file("trace.pb")
257
258    result = LocalProbeResult(trace=(trace,))
259    self.assertFalse(result.is_empty)
260    self.assertListEqual(list(result.all_files()), [trace])
261    self.assertListEqual(list(result.trace_list), [trace])
262
263  def test_traces_can_be_duplicate(self):
264    trace = self.create_file("trace.pb")
265
266    result = LocalProbeResult(trace=(trace,), file=(trace,))
267    self.assertFalse(result.is_empty)
268    self.assertListEqual(list(result.all_files()), [trace])
269    self.assertListEqual(list(result.trace_list), [trace])
270
271
272class MockRun:
273
274  def __init__(self, browser, browser_platform):
275    self.brower = browser
276    self.browser_platform = browser_platform
277    self.is_remote = False
278
279
280class BrowserProbeResultTestCase(BaseCrossbenchTestCase):
281
282  def setUp(self) -> None:
283    super().setUp()
284    self.run = MockRun(self.browsers[0], self.platform)
285
286  def test_empty(self):
287    result = BrowserProbeResult(self.run)
288    self.assertTrue(result.is_empty)
289    self.assertFalse(result.is_remote)
290
291  def test_remote_empty(self):
292    self.run.is_remote = True
293    result = BrowserProbeResult(self.run)
294    self.assertTrue(result.is_empty)
295    self.assertEqual(result, EmptyProbeResult())
296    self.assertEqual(result, LocalProbeResult())
297
298  def test_remote_only_urls(self):
299    self.run.is_remote = True
300    result = BrowserProbeResult(self.run, url=("http://test.com",))
301    self.assertFalse(result.is_empty)
302    self.assertNotEqual(result, EmptyProbeResult())
303    self.assertEqual(result, LocalProbeResult(url=("http://test.com",)))
304    self.assertNotEqual(result, LocalProbeResult(url=("http://test.com/bar",)))
305
306  def test_local_browser_equal_local_run(self):
307    url = "http://foo.bar.com"
308    file = self.create_file("results/file.any")
309    json = self.create_file("results/file.json")
310    csv = self.create_file("results/file.csv")
311    local = LocalProbeResult((url,), (file, json, csv))
312    browser = BrowserProbeResult(self.run, (url,), (file, json, csv))
313    self.assertEqual(local, browser)
314    local_kwargs = LocalProbeResult(
315        url=(url,), file=(file,), json=(json,), csv=(csv,))
316    browser_kwargs = BrowserProbeResult(
317        self.run, url=(url,), file=(file,), json=(json,), csv=(csv,))
318    self.assertEqual(local, browser)
319    self.assertEqual(local, browser_kwargs)
320    self.assertEqual(local_kwargs, browser)
321    self.assertEqual(local_kwargs, browser_kwargs)
322
323  def test_copy_remote_files(self):
324    # pylint: disable=attribute-defined-outside-init
325    out_dir_local = pathlib.Path("local/results")
326    out_dir_remote = pathlib.Path("remote/results")
327    out_dir_local.mkdir(parents=True)
328    out_dir_remote.mkdir(parents=True)
329    self.assertTrue(out_dir_local.is_dir())
330    self.assertTrue(out_dir_remote.is_dir())
331    remote_txt = self.create_file(
332        out_dir_remote / "result.txt", contents="a1 b2 c3")
333    remote_json = self.create_file(
334        out_dir_remote / "result.json", contents="[]")
335    self.run.is_remote = True
336    self.run.out_dir = out_dir_local
337    self.run.browser_tmp_dir = out_dir_remote
338    with self.assertRaises(DuplicateProbeResult):
339      _ = BrowserProbeResult(self.run, file=(remote_txt, remote_txt))
340    with self.assertRaises(DuplicateProbeResult):
341      _ = BrowserProbeResult(self.run, json=(remote_json, remote_json))
342    result = BrowserProbeResult(
343        self.run, file=(remote_txt,), json=(remote_json,))
344    self.assertTrue(remote_txt.is_file())
345    with self.assertRaises(ValueError):
346      _ = result.file
347    self.assertTrue(result.has_files)
348    self.assertNotEqual(remote_json, result.json)
349    self.assertTrue(result.json.is_file())
350    self.assertEqual(result.json.read_text(), "[]")
351    self.assertNotEqual(remote_txt, result.get("txt"))
352    self.assertEqual(result.get("txt").read_text(), "a1 b2 c3")
353
354
355class MockProbe(Probe):
356  """
357  Probe DOC Text
358  """
359  NAME = "mock-probe"
360
361  def get_context(self, run: Run):
362    pass
363
364
365class ProbeResultDictTestCase(CrossbenchFakeFsTestCase):
366
367  def setUp(self) -> None:
368    super().setUp()
369    self.result_location = pathlib.Path("test/out/results")
370    self.result_dict = ProbeResultDict(self.result_location)
371    self.local_result = LocalProbeResult()
372
373  def test_create_empty(self):
374    self.assertEqual(len(self.result_dict), 0)
375    self.assertFalse(self.result_dict)
376    self.assertNotIn(MockProbe(), self.result_dict)
377
378  def test_get_item(self):
379    probe = MockProbe()
380    with self.assertRaises(KeyError):
381      _ = self.result_dict[probe]
382    self.result_dict[probe] = self.local_result
383    self.assertIs(self.result_dict[probe], self.local_result)
384
385  def test_get(self):
386    probe = MockProbe()
387    self.assertIsNone(self.result_dict.get(probe))
388    self.assertEqual(self.result_dict.get(probe, 1234), 1234)
389    self.result_dict[probe] = self.local_result
390    self.assertIs(self.result_dict.get(probe), self.local_result)
391
392  def test_to_json_empty(self):
393    probe = MockProbe()
394    self.result_dict[probe] = self.local_result
395    json = self.result_dict.to_json()
396    self.assertDictEqual(json, {MockProbe.NAME: None})
397
398  def test_to_json_result(self):
399    csv = self.create_file("result.csv")
400    txt = self.create_file("result.txt")
401    self.local_result = LocalProbeResult(csv=(csv,), txt=(txt,))
402    probe = MockProbe()
403    self.result_dict[probe] = self.local_result
404    json = self.result_dict.to_json()
405    self.assertDictEqual(
406        json, {MockProbe.NAME: {
407            "csv": [str(csv)],
408            "txt": [str(txt)]
409        }})
410
411
412if __name__ == "__main__":
413  test_helper.run_pytest(__file__)
414