diff --git a/README.md b/README.md index 2d3322e..0aeee96 100644 --- a/README.md +++ b/README.md @@ -47,8 +47,10 @@ Row 0 holds the group buttons. Each one shows the first problem in its group, pl | APPS: PROPRES OFFLINE / NOT RUNNING | restarts ProPresenter on its Mac (the newest version installed) | | APPS: ABLETON / TRANSLATE NOT RUNNING | starts Ableton on the Tracks Mac / Church Translate on NA Translation | | DANTE: PROP MAC / TRACKS / TRANSL / INGEST GONE | starts Dante Virtual Soundcard's engine on that Mac (`remote-access/mac-dvs-helper.sh`) | - | DANTE: STB MON / SOLO / TB / TB MON GONE | power-cycles that box's PoE port on the Netgear Dante switch 10.66.1.102 (ports 0/14–0/17), after checking it is still that box on the port (`netgear.py`) | + | DANTE: any device powered by its switch port (today the four Studio B boxes) | power-cycles the port it is on **now**, found by its MAC on the Netgear Dante switches | | DANTE: any other device GONE | restarts Companion's Dante module (hardware: if still gone, it is off or unplugged) | + + **The Dante map is automatic.** Every 10 minutes the watchdog maps each Dante device: its name and IP from Companion, its MAC from NA Ingest's Dante port, and its switch port and PoE from the Netgear switches listed in `"dante_switches"`. It saves the map to `dante-map.json` in its data folder. Re-patching, adding or renaming a device needs no config change. A device that has been dead for a while ages out of the switch tables, so its last known port is used, but only if no other device has appeared there. A port carrying more than 2 devices (an uplink) is never power-cycled. The switch login is in NA Ingest's login keychain (service `netgear-status-panel`). | INGEST: BRIDGE / CAPTIONS / PRODDASH | restarts that service on NA Ingest | | HD1–6 lost by Companion | reboots that deck. Never one that is recording, and never for NO DISK or NOT REC | | anything else with a Companion connection | restarts that Companion connection | diff --git a/config.json b/config.json index e422259..1ebdea6 100644 --- a/config.json +++ b/config.json @@ -26,6 +26,8 @@ "alert_webhook": "", "ssh_key": "~/.ssh/id_status_panel", + + "dante_switches": ["10.66.1.100", "10.66.1.102"], "hosts": { "prop": { "ssh": "napropresenter@10.1.14.225" }, "tracks": { "ssh": "waterschurchmedia@10.1.3.227" }, @@ -48,11 +50,7 @@ "captions": { "kind": "local", "short": "CAPTIONS", "label": "restart lobby captions", "run": "launchctl kickstart -k gui/$(id -u)/org.waterschurch.lobby-captions || launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/org.waterschurch.lobby-captions.plist" }, "proddash": { "kind": "local", "short": "PRODDASH", "label": "start ProdDash", "run": "open /Applications/ProdDash.app" }, - "hyperdeck": { "kind": "hyperdeck", "short": "HYPERDECK", "label": "reconnect Companion to the HyperDeck, or reboot the deck only if it is hung" }, - "poe_stb_solo": { "kind": "netgear_poe", "switch": "10.66.1.102", "port": "0/14", "mac": "00:1d:c1:51:34:c8", "short": "POWER CYCLE", "label": "power-cycle STUDIO-B-SOLO-MON (Dante switch 10.66.1.102 port 0/14)" }, - "poe_stb_mon": { "kind": "netgear_poe", "switch": "10.66.1.102", "port": "0/15", "mac": "00:1d:c1:51:18:84", "short": "POWER CYCLE", "label": "power-cycle STUDIO-B-MON (Dante switch 10.66.1.102 port 0/15)" }, - "poe_stb_tb": { "kind": "netgear_poe", "switch": "10.66.1.102", "port": "0/16", "mac": "00:1d:c1:89:1a:0a", "short": "POWER CYCLE", "label": "power-cycle STUDIO-B-TB talkback (Dante switch 10.66.1.102 port 0/16)" }, - "poe_stb_tbmon": { "kind": "netgear_poe", "switch": "10.66.1.102", "port": "0/17", "mac": "00:1d:c1:51:34:ea", "short": "POWER CYCLE", "label": "power-cycle STUDIO-B-TB-MON (Dante switch 10.66.1.102 port 0/17)" } + "hyperdeck": { "kind": "hyperdeck", "short": "HYPERDECK", "label": "reconnect Companion to the HyperDeck, or reboot the deck only if it is hung" } }, "tiles": [ @@ -77,9 +75,7 @@ { "type": "dante", "conn": "audinate-dantecontroller", "fix": { "NA-ProPresenter-Mac": "dvs_prop", "NA-Tracks": "dvs_tracks", "NA-Translation": "dvs_translation", "STUDIO-A-Ingest": "dvs_ingest", - "STUDIO-B-SOLO-MON": "poe_stb_solo", "STUDIO-B-MON": "poe_stb_mon", - "STUDIO-B-TB": "poe_stb_tb", "STUDIO-B-TB-MON": "poe_stb_tbmon", - "*": "restart-connection" }, + "*": "dante-auto" }, "devices": { "FOH-DiGiCo-DMI-1": "DMI-1", "FOH-DiGiCo-DMI-2": "DMI-2", "BOH-WING": "WING", "BOH-DQ-Rack": "DQ RACK", "BOH-AD4Q-A": "AD4Q-A", "BOH-AD4Q-B": "AD4Q-B", "STUDIO-A-X32": "STA X32", diff --git a/netgear.py b/netgear.py index 40f7d09..d253faa 100644 --- a/netgear.py +++ b/netgear.py @@ -108,6 +108,7 @@ MAC_RE = re.compile(r"(?:[0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}") def poe_cycle(host, port, expect_mac=None): + # Used directly only for a known port; poe_cycle_device finds the port by MAC. """Power-cycle PoE on ONE port ("poe reset" in Interface Config mode). Nothing is saved to the switch's startup config: the port's settings are unchanged afterwards. @@ -136,6 +137,79 @@ def poe_cycle(host, port, expect_mac=None): print("POE RESET %s %s ok%s" % (host, port, (": " + answer.strip()) if answer.strip() else "")) +# --- Find devices by MAC, wherever they are plugged in --------------------- + +TABLE_RE = re.compile(r"((?:[0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2})\s+(\d+/\d+)") +MAX_SHARED = 2 # more devices than this on one port means an uplink: never power-cycle it + + +def parse_mac_table(text): + """'show mac-addr-table' -> ({mac: port}, {port: number of MACs learned on it}).""" + where, counts = {}, {} + for mac, port in TABLE_RE.findall(text or ""): + where[mac.lower()] = port + counts[port] = counts.get(port, 0) + 1 + return where, counts + + +def best_ports(tables, macs): + """tables: {switch: table text}. -> {mac: (switch, port, devices on that port)}. + A MAC is seen on every switch it crosses; its own port is where it has the fewest + neighbours (on the far switch it shows up on the uplink with everything else).""" + best = {} + for host, text in tables.items(): + where, counts = parse_mac_table(text) + for mac in macs: + if mac in where: + cand = (host, where[mac], counts[where[mac]]) + if mac not in best or cand[2] < best[mac][2]: + best[mac] = cand + return best + + +def poe_delivering(text): + return bool(re.search(r"(?i)delivering power", text or "")) + + +def survey(switches, macs): + """Read-only: {mac: {"switch", "port", "shared", "poe"}} for every MAC found.""" + macs = [m.lower() for m in macs] + user, password = credentials() + sessions, tables = {}, {} + try: + for host in switches: + sessions[host] = Switch(host, user, password) + tables[host] = sessions[host].run("show mac-addr-table") + found = {} + for mac, (host, port, shared) in best_ports(tables, macs).items(): + poe = shared <= MAX_SHARED and poe_delivering(sessions[host].run("show poe port info %s" % port)) + found[mac] = {"switch": host, "port": port, "shared": shared, "poe": poe} + return found + finally: + for sw in sessions.values(): + sw.close() + + +def poe_cycle_device(switches, mac, last_known=None): + """Power-cycle the PoE port a device is on NOW, found by its MAC. If it has aged out of + the switch tables (dead long enough), use its last known port - poe_cycle then refuses + unless that port is empty of other devices. Never touches a port shared by more than + MAX_SHARED devices. Returns (switch, port, how it was found).""" + mac = mac.lower() + live = survey(switches, [mac]).get(mac) + if live: + if live["shared"] > MAX_SHARED: + raise RuntimeError("%s is behind %s port %s with %d devices (an uplink) - not power-cycling" + % (mac, live["switch"], live["port"], live["shared"])) + host, port, how = live["switch"], live["port"], "found now" + elif last_known and last_known.get("switch") and last_known.get("port"): + host, port, how = last_known["switch"], last_known["port"], "last known port" + else: + raise RuntimeError("%s is not on any Dante switch and has no last known port" % mac) + poe_cycle(host, port, expect_mac=mac) + return host, port, how + + if __name__ == "__main__": if len(sys.argv) >= 3 and sys.argv[1] == "lookup": lookup(sys.argv[2:]) diff --git a/status_panel.py b/status_panel.py index 31b1f85..2ba4d58 100644 --- a/status_panel.py +++ b/status_panel.py @@ -145,6 +145,8 @@ class Snapshot: self.disk = {} # path -> free GB self.listeners = {} # MIDI source name -> coremidi.Listener self.remote = {} # (host, command) -> True / False / None (could not ask) + self.dante_map = {} # Dante device name -> {"ip", "mac", "switch", "port", "shared", "poe", "seen"} + self.dante_map_time = 0.0 # last successful survey of the Dante switches def var(self, label, name): return self.vars.get((label, name)) @@ -455,6 +457,10 @@ RESTARTABLE = ("conn", "hyperdeck", "shure", "psm", "tascam") # A check's "fix" may name this instead of an entry in "fixes": restart its own Companion # connection. The Dante button uses it for hardware, which cannot be restarted from here. RESTART_CONNECTION = "restart-connection" +# The Dante button's catch-all: power-cycle the device's PoE port if the self-updating +# Dante map says it is PoE-powered (found by MAC, so re-patching needs no config change), +# otherwise restart Companion's Dante module. +DANTE_AUTO = "dante-auto" def needed_variables(cfg): @@ -599,6 +605,7 @@ class Engine: self.meta = {} # key -> (tile id, check, sub key) self.fix_last = {} # fix identity -> last run time self.executor = execute_fix + self.poe_executor = execute_poe self.notes = {} # tile id -> hold feedback {"text", "level", "until"} self.now = 0.0 @@ -754,6 +761,15 @@ class Engine: fix_id = chk.get("fix") if isinstance(fix_id, dict): fix_id = fix_id.get(sub, fix_id.get("*")) # "*": every other sub-check + if fix_id == DANTE_AUTO: + entry = snap.dante_map.get(sub) if chk["type"] == "dante" else None + if entry and entry.get("poe") and entry.get("mac"): + ident = ("poe", entry["mac"]) + if ident not in seen: + seen.add(ident) + plan.append((ident, "poe", sub, chk)) + continue + fix_id = RESTART_CONNECTION # not PoE-powered, or not mapped yet if fix_id == RESTART_CONNECTION and chk.get("conn"): ident = ("restart", chk["conn"]) # e.g. Companion's Dante module if ident not in seen: @@ -817,11 +833,15 @@ class Engine: ok = comp.restart_connection(c["id"]) log.info("RESTART %s -> %s", what, "ok" if ok else "failed") names.append("RECONNECT") + elif kind == "poe": + log.info("FIX %s: power-cycle %s's PoE port (found by MAC)", tile_id, what) + names.append("POWER CYCLE") + started.append((self.poe_executor, (self.cfg, what, dict(snap.dante_map.get(what, {}))))) else: fix = self.cfg["fixes"][what] log.info("FIX %s: %s", tile_id, fix.get("label", what)) names.append(fix.get("short", what.upper())) - started.append((what, chk)) + started.append((self.executor, (self.cfg, what, chk))) if not names: self.note(tile_id, "JUST TRIED\nWAIT 1 MIN" if waited else "NOTHING\nTO FIX", "idle", now, 4) return plan @@ -832,8 +852,8 @@ class Engine: self.note(tile_id, "FIXING\n" + summary, "fixing", now, 180) results = [] - def work(fix_id, chk): - ok = self.executor(self.cfg, fix_id, chk) + def work(fn, args): + ok = fn(*args) results.append(ok is not False) if len(results) == len(started): # the last one to finish reports for all done = time.time() @@ -841,8 +861,8 @@ class Engine: self.note(tile_id, "CHECKING\n" + summary, "fixing", done, 20) else: self.note(tile_id, "FIX FAILED\n" + summary, "failed", done, 20) - for fix_id, chk in started: - threading.Thread(target=work, args=(fix_id, chk), daemon=True).start() + for fn, args in started: + threading.Thread(target=work, args=(fn, args), daemon=True).start() return plan @@ -1163,6 +1183,90 @@ def hyperdeck_probe(ip, timeout=3, port=9993): return "answered" if buf else "hung" +# --- The self-updating Dante map: name -> IP -> MAC -> switch port -> PoE -------- + +def dante_map_path(data_dir): + return os.path.join(data_dir, "dante-map.json") if data_dir else None + + +def load_dante_map(path): + try: + with open(path) as f: + return json.load(f) + except (OSError, ValueError, TypeError): + return {} + + +def normalize_mac(mac): + return ":".join("%02x" % int(part, 16) for part in mac.split(":")) + + +def arp_table(): + """{ip: mac} from this Mac's ARP cache (macOS prints MACs without leading zeros).""" + out = subprocess.run(["/usr/sbin/arp", "-an"], capture_output=True, text=True, timeout=5).stdout + return {ip: normalize_mac(mac) for ip, mac in re.findall(r"\(([\d.]+)\) at ((?:[0-9a-fA-F]{1,2}:){5}[0-9a-fA-F]{1,2}) ", out)} + + +def dante_label(cfg): + for t in cfg["tiles"]: + for c in t.get("checks", []): + if c["type"] == "dante": + return c["conn"] + return None + + +def poll_dante_map(snap, cfg, comp, path, arp=arp_table, survey=None): + """Every Dante device that is present now: where it is plugged in and whether that port + powers it. Kept for devices that later disappear, so a hold can still find them.""" + switches, label = cfg.get("dante_switches"), dante_label(cfg) + if not switches or not label: + return + # Polled every 30 s so the first map appears soon after start; a full survey + # of the switches only every dante_map_seconds once one has succeeded. + if time.time() - snap.dante_map_time < cfg.get("dante_map_seconds", 600): + return + try: + present = json.loads(snap.var(label, "devices") or "[]") + except ValueError: + return + ips = arp() + wanted = {} + for name in present: + ip = comp.variable(label, name + "_ip") + if ip and ips.get(ip): + wanted[name] = (ip, ips[ip]) + if not wanted: + return + if survey is None: + import netgear + survey = netgear.survey + found = survey(switches, [mac for _, mac in wanted.values()]) + new, now = dict(snap.dante_map), int(time.time()) + for name, (ip, mac) in wanted.items(): + entry = dict(new.get(name, {}), ip=ip, mac=mac, seen=now) + entry.update(found.get(mac, {})) + new[name] = entry + snap.dante_map, snap.dante_map_time = new, time.time() + log.info("dante map: %d devices, %d on PoE ports", len(new), sum(1 for e in new.values() if e.get("poe"))) + if path: + tmp = path + ".tmp" + with open(tmp, "w") as f: + json.dump(new, f, indent=1, sort_keys=True) + os.replace(tmp, path) + + +def execute_poe(cfg, name, entry): + """Power-cycle a Dante device's PoE port, found by its MAC now (or its last known port).""" + try: + import netgear + host, port, how = netgear.poe_cycle_device(cfg.get("dante_switches", []), entry["mac"], entry) + log.info("FIX power-cycled %s on switch %s port %s (%s)", name, host, port, how) + return True + except Exception as e: + log.warning("FIX power-cycle %s failed: %s", name, e) + return False + + def execute_fix(cfg, fix_id, chk): """Run one entry from config "fixes" (in its own thread). Only listed commands ever run. Returns True if the command ran and succeeded.""" @@ -1248,9 +1352,9 @@ def load_config(path): for c in t.get("checks", []): fixes = c.get("fix") if isinstance(c.get("fix"), dict) else {"": c.get("fix")} for f in fixes.values(): - if f and f != RESTART_CONNECTION and f not in cfg.get("fixes", {}): + if f and f not in (RESTART_CONNECTION, DANTE_AUTO) and f not in cfg.get("fixes", {}): raise ValueError("tile %s: unknown fix %s" % (t["id"], f)) - if (f and f != RESTART_CONNECTION and cfg["fixes"][f].get("kind", "ssh") == "ssh" + if (f and f not in (RESTART_CONNECTION, DANTE_AUTO) and cfg["fixes"][f].get("kind", "ssh") == "ssh" and cfg["fixes"][f]["host"] not in cfg.get("hosts", {})): raise ValueError("fix %s: unknown host" % f) if c["type"] == "remote" and c["host"] not in cfg.get("hosts", {}): @@ -1278,9 +1382,9 @@ def make_companion(cfg): return Companion(cfg["companion"].get("host", "127.0.0.1"), cfg["companion"].get("port", 8000)) -def start_pollers(cfg, snap, stop): +def start_pollers(cfg, snap, stop, data_dir=None): fast, slow = needed_variables(cfg) - c1, c2, c3 = make_companion(cfg), make_companion(cfg), make_companion(cfg) + c1, c2, c3, c4 = (make_companion(cfg) for _ in range(4)) iface = cfg.get("network", {}).get("interface") threads = [ Periodic("connections", 1.0, lambda: poll_connections(c1, snap), stop), @@ -1292,6 +1396,7 @@ def start_pollers(cfg, snap, stop): Periodic("bonjour", 15.0, lambda: poll_mdns(snap), stop), Periodic("http", 5.0, lambda: poll_http(snap, http_checks(cfg)), stop), Periodic("remote", 10.0, lambda: poll_remote(snap, cfg), stop), + Periodic("dante-map", 30.0, lambda: poll_dante_map(snap, cfg, c4, dante_map_path(data_dir)), stop), ] for t in threads: t.start() @@ -1335,7 +1440,8 @@ def run(cfg, data_dir, dry_run=False, stop=None): comp = make_companion(cfg) snap = Snapshot() engine = Engine(cfg, os.path.join(data_dir, "service-mode.json")) - threads = start_pollers(cfg, snap, stop) + snap.dante_map = load_dante_map(dante_map_path(data_dir)) + threads = start_pollers(cfg, snap, stop, data_dir) stop.wait(2.5) # let the first polls land before judging anything last_json, last_write, last_eval = None, 0.0, 0.0 down_since, alerted, warned_missing = None, False, set() diff --git a/tests/test_status_panel.py b/tests/test_status_panel.py index ce26901..a5bf928 100644 --- a/tests/test_status_panel.py +++ b/tests/test_status_panel.py @@ -538,28 +538,51 @@ class TestFixes(unittest.TestCase): tiles = self.engine.step(self.snap, time.time()) self.assertEqual((tiles["dante"]["x"], level(tiles["dante"])), ("DANTE\nFIX FAILED\nDVS", "failed")) - def test_studio_b_poe_boxes_are_power_cycled(self): + def test_dante_hold_power_cycles_any_poe_device_from_the_map(self): + devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")]) + devices.remove("STUDIO-B-MON") # PoE box (per the map) + devices.remove("BOH-WING") # console: not PoE + self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices) + self.snap.dante_map = { + "STUDIO-B-MON": {"mac": "00:1d:c1:51:18:84", "switch": "10.66.1.102", "port": "0/15", "shared": 1, "poe": True}, + "BOH-WING": {"mac": "00:1d:c1:03:ec:3a", "switch": "10.66.1.102", "port": "0/3", "shared": 1, "poe": False}, + } + self.assertEqual(self.plan("dante"), [("restart", "audinate-dantecontroller"), ("poe", "STUDIO-B-MON")]) + self.snap.dante_map = {} # not mapped yet: just re-check via the module + self.assertEqual(self.plan("dante"), [("restart", "audinate-dantecontroller")]) + + def test_poe_hold_hands_the_mapped_entry_to_the_switch_code(self): + calls = [] + self.engine.poe_executor = lambda cfg, name, entry: calls.append((name, entry["mac"])) or True devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")]) - devices.remove("STUDIO-B-MON") devices.remove("STUDIO-B-TB") self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices) - self.assertEqual(self.plan("dante"), [("fix", "poe_stb_mon"), ("fix", "poe_stb_tb")]) - for fid in ("poe_stb_solo", "poe_stb_mon", "poe_stb_tb", "poe_stb_tbmon"): - f = CONFIG["fixes"][fid] - self.assertEqual(f["switch"], "10.66.1.102") - self.assertRegex(f["mac"], r"^00:1d:c1:") + self.snap.dante_map = {"STUDIO-B-TB": {"mac": "00:1d:c1:89:1a:0a", "switch": "10.66.1.102", + "port": "0/16", "shared": 1, "poe": True}} + _, t = run_steps(self.engine, self.snap, SUNDAY, 4) + self.engine.run_fixes("dante", self.snap, None, t) + time.sleep(0.2) + self.assertEqual(calls, [("STUDIO-B-TB", "00:1d:c1:89:1a:0a")]) + tiles = self.engine.step(self.snap, time.time()) + self.assertEqual(tiles["dante"]["x"], "DANTE\nCHECKING\nPOWER CYCLE") - def test_netgear_fix_calls_the_switch_with_the_mac_guard(self): - import types - calls = [] - fake = types.ModuleType("netgear") - fake.poe_cycle = lambda host, port, mac=None: calls.append((host, port, mac)) - sys.modules["netgear"] = fake - try: - self.assertTrue(sp.execute_fix(CONFIG, "poe_stb_solo", {})) - finally: - del sys.modules["netgear"] - self.assertEqual(calls, [("10.66.1.102", "0/14", "00:1d:c1:51:34:c8")]) + def test_dante_map_poller(self): + class Comp: + def variable(self, label, name): + return {"STUDIO-B-MON_ip": "169.254.133.24", "NA-Tracks_ip": "169.254.99.70"}.get(name) + self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(["STUDIO-B-MON", "NA-Tracks"]) + self.snap.dante_map = {"BOH-WING": {"mac": "x", "poe": False}} # an older entry is kept + arp = lambda: {"169.254.133.24": "00:1d:c1:51:18:84", "169.254.99.70": "00:e0:4c:ff:b6:8a"} + survey = lambda switches, macs: {"00:1d:c1:51:18:84": {"switch": "10.66.1.102", "port": "0/15", + "shared": 1, "poe": True}} + path = os.path.join(tempfile.mkdtemp(), "dante-map.json") + sp.poll_dante_map(self.snap, CONFIG, Comp(), path, arp=arp, survey=survey) + self.assertEqual(self.snap.dante_map["STUDIO-B-MON"]["port"], "0/15") + self.assertTrue(self.snap.dante_map["STUDIO-B-MON"]["poe"]) + self.assertNotIn("poe", self.snap.dante_map["NA-Tracks"]) # a Mac: not on a PoE port + self.assertIn("BOH-WING", self.snap.dante_map) + self.assertEqual(sp.load_dante_map(path), self.snap.dante_map) + self.assertEqual(sp.normalize_mac("0:1d:c1:89:1a:a"), "00:1d:c1:89:1a:0a") def test_ingest_dvs_and_no_auto_fix(self): devices = json.loads(self.snap.vars[("audinate-dantecontroller", "devices")]) @@ -588,7 +611,7 @@ class TestFixes(unittest.TestCase): def test_every_fix_is_valid(self): for f in CONFIG["fixes"].values(): - self.assertIn(f.get("kind", "ssh"), ("ssh", "local", "hyperdeck", "netgear_poe")) + self.assertIn(f.get("kind", "ssh"), ("ssh", "local", "hyperdeck")) class FakeDeck: @@ -663,6 +686,39 @@ class TestHyperDeckProbe(unittest.TestCase): deck.close() +class TestNetgearParsing(unittest.TestCase): + BIG = """ VLAN ID MAC Address Interface IfIndex Status + 66 00:1D:C1:51:34:C8 0/14 14 Learned + 66 00:1D:C1:51:18:84 0/15 15 Learned + 66 00:E0:4C:9A:1A:F3 0/26 26 Learned + 66 00:0E:DD:E2:8E:C2 0/26 26 Learned + 66 00:0E:DD:E2:8F:60 0/26 26 Learned""" + SMALL = """ 66 00:1D:C1:51:34:C8 0/10 10 Learned + 66 00:1D:C1:51:18:84 0/10 10 Learned + 66 00:E0:4C:9A:1A:F3 0/3 3 Learned""" + + def test_access_port_is_where_the_device_has_fewest_neighbours(self): + import netgear + best = netgear.best_ports({"10.66.1.100": self.SMALL, "10.66.1.102": self.BIG}, + ["00:1d:c1:51:34:c8", "00:e0:4c:9a:1a:f3", "00:0e:dd:e2:8e:c2"]) + self.assertEqual(best["00:1d:c1:51:34:c8"], ("10.66.1.102", "0/14", 1)) # not the uplink 0/10 + self.assertEqual(best["00:e0:4c:9a:1a:f3"], ("10.66.1.100", "0/3", 1)) + self.assertEqual(best["00:0e:dd:e2:8e:c2"][2], 3) # shared port: too many + self.assertGreater(3, netgear.MAX_SHARED) # 3 = uplink + self.assertTrue(netgear.poe_delivering("0/14 Yes 32000 1 1500 29 54 Delivering Power No Error")) + self.assertFalse(netgear.poe_delivering("0/14 Yes 32000 0 0 0 0 Searching No Error")) + + def test_never_cycles_an_uplink(self): + import netgear + orig = netgear.survey + netgear.survey = lambda switches, macs: {macs[0]: {"switch": "s", "port": "0/26", "shared": 3, "poe": False}} + try: + with self.assertRaises(RuntimeError): + netgear.poe_cycle_device(["s"], "00:0e:dd:e2:8e:c2") + finally: + netgear.survey = orig + + class FakeCompanion(BaseHTTPRequestHandler): """Just enough of Companion 5's HTTP API.""" store = {} @@ -777,7 +833,8 @@ class TestAgainstFakeCompanion(unittest.TestCase): store["conns"]["ATEM"]["category"] = "error" store["custom"]["sp_cmd"] = ";ack atem" data_dir = tempfile.mkdtemp() - originals = sp.poll_mdns, sp.poll_system, sp.poll_midi, sp.poll_remote + originals = sp.poll_mdns, sp.poll_system, sp.poll_midi, sp.poll_remote, sp.poll_dante_map + sp.poll_dante_map = lambda *a, **k: None # never log in to a switch from a test sp.poll_mdns = lambda snap, seconds=3: None sp.poll_system = lambda snap, cfg: None sp.poll_midi = lambda snap, cfg: None # CoreMIDI has its own test @@ -797,7 +854,7 @@ class TestAgainstFakeCompanion(unittest.TestCase): finally: stop.set() loop.join(timeout=15) - sp.poll_mdns, sp.poll_system, sp.poll_midi, sp.poll_remote = originals + sp.poll_mdns, sp.poll_system, sp.poll_midi, sp.poll_remote, sp.poll_dante_map = originals store["conns"]["ATEM"]["category"] = "good" self.assertFalse(loop.is_alive()) # it shuts down cleanly self.assertTrue(store["custom"]["sp_hb"].isdigit())