| 1 | """Tests for the netshape instrument: config, pure command builders,
|
| 2 | availability honesty, the RTT self-test, and shaped-row comparison legality."""
|
| 3 |
|
| 4 | from __future__ import annotations
|
| 5 |
|
| 6 | import json
|
| 7 | import subprocess
|
| 8 | import sys
|
| 9 | import tempfile
|
| 10 | import unittest
|
| 11 | from pathlib import Path
|
| 12 |
|
| 13 | sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
| 14 |
|
| 15 | from oakbench import netshape, remotes
|
| 16 |
|
| 17 | ALL_PROFILES = {
|
| 18 | "loopback-0ms",
|
| 19 | "lan-1ms",
|
| 20 | "wan-30ms",
|
| 21 | "wan-100ms-1pct",
|
| 22 | "cellular-300ms-jitter",
|
| 23 | }
|
| 24 |
|
| 25 |
|
| 26 | def write_config(tmp: str, document: dict) -> Path:
|
| 27 | path = Path(tmp) / "netshapes.json"
|
| 28 | path.write_text(json.dumps(document))
|
| 29 | return path
|
| 30 |
|
| 31 |
|
| 32 | def profile(**overrides) -> dict:
|
| 33 | base = {"rtt_ms": 10, "bandwidth_mbit": 100, "loss_pct": 0, "jitter_ms": 0}
|
| 34 | base.update(overrides)
|
| 35 | return base
|
| 36 |
|
| 37 |
|
| 38 | class LoadNetshapesTest(unittest.TestCase):
|
| 39 | def test_default_config_loads_all_five_profiles(self) -> None:
|
| 40 | profiles = netshape.load_netshapes()
|
| 41 | self.assertEqual(set(profiles), ALL_PROFILES)
|
| 42 | for item in profiles.values():
|
| 43 | self.assertIsInstance(item, netshape.NetshapeProfile)
|
| 44 |
|
| 45 | def test_profile_values_match_plan(self) -> None:
|
| 46 | profiles = netshape.load_netshapes()
|
| 47 | loopback = profiles["loopback-0ms"]
|
| 48 | self.assertEqual(loopback.rtt_ms, 0)
|
| 49 | self.assertIsNone(loopback.bandwidth_mbit) # null means unshaped bandwidth
|
| 50 | self.assertEqual(profiles["lan-1ms"].rtt_ms, 1)
|
| 51 | self.assertEqual(profiles["wan-30ms"].rtt_ms, 30)
|
| 52 | wan_lossy = profiles["wan-100ms-1pct"]
|
| 53 | self.assertEqual(wan_lossy.rtt_ms, 100)
|
| 54 | self.assertEqual(wan_lossy.loss_pct, 1)
|
| 55 | cellular = profiles["cellular-300ms-jitter"]
|
| 56 | self.assertEqual(cellular.rtt_ms, 300)
|
| 57 | self.assertEqual(cellular.jitter_ms, 50)
|
| 58 |
|
| 59 | def test_rejects_wrong_schema_version(self) -> None:
|
| 60 | with tempfile.TemporaryDirectory() as tmp:
|
| 61 | path = write_config(tmp, {"schema_version": 2, "profiles": {"p": profile()}})
|
| 62 | with self.assertRaisesRegex(netshape.NetshapeConfigError, "schema_version"):
|
| 63 | netshape.load_netshapes(path)
|
| 64 |
|
| 65 | def test_rejects_missing_profile_key(self) -> None:
|
| 66 | with tempfile.TemporaryDirectory() as tmp:
|
| 67 | bad = profile()
|
| 68 | del bad["loss_pct"]
|
| 69 | path = write_config(tmp, {"schema_version": 1, "profiles": {"p": bad}})
|
| 70 | with self.assertRaisesRegex(netshape.NetshapeConfigError, "loss_pct"):
|
| 71 | netshape.load_netshapes(path)
|
| 72 |
|
| 73 | def test_rejects_unknown_profile_key(self) -> None:
|
| 74 | with tempfile.TemporaryDirectory() as tmp:
|
| 75 | path = write_config(tmp, {"schema_version": 1, "profiles": {"p": profile(latency_ms=5)}})
|
| 76 | with self.assertRaisesRegex(netshape.NetshapeConfigError, "latency_ms"):
|
| 77 | netshape.load_netshapes(path)
|
| 78 |
|
| 79 | def test_rejects_negative_rtt(self) -> None:
|
| 80 | with tempfile.TemporaryDirectory() as tmp:
|
| 81 | path = write_config(tmp, {"schema_version": 1, "profiles": {"p": profile(rtt_ms=-1)}})
|
| 82 | with self.assertRaisesRegex(netshape.NetshapeConfigError, "rtt_ms"):
|
| 83 | netshape.load_netshapes(path)
|
| 84 |
|
| 85 | def test_rejects_loss_over_100(self) -> None:
|
| 86 | with tempfile.TemporaryDirectory() as tmp:
|
| 87 | path = write_config(tmp, {"schema_version": 1, "profiles": {"p": profile(loss_pct=101)}})
|
| 88 | with self.assertRaisesRegex(netshape.NetshapeConfigError, "loss_pct"):
|
| 89 | netshape.load_netshapes(path)
|
| 90 |
|
| 91 | def test_rejects_empty_profiles(self) -> None:
|
| 92 | with tempfile.TemporaryDirectory() as tmp:
|
| 93 | path = write_config(tmp, {"schema_version": 1, "profiles": {}})
|
| 94 | with self.assertRaisesRegex(netshape.NetshapeConfigError, "profiles"):
|
| 95 | netshape.load_netshapes(path)
|
| 96 |
|
| 97 | def test_rejects_missing_file(self) -> None:
|
| 98 | with self.assertRaises(netshape.NetshapeConfigError):
|
| 99 | netshape.load_netshapes(Path("/nonexistent/netshapes.json"))
|
| 100 |
|
| 101 |
|
| 102 | class NetemArgsTest(unittest.TestCase):
|
| 103 | def test_wan_100ms_1pct_args(self) -> None:
|
| 104 | wan = netshape.load_netshapes()["wan-100ms-1pct"]
|
| 105 | # rtt/jitter are round-trip: half per direction; loss/rate per direction.
|
| 106 | self.assertEqual(netshape.netem_args(wan), ["delay", "50ms", "loss", "1%", "rate", "50mbit"])
|
| 107 |
|
| 108 | def test_cellular_jitter_args(self) -> None:
|
| 109 | cellular = netshape.load_netshapes()["cellular-300ms-jitter"]
|
| 110 | self.assertEqual(netshape.netem_args(cellular), ["delay", "150ms", "25ms", "rate", "10mbit"])
|
| 111 |
|
| 112 | def test_loopback_args(self) -> None:
|
| 113 | loopback = netshape.load_netshapes()["loopback-0ms"]
|
| 114 | self.assertEqual(netshape.netem_args(loopback), ["delay", "0ms"])
|
| 115 |
|
| 116 | def test_lan_half_millisecond_delay(self) -> None:
|
| 117 | lan = netshape.load_netshapes()["lan-1ms"]
|
| 118 | self.assertEqual(netshape.netem_args(lan), ["delay", "0.5ms", "rate", "1000mbit"])
|
| 119 |
|
| 120 |
|
| 121 | class NetnsCommandsTest(unittest.TestCase):
|
| 122 | def test_exact_command_lists_for_wan_100ms_1pct(self) -> None:
|
| 123 | wan = netshape.load_netshapes()["wan-100ms-1pct"]
|
| 124 | shaping = ["delay", "50ms", "loss", "1%", "rate", "50mbit"]
|
| 125 | expected = [
|
| 126 | ["ip", "netns", "add", "bench0"],
|
| 127 | ["ip", "link", "add", "ve-bench0-h", "type", "veth", "peer", "name", "ve-bench0-n"],
|
| 128 | ["ip", "link", "set", "ve-bench0-n", "netns", "bench0"],
|
| 129 | ["ip", "addr", "add", "10.231.47.1/24", "dev", "ve-bench0-h"],
|
| 130 | ["ip", "link", "set", "ve-bench0-h", "up"],
|
| 131 | ["ip", "netns", "exec", "bench0", "ip", "addr", "add", "10.231.47.2/24", "dev", "ve-bench0-n"],
|
| 132 | ["ip", "netns", "exec", "bench0", "ip", "link", "set", "ve-bench0-n", "up"],
|
| 133 | ["ip", "netns", "exec", "bench0", "ip", "link", "set", "lo", "up"],
|
| 134 | ["tc", "qdisc", "add", "dev", "ve-bench0-h", "root", "netem", *shaping],
|
| 135 | ["ip", "netns", "exec", "bench0", "tc", "qdisc", "add", "dev", "ve-bench0-n", "root", "netem", *shaping],
|
| 136 | ]
|
| 137 | self.assertEqual(netshape.netns_commands("bench0", wan), expected)
|
| 138 |
|
| 139 | def test_teardown_commands(self) -> None:
|
| 140 | self.assertEqual(
|
| 141 | netshape.teardown_commands("bench0"),
|
| 142 | [["ip", "link", "del", "ve-bench0-h"], ["ip", "netns", "del", "bench0"]],
|
| 143 | )
|
| 144 |
|
| 145 | def test_rejects_bad_netns_name(self) -> None:
|
| 146 | wan = netshape.load_netshapes()["wan-30ms"]
|
| 147 | with self.assertRaises(ValueError):
|
| 148 | netshape.netns_commands("bad name; rm -rf", wan)
|
| 149 | with self.assertRaises(ValueError):
|
| 150 | netshape.netns_commands("a-name-far-too-long-for-ifnamsiz", wan)
|
| 151 |
|
| 152 | def test_netns_exec_wraps_command(self) -> None:
|
| 153 | self.assertEqual(
|
| 154 | netshape.netns_exec("bench0", ["git", "clone", "url"]),
|
| 155 | ["ip", "netns", "exec", "bench0", "git", "clone", "url"],
|
| 156 | )
|
| 157 |
|
| 158 |
|
| 159 | class NetshapeAvailableTest(unittest.TestCase):
|
| 160 | def test_this_host_reports_honestly(self) -> None:
|
| 161 | ok, reason = netshape.netshape_available()
|
| 162 | if sys.platform != "linux":
|
| 163 | # This dev host is macOS: the netem path cannot be exercised here,
|
| 164 | # and the probe must say so instead of pretending.
|
| 165 | self.assertFalse(ok)
|
| 166 | self.assertTrue(reason.startswith("requires_linux:"), reason)
|
| 167 | else:
|
| 168 | self.assertIsInstance(ok, bool)
|
| 169 |
|
| 170 | def test_non_linux_reason(self) -> None:
|
| 171 | ok, reason = netshape.netshape_available(platform="darwin")
|
| 172 | self.assertFalse(ok)
|
| 173 | self.assertEqual(reason, "requires_linux:darwin")
|
| 174 |
|
| 175 | def test_missing_ip_binary(self) -> None:
|
| 176 | ok, reason = netshape.netshape_available(platform="linux", which=lambda name: None)
|
| 177 | self.assertFalse(ok)
|
| 178 | self.assertEqual(reason, "missing_binary:ip")
|
| 179 |
|
| 180 | def test_missing_tc_binary(self) -> None:
|
| 181 | which = lambda name: "/sbin/ip" if name == "ip" else None # noqa: E731
|
| 182 | ok, reason = netshape.netshape_available(platform="linux", which=which)
|
| 183 | self.assertFalse(ok)
|
| 184 | self.assertEqual(reason, "missing_binary:tc")
|
| 185 |
|
| 186 | def test_no_cap_net_admin(self) -> None:
|
| 187 | ok, reason = netshape.netshape_available(
|
| 188 | platform="linux", which=lambda name: f"/sbin/{name}", geteuid=lambda: 1000
|
| 189 | )
|
| 190 | self.assertFalse(ok)
|
| 191 | self.assertEqual(reason, "requires_cap_net_admin:euid_not_root")
|
| 192 |
|
| 193 | def test_root_linux_available(self) -> None:
|
| 194 | ok, reason = netshape.netshape_available(
|
| 195 | platform="linux", which=lambda name: f"/sbin/{name}", geteuid=lambda: 0
|
| 196 | )
|
| 197 | self.assertTrue(ok)
|
| 198 | self.assertIsNone(reason)
|
| 199 |
|
| 200 |
|
| 201 | class NetnsShaperTest(unittest.TestCase):
|
| 202 | def test_create_runs_exact_commands_and_teardown(self) -> None:
|
| 203 | wan = netshape.load_netshapes()["wan-30ms"]
|
| 204 | executed: list[list[str]] = []
|
| 205 | shaper = netshape.NetnsShaper(
|
| 206 | "bench0", run=executed.append, available_fn=lambda: (True, None)
|
| 207 | )
|
| 208 | shaper.create(wan)
|
| 209 | self.assertEqual(executed, netshape.netns_commands("bench0", wan))
|
| 210 | self.assertEqual(shaper.profile, wan)
|
| 211 | executed.clear()
|
| 212 | shaper.teardown()
|
| 213 | self.assertEqual(executed, netshape.teardown_commands("bench0"))
|
| 214 | self.assertIsNone(shaper.profile)
|
| 215 |
|
| 216 | def test_create_refuses_when_unavailable(self) -> None:
|
| 217 | wan = netshape.load_netshapes()["wan-30ms"]
|
| 218 | shaper = netshape.NetnsShaper(
|
| 219 | "bench0", run=lambda command: None, available_fn=lambda: (False, "requires_linux:darwin")
|
| 220 | )
|
| 221 | with self.assertRaisesRegex(RuntimeError, "netshape_unavailable:requires_linux:darwin"):
|
| 222 | shaper.create(wan)
|
| 223 |
|
| 224 | def test_partial_create_failure_runs_teardown_and_reraises(self) -> None:
|
| 225 | """A mid-create failure must not leak namespace/veth state: the
|
| 226 | teardown commands run (best-effort) and the original error
|
| 227 | propagates."""
|
| 228 | wan = netshape.load_netshapes()["wan-30ms"]
|
| 229 | expected_create = netshape.netns_commands("bench0", wan)
|
| 230 | executed: list[list[str]] = []
|
| 231 |
|
| 232 | def runner(command: list[str]) -> None:
|
| 233 | executed.append(command)
|
| 234 | if command == expected_create[1]: # fail on command 2
|
| 235 | raise subprocess.CalledProcessError(1, command)
|
| 236 |
|
| 237 | shaper = netshape.NetnsShaper("bench0", run=runner, available_fn=lambda: (True, None))
|
| 238 | with self.assertRaises(subprocess.CalledProcessError) as ctx:
|
| 239 | shaper.create(wan)
|
| 240 | self.assertEqual(ctx.exception.cmd, expected_create[1])
|
| 241 | # Commands 1 and 2 ran, then BOTH teardown commands were attempted.
|
| 242 | self.assertEqual(executed, expected_create[:2] + netshape.teardown_commands("bench0"))
|
| 243 | self.assertIsNone(shaper.profile)
|
| 244 | # The in-create cleanup already tore down; a later (e.g. context
|
| 245 | # manager) teardown is a no-op, not a second command volley.
|
| 246 | executed.clear()
|
| 247 | shaper.teardown()
|
| 248 | self.assertEqual(executed, [])
|
| 249 |
|
| 250 | def test_partial_create_failure_reraises_even_if_teardown_commands_fail(self) -> None:
|
| 251 | wan = netshape.load_netshapes()["wan-30ms"]
|
| 252 | expected_create = netshape.netns_commands("bench0", wan)
|
| 253 | attempted: list[list[str]] = []
|
| 254 |
|
| 255 | def runner(command: list[str]) -> None:
|
| 256 | attempted.append(command)
|
| 257 | if command == expected_create[1]:
|
| 258 | raise OSError("create failed on command 2")
|
| 259 | if command in netshape.teardown_commands("bench0"):
|
| 260 | raise subprocess.CalledProcessError(2, command)
|
| 261 |
|
| 262 | shaper = netshape.NetnsShaper("bench0", run=runner, available_fn=lambda: (True, None))
|
| 263 | with self.assertRaisesRegex(OSError, "create failed on command 2"):
|
| 264 | shaper.create(wan)
|
| 265 | self.assertEqual(attempted, expected_create[:2] + netshape.teardown_commands("bench0"))
|
| 266 |
|
| 267 | def test_teardown_without_create_is_noop(self) -> None:
|
| 268 | executed: list[list[str]] = []
|
| 269 | shaper = netshape.NetnsShaper("bench0", run=executed.append)
|
| 270 | shaper.teardown()
|
| 271 | self.assertEqual(executed, [])
|
| 272 |
|
| 273 | def test_teardown_swallows_command_failures(self) -> None:
|
| 274 | wan = netshape.load_netshapes()["wan-30ms"]
|
| 275 |
|
| 276 | def runner(command: list[str]) -> None:
|
| 277 | if command[:2] == ["ip", "link"] and "del" in command:
|
| 278 | raise subprocess.CalledProcessError(1, command)
|
| 279 |
|
| 280 | shaper = netshape.NetnsShaper("bench0", run=runner, available_fn=lambda: (True, None))
|
| 281 | shaper.create(wan)
|
| 282 | shaper.teardown() # must not raise
|
| 283 |
|
| 284 |
|
| 285 | class SelfTestTest(unittest.TestCase):
|
| 286 | def setUp(self) -> None:
|
| 287 | self.wan = netshape.load_netshapes()["wan-30ms"]
|
| 288 |
|
| 289 | def test_within_tolerance(self) -> None:
|
| 290 | row = netshape.self_test(self.wan, ping_fn=lambda: 31.0)
|
| 291 | self.assertEqual(row["operation"], "netshape.selftest")
|
| 292 | self.assertEqual(row["netshape_profile"], "wan-30ms")
|
| 293 | self.assertEqual(row["configured_rtt_ms"], 30.0)
|
| 294 | self.assertEqual(row["measured_rtt_ms"], 31.0)
|
| 295 | self.assertTrue(row["within_tolerance"])
|
| 296 |
|
| 297 | def test_outside_tolerance(self) -> None:
|
| 298 | # 15% of 30ms is 4.5ms; 40ms is out.
|
| 299 | row = netshape.self_test(self.wan, ping_fn=lambda: 40.0)
|
| 300 | self.assertFalse(row["within_tolerance"])
|
| 301 |
|
| 302 | def test_boundary_is_inclusive(self) -> None:
|
| 303 | row = netshape.self_test(self.wan, ping_fn=lambda: 34.5)
|
| 304 | self.assertTrue(row["within_tolerance"])
|
| 305 | row = netshape.self_test(self.wan, ping_fn=lambda: 34.6)
|
| 306 | self.assertFalse(row["within_tolerance"])
|
| 307 |
|
| 308 | def test_loopback_uses_absolute_floor(self) -> None:
|
| 309 | loopback = netshape.load_netshapes()["loopback-0ms"]
|
| 310 | # 15% of 0ms is 0; without the 1ms floor every real syscall would fail.
|
| 311 | self.assertTrue(netshape.self_test(loopback, ping_fn=lambda: 0.4)["within_tolerance"])
|
| 312 | self.assertFalse(netshape.self_test(loopback, ping_fn=lambda: 1.5)["within_tolerance"])
|
| 313 |
|
| 314 | def test_uses_median_of_samples(self) -> None:
|
| 315 | samples = iter([29.0, 30.0, 31.0, 200.0, 28.0])
|
| 316 | row = netshape.self_test(self.wan, ping_fn=lambda: next(samples), samples=5)
|
| 317 | self.assertEqual(row["measured_rtt_ms"], 30.0) # median of the 5 samples, outlier ignored
|
| 318 | self.assertEqual(len(row["rtt_samples_ms"]), 5)
|
| 319 | self.assertTrue(row["within_tolerance"])
|
| 320 |
|
| 321 | def test_requires_ping_fn_or_shaper(self) -> None:
|
| 322 | with self.assertRaises(ValueError):
|
| 323 | netshape.self_test(self.wan)
|
| 324 |
|
| 325 |
|
| 326 | class ShapedComparisonLegalTest(unittest.TestCase):
|
| 327 | @staticmethod
|
| 328 | def shaped(profile_name: str = "wan-30ms", server: str = "netshape-local") -> dict:
|
| 329 | return {
|
| 330 | "remote_transport": remotes.TRANSPORT_NETWORK_SHAPED,
|
| 331 | "netshape_profile": profile_name,
|
| 332 | "remote_server": server,
|
| 333 | }
|
| 334 |
|
| 335 | def test_same_profile_same_server_is_legal(self) -> None:
|
| 336 | self.assertTrue(remotes.shaped_comparison_legal(self.shaped(), self.shaped()))
|
| 337 |
|
| 338 | def test_different_profile_is_illegal(self) -> None:
|
| 339 | self.assertFalse(
|
| 340 | remotes.shaped_comparison_legal(self.shaped("wan-30ms"), self.shaped("wan-100ms-1pct"))
|
| 341 | )
|
| 342 |
|
| 343 | def test_different_server_is_illegal(self) -> None:
|
| 344 | self.assertFalse(
|
| 345 | remotes.shaped_comparison_legal(self.shaped(server="a"), self.shaped(server="b"))
|
| 346 | )
|
| 347 |
|
| 348 | def test_unshaped_transport_is_illegal_either_side(self) -> None:
|
| 349 | network_row = {
|
| 350 | "remote_transport": remotes.TRANSPORT_NETWORK,
|
| 351 | "netshape_profile": "wan-30ms",
|
| 352 | "remote_server": "netshape-local",
|
| 353 | }
|
| 354 | self.assertFalse(remotes.shaped_comparison_legal(network_row, self.shaped()))
|
| 355 | self.assertFalse(remotes.shaped_comparison_legal(self.shaped(), network_row))
|
| 356 | local_row = dict(network_row, remote_transport=remotes.TRANSPORT_LOCAL_FILE)
|
| 357 | self.assertFalse(remotes.shaped_comparison_legal(local_row, local_row))
|
| 358 |
|
| 359 | def test_missing_profile_or_server_is_illegal(self) -> None:
|
| 360 | no_profile = self.shaped()
|
| 361 | no_profile["netshape_profile"] = None
|
| 362 | self.assertFalse(remotes.shaped_comparison_legal(no_profile, no_profile))
|
| 363 | no_server = self.shaped()
|
| 364 | no_server["remote_server"] = None
|
| 365 | self.assertFalse(remotes.shaped_comparison_legal(no_server, no_server))
|
| 366 | self.assertFalse(remotes.shaped_comparison_legal({}, {}))
|
| 367 |
|
| 368 | def test_existing_transport_constants_untouched(self) -> None:
|
| 369 | self.assertEqual(remotes.TRANSPORT_NETWORK, "network")
|
| 370 | self.assertEqual(remotes.TRANSPORT_LOCAL_FILE, "local_file")
|
| 371 | self.assertEqual(remotes.TRANSPORT_NETWORK_SHAPED, "network_shaped")
|
| 372 |
|
| 373 |
|
| 374 | if __name__ == "__main__":
|
| 375 | unittest.main()
|