• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2024 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
5from __future__ import annotations
6
7import argparse
8import json
9
10from crossbench import path as pth
11from crossbench.cli.config.network import (NetworkConfig, NetworkSpeedConfig,
12                                           NetworkSpeedPreset, NetworkType)
13from tests import test_helper
14from tests.crossbench.cli.config.base import BaseConfigTestCase
15
16
17class NetworkSpeedConfigTestCase(BaseConfigTestCase):
18
19  def test_parse_invalid(self):
20    for invalid in ("", None, "---"):
21      with self.subTest(invalid=invalid):
22        with self.assertRaises(argparse.ArgumentTypeError):
23          NetworkSpeedConfig.parse(invalid)
24        with self.assertRaises(argparse.ArgumentTypeError):
25          NetworkSpeedConfig.parse_str(str(invalid))
26      with self.assertRaises(argparse.ArgumentTypeError) as cm:
27        NetworkSpeedConfig.parse("not a speed preset value")
28      self.assertIn("choices are", str(cm.exception).lower())
29
30  def test_parse_default(self):
31    config = NetworkSpeedConfig.parse("default")
32    self.assertEqual(config, NetworkSpeedConfig.default())
33
34  def test_default(self):
35    config = NetworkSpeedConfig.default()
36    self.assertIsNone(config.rtt_ms)
37    self.assertIsNone(config.in_kbps)
38    self.assertIsNone(config.out_kbps)
39    self.assertIsNone(config.window)
40
41  def test_parse_speed_preset(self):
42    config = NetworkSpeedConfig.parse("4G")
43    self.assertNotEqual(config, NetworkSpeedConfig.default())
44
45    for preset in NetworkSpeedPreset:  # pytype: disable=missing-parameter
46      config = NetworkSpeedConfig.parse(str(preset))
47      self.assertEqual(config, NetworkSpeedConfig.parse_preset(preset))
48
49  def test_parse_invalid_preset(self):
50    with self.assertRaises(argparse.ArgumentTypeError) as cm:
51      NetworkSpeedConfig.parse("1xx4")
52    self.assertIn("1xx4", str(cm.exception))
53    self.assertIn("Choices are", str(cm.exception))
54
55  def test_parse_dict(self):
56    config = NetworkSpeedConfig.parse({
57        "rtt_ms": 100,
58        "in_kbps": 200,
59        "out_kbps": 300,
60        "window": 400
61    })
62    self.assertIsNone(config.ts_proxy)
63    self.assertEqual(config.rtt_ms, 100)
64    self.assertEqual(config.in_kbps, 200)
65    self.assertEqual(config.out_kbps, 300)
66    self.assertEqual(config.window, 400)
67
68  def test_parse_invalid_dict(self):
69    for int_property in ("rtt_ms", "in_kbps", "out_kbps", "window"):
70      with self.subTest(config_property=int_property):
71        with self.assertRaises(argparse.ArgumentTypeError) as cm:
72          _ = NetworkSpeedConfig.parse({int_property: -100})
73        self.assertIn(int_property, str(cm.exception))
74
75  def test_parse_ts_proxy_path(self):
76    with self.assertRaises(argparse.ArgumentTypeError) as cm:
77      _ = NetworkSpeedConfig.parse({"ts_proxy": "/some/random/path"})
78    self.assertIn("ts_proxy", str(cm.exception))
79    ts_proxy = pth.LocalPath("/python/ts_proxy.py")
80    self.fs.create_file(ts_proxy, st_size=100)
81    config = NetworkSpeedConfig.parse({"ts_proxy": str(ts_proxy)})
82    self.assertEqual(config.ts_proxy, ts_proxy)
83
84
85class NetworkConfigTestCase(BaseConfigTestCase):
86
87  def test_parse_invalid(self):
88    for invalid in ("", None, "---", "something"):
89      with self.assertRaises(argparse.ArgumentTypeError):
90        NetworkConfig.parse(invalid)
91      with self.assertRaises(argparse.ArgumentTypeError):
92        NetworkConfig.parse_str(str(invalid))
93
94  def test_parse_default(self):
95    config = NetworkConfig.parse("default")
96    self.assertEqual(config, NetworkConfig.default())
97
98  def test_default(self):
99    config = NetworkConfig.default()
100    self.assertEqual(config.type, NetworkType.LIVE)
101    self.assertEqual(config.speed, NetworkSpeedConfig.default())
102    config_1 = NetworkConfig.parse({})
103    self.assertEqual(config, config_1)
104    config_2 = NetworkConfig.parse("{}")
105    self.assertEqual(config, config_2)
106
107  def test_parse_replay_archive_invalid(self):
108    path = pth.LocalPath("/foo/bar/wprgo.archive")
109    with self.assertRaises(argparse.ArgumentTypeError) as cm:
110      NetworkConfig.parse(str(path))
111    message = str(cm.exception)
112    self.assertIn("wpr.go archive", message)
113    self.assertIn("exist", message)
114
115    self.fs.create_file(path)
116    with self.assertRaises(argparse.ArgumentTypeError) as cm:
117      NetworkConfig.parse(str(path))
118    message = str(cm.exception)
119    self.assertIn("wpr.go archive", message)
120    self.assertIn("empty", message)
121
122  def test_parse_wprgo_archive(self):
123    path = pth.LocalPath("/foo/bar/wprgo.archive")
124    self.fs.create_file(path, st_size=1024)
125    config = NetworkConfig.parse(str(path))
126    assert isinstance(config, NetworkConfig)
127    self.assertEqual(config.type, NetworkType.WPR)
128    self.assertEqual(config.path, path)
129    self.assertEqual(config.speed, NetworkSpeedConfig.default())
130
131  def test_parse_wprgo_archive_url(self):
132    url = "gs://bucket/wprgo.archive"
133    config = NetworkConfig.parse(url)
134    assert isinstance(config, NetworkConfig)
135    self.assertEqual(config.type, NetworkType.WPR)
136    self.assertEqual(config.url, url)
137    self.assertEqual(config.speed, NetworkSpeedConfig.default())
138
139  def test_invalid_constructor_params(self):
140    with self.assertRaises(argparse.ArgumentTypeError):
141      _ = NetworkConfig(path=pth.LocalPath("foo/bar"))
142    with self.assertRaises(argparse.ArgumentTypeError):
143      _ = NetworkConfig(type=NetworkType.LOCAL, path=None)
144    with self.assertRaises(argparse.ArgumentTypeError):
145      _ = NetworkConfig(type=NetworkType.WPR, path=None)
146
147  def test_parse_speed_preset(self):
148    for preset in NetworkSpeedPreset:  # pytype: disable=missing-parameter
149      config = NetworkConfig.parse_str(preset.value)
150      self.assertEqual(config.speed, NetworkSpeedConfig.parse_preset(preset))
151
152  def test_parse_live_preset(self):
153    live_a = NetworkConfig.parse_live("4G")
154    live_b = NetworkConfig.parse_live(NetworkSpeedConfig.parse("4G"))
155    live_c = NetworkConfig.parse_live(
156        NetworkSpeedConfig.parse(NetworkSpeedPreset.MOBILE_4G))
157    live_d = NetworkConfig.parse_live(NetworkSpeedPreset.MOBILE_4G)
158    self.assertEqual(live_a, live_b)
159    self.assertEqual(live_a, live_c)
160    self.assertEqual(live_a, live_d)
161
162  def test_parse_wpr_invalid(self):
163    dir_path = pth.LocalPath("test/dir")
164    dir_path.mkdir(parents=True)
165    for invalid in ("", "default", "4G", dir_path, str(dir_path)):
166      with self.assertRaises(argparse.ArgumentTypeError):
167        NetworkConfig.parse_wpr(invalid)
168
169  def test_parse_wpr(self):
170    archive_path = pth.LocalPath("test/archive.wprgo")
171    with self.assertRaises(argparse.ArgumentTypeError) as cm:
172      NetworkConfig.parse_wpr(archive_path)
173    self.assertIn(str(archive_path), str(cm.exception))
174    self.fs.create_file(archive_path, st_size=100)
175    config = NetworkConfig.parse_wpr(archive_path)
176    self.assertEqual(config.type, NetworkType.WPR)
177    self.assertEqual(config.speed, NetworkSpeedConfig.default())
178    self.assertEqual(config.path, archive_path)
179    self.assertEqual(config, NetworkConfig.parse(archive_path))
180
181  def test_parse_dict_default(self):
182    config = NetworkConfig.parse({})
183    self.assertEqual(config, NetworkConfig.default())
184
185  def test_parse_local_dict_default(self):
186    with self.assertRaises(argparse.ArgumentTypeError):
187      # Missing path
188      NetworkConfig.parse_local({})
189
190  def test_parse_dict_speed(self):
191    config_dict = {"speed": "4G"}
192    config: NetworkConfig = NetworkConfig.parse(dict(config_dict))
193    self.assertNotEqual(config, NetworkConfig.default())
194    self.assertEqual(config.type, NetworkType.LIVE)
195    self.assertEqual(
196        config.speed,
197        NetworkSpeedConfig.parse_preset(NetworkSpeedPreset.MOBILE_4G))
198    self.assertIsNone(config.path)
199    self.assertTrue(config_dict)
200    config_1 = NetworkConfig.parse(json.dumps(config_dict))
201    self.assertEqual(config, config_1)
202
203  def test_parse_dict_wpr(self):
204    archive_path = pth.LocalPath("test/archive.wprgo")
205    with self.assertRaises(argparse.ArgumentTypeError) as cm:
206      NetworkConfig.parse({"type": "wpr", "path": archive_path})
207    self.assertIn(str(archive_path), str(cm.exception))
208    self.fs.create_file(archive_path, st_size=100)
209    config_dict = {"type": "wpr", "path": str(archive_path)}
210    config = NetworkConfig.parse(dict(config_dict))
211    self.assertEqual(config, NetworkConfig.parse_wpr(archive_path))
212    self.assertTrue(config_dict)
213    config_1 = NetworkConfig.parse(json.dumps(config_dict))
214    self.assertEqual(config, config_1)
215
216  def test_parse_dict_local(self):
217    benchmark_folder = pth.LocalPath("third_party/speedometer/v3.0")
218    with self.assertRaises(argparse.ArgumentTypeError) as cm:
219      NetworkConfig.parse({"type": "local", "path": benchmark_folder})
220    self.assertIn(str(benchmark_folder), str(cm.exception))
221    self.fs.create_file(benchmark_folder / "index.html", st_size=100)
222    url = "http://foo:1234"
223    config_dict = {"type": "local", "path": str(benchmark_folder), "url": url}
224    config = NetworkConfig.parse(dict(config_dict))
225    self.assertEqual(config.type, NetworkType.LOCAL)
226    self.assertEqual(config.path, benchmark_folder)
227    self.assertEqual(config.url, url)
228    self.assertTrue(config_dict)
229    config_1 = NetworkConfig.parse(json.dumps(config_dict))
230    self.assertEqual(config, config_1)
231    local_config_dict = dict(config_dict)
232    del local_config_dict["type"]
233    config_local = NetworkConfig.parse_local(dict(local_config_dict))
234    self.assertEqual(config, config_local)
235    config_local = NetworkConfig.parse_local(json.dumps(local_config_dict))
236    self.assertEqual(config, config_local)
237
238  def test_parse_local_file(self):
239    benchmark_folder = pth.LocalPath("third_party/speedometer/v3.0")
240    with self.assertRaises(argparse.ArgumentTypeError) as cm:
241      NetworkConfig.parse(benchmark_folder)
242    self.assertIn(str(benchmark_folder), str(cm.exception))
243    with self.assertRaises(argparse.ArgumentTypeError) as cm:
244      NetworkConfig.parse_local(benchmark_folder)
245    self.assertIn(str(benchmark_folder), str(cm.exception))
246    self.fs.create_file(benchmark_folder / "index.html", st_size=100)
247    config = NetworkConfig.parse(str(benchmark_folder))
248    self.assertEqual(config.type, NetworkType.LOCAL)
249    self.assertEqual(config.path, benchmark_folder)
250    self.assertIsNone(config.url)
251    self.assertEqual(config, NetworkConfig.parse(benchmark_folder))
252    self.assertEqual(config, NetworkConfig.parse_local(str(benchmark_folder)))
253    self.assertEqual(config, NetworkConfig.parse_local(benchmark_folder))
254
255
256if __name__ == "__main__":
257  test_helper.run_pytest(__file__)
258