"""Tests for net.py — iw parsing, the trace verdict, tethering, the connection-name cache. Run: python3 -m unittest discover -s test """ import sys import unittest from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from nexthopd import net # noqa: E402 from nexthopd.net import nm_metered, tether_from_gateway # noqa: E402 from support import FIXTURES # noqa: E402 class IwParsing(unittest.TestCase): def test_link_fixture(self): raw = (FIXTURES / "iw-link.txt").read_text() original = net._run net._run = lambda cmd, timeout=2.0: raw try: info = net.wifi_link("wlo1") finally: net._run = original self.assertEqual(info["ssid"], "Excitel") self.assertEqual(info["freq_mhz"], 5180) # float in fixture, int out self.assertEqual(info["signal_dbm"], -64) self.assertEqual(info["band"], "5 GHz") self.assertEqual(info["channel"], 36) self.assertEqual(info["standard"], "802.11ax") self.assertEqual(info["width_mhz"], 40) def test_channel_map(self): self.assertEqual(net._freq_to_channel(2412), 1) self.assertEqual(net._freq_to_channel(2484), 14) self.assertEqual(net._freq_to_channel(5180), 36) self.assertEqual(net._freq_to_channel(5955), 1) class TraceParsing(unittest.TestCase): """The cdn-cgi/trace response yields one validated address or nothing.""" def test_recorded_response(self): # Recorded from a real fetch of speed.cloudflare.com/cdn-cgi/trace # (address substituted): sixteen key=value lines, ip= among them. text = (FIXTURES / "cf-trace.txt").read_text() # The recorded fixture carries loc=XX (Cloudflare's "unknown"), so # no country survives; its colo passes through validated. self.assertEqual(net.parse_trace(text), {"ip": "198.51.100.7", "family": "v4", "edge": "XXX"}) def test_country_and_edge_come_free_with_the_address(self): # Both are already in the response the reachability check fetches. self.assertEqual( net.parse_trace("ip=1.2.3.4\nloc=IN\ncolo=DEL\n"), {"ip": "1.2.3.4", "family": "v4", "country": "IN", "edge": "DEL"}) def test_unknown_country_is_withheld_not_shown(self): # Cloudflare answers XX when it does not know. Showing a country # called XX would be inventing one. got = net.parse_trace("ip=1.2.3.4\nloc=XX\ncolo=DEL\n") self.assertNotIn("country", got) self.assertEqual(got["edge"], "DEL") def test_only_a_country_shaped_country_gets_out(self): # Nothing free-form from the wire reaches the shell, same rule as # the address itself. for bad in ("in", "IND", "I", "I1", "", "\u00cd\u00d1"): got = net.parse_trace("ip=1.2.3.4\nloc=%s\n" % bad) self.assertNotIn("country", got, bad) for bad in ("del", "D", "TOOLONG", "D3L", ""): got = net.parse_trace("ip=1.2.3.4\ncolo=%s\n" % bad) self.assertNotIn("edge", got, bad) def test_country_and_edge_never_stand_in_for_an_address(self): # The address is the point; decoration alone is not a result. self.assertIsNone(net.parse_trace("loc=IN\ncolo=DEL\n")) self.assertIsNone(net.parse_trace("ip=nope\nloc=IN\ncolo=DEL\n")) def test_v6_is_labelled(self): self.assertEqual(net.parse_trace("h=x\nip=2001:db8::7\nts=1\n"), {"ip": "2001:db8::7", "family": "v6"}) def test_only_a_real_address_gets_out(self): # Whatever else the response holds must never reach the shell. self.assertIsNone(net.parse_trace("ip=not-an-ip\n")) self.assertIsNone(net.parse_trace("ip=1.2.3.4.5\n")) self.assertIsNone(net.parse_trace("h=x\nts=1\n")) self.assertIsNone(net.parse_trace("")) def test_input_is_bounded_before_parsing(self): # An ip= line beyond the size cap is as good as absent. self.assertIsNone(net.parse_trace("x=" + "a" * 5000 + "\nip=1.2.3.4\n")) self.assertIsNone(net.parse_trace("k=v\n" * 100 + "ip=1.2.3.4\n")) class TetherDetection(unittest.TestCase): """A phone sharing its data, from the one signal that is reliable.""" def test_the_documented_ranges(self): self.assertEqual(tether_from_gateway("172.20.10.1"), {"kind": "ios", "label": "iPhone"}) self.assertEqual(tether_from_gateway("192.168.43.1")["kind"], "android") self.assertEqual(tether_from_gateway("192.168.42.129")["kind"], "android") self.assertEqual(tether_from_gateway("192.168.137.1")["kind"], "windows") def test_ordinary_gateways_are_not_phones(self): # Including the hotel gateway that started this: 172.20.0.1 is close # to the iOS range and outside it, so the /28 matters. for gw in ("192.168.1.1", "172.20.0.1", "10.0.0.1", "172.20.11.1"): self.assertIsNone(tether_from_gateway(gw), gw) def test_nothing_and_nonsense_are_not_phones(self): for gw in ("", None, "not-an-ip", "999.1.1.1"): self.assertIsNone(tether_from_gateway(gw)) def test_ipv6_gateway_does_not_raise(self): self.assertIsNone(tether_from_gateway("fe80::1")) def test_a_guess_from_networkmanager_is_not_evidence(self): # A live iPhone hotspot reports "no (guessed)", so only an explicit # answer may count — and a guessed YES must not either. import nexthopd.net as netmod original = netmod._run try: for raw, expected in ( ("GENERAL.METERED:yes", True), ("GENERAL.METERED:yes (guessed)", False), ("GENERAL.METERED:no", False), ("GENERAL.METERED:no (guessed)", False), ("", False), (None, False), ): netmod._run = lambda *a, **k: raw self.assertEqual(nm_metered("wlan0"), expected, raw) finally: netmod._run = original class ConnectionNameCache(unittest.TestCase): def test_nmcli_is_asked_on_a_new_key_or_after_the_ttl_only(self): calls = [] orig = net.connection_name net.connection_name = lambda iface: calls.append(iface) or "Home" net._name_cache.clear() self.addCleanup(setattr, net, "connection_name", orig) self.addCleanup(net._name_cache.clear) key = ("wlo1", "192.168.1.1", "aa:bb") self.assertEqual(net.connection_name_cached("wlo1", key, now=100), "Home") self.assertEqual(net.connection_name_cached("wlo1", key, now=130), "Home") self.assertEqual(len(calls), 1) # A new gateway or BSSID is a new network: ask again. net.connection_name_cached("wlo1", ("wlo1", "10.0.0.1", "aa:bb"), now=131) self.assertEqual(len(calls), 2) # ...and so is a rename the user made, eventually. net.connection_name_cached("wlo1", ("wlo1", "10.0.0.1", "aa:bb"), now=131 + net.NAME_CACHE_TTL_S + 1) self.assertEqual(len(calls), 3) if __name__ == "__main__": unittest.main()