From b43ea79c48e49741e58f1ea312a58df647ad8c71 Mon Sep 17 00:00:00 2001 From: Cort Buffington Date: Sun, 12 Jul 2026 13:42:08 -0500 Subject: [PATCH] tools: OBP config migration tool (inline members -> OBP_BRIDGES) tools/migrate_obp_rules.py reads hblink.cfg (to learn which systems are OPENBRIDGE) and an old-form rules.py, lifts every inline OBP member into an OBP_BRIDGES table, drops any bridge left empty (OBP_BRIDGES recreates it at load), flags ingress-fork ERRORs and renumber WARNINGs, prints a summary, and writes a new file (never overwrites rules.py). Comments are not preserved. tests/test_obp_bridges.py gains a round-trip proof: an old inline config migrated then loaded the new way yields identical routing indexes, plus fork/renumber/TS-override/empty-bridge cases (74 tests total, all green). Co-Authored-By: Claude Opus 4.8 --- tests/test_obp_bridges.py | 74 +++++++++++++++- tools/migrate_obp_rules.py | 167 +++++++++++++++++++++++++++++++++++++ 2 files changed, 240 insertions(+), 1 deletion(-) create mode 100644 tools/migrate_obp_rules.py diff --git a/tests/test_obp_bridges.py b/tests/test_obp_bridges.py index bea664e..05f97a7 100644 --- a/tests/test_obp_bridges.py +++ b/tests/test_obp_bridges.py @@ -7,13 +7,17 @@ # # Run from the repo root: venv/bin/python -m unittest discover -s tests +import copy import os import sys import unittest -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, _ROOT) +sys.path.insert(0, os.path.join(_ROOT, 'tools')) import bridge +import migrate_obp_rules def _member(system, ts, tgid): @@ -103,5 +107,73 @@ class TestExpandOBPBridges(unittest.TestCase): self.assertEqual(len(bridges['KS-STATEWIDE']), 2) # both still added +class TestMigrateOBPRules(unittest.TestCase): + """The migration tool (tools/migrate_obp_rules.py) must move inline OBP members + into OBP_BRIDGES while preserving the exact routing.""" + + OBP_SYSTEMS = {'VESTA_OBP', 'CC_OBP'} + + def setUp(self): + bridge.CONFIG = {'SYSTEMS': { + 'VESTA_OBP': {'MODE': 'OPENBRIDGE'}, + 'CC_OBP': {'MODE': 'OPENBRIDGE'}, + 'REPEATERS': {'MODE': 'SERVER'}, + 'IPSC': {'MODE': 'SERVER'}, + }} + + def _old_bridges(self): + # Old inline form: OBP systems mixed in as ordinary members. + return { + 'B2': [_member('REPEATERS', 1, 2), _member('IPSC', 1, 2), _member('VESTA_OBP', 1, 2)], + 'B9': [_member('REPEATERS', 1, 9), _member('VESTA_OBP', 1, 9), _member('CC_OBP', 1, 9)], + 'B8': [_member('CC_OBP', 1, 8)], # OBP-only -> should drop from BRIDGES + 'BSW': [_member('REPEATERS', 2, 3120), _member('VESTA_OBP', 1, 3120)], + } + + def _routing_projection(self, bridges, obp_bridges=None): + # {(system, ts, tgid_bytes): sorted[bridge names]} -- identity-independent. + # obp_bridges=None => old inline path (no expand); a dict => new expanded path. + b = copy.deepcopy(bridges) + if obp_bridges is not None: + b = bridge.expand_obp_bridges(b, obp_bridges) + src, _ = bridge.index_bridges(bridge.make_bridges(b)) + return {k: sorted(hit[0] for hit in v) for k, v in src.items()} + + def test_roundtrip_preserves_routing(self): + old = self._old_bridges() + # OLD routing: inline OBP members, no OBP_BRIDGES table. + want = self._routing_projection(old) + + new_bridges, obp_bridges, moved, errors, warnings = migrate_obp_rules.migrate( + copy.deepcopy(old), self.OBP_SYSTEMS) + self.assertFalse(errors) + # NEW routing: trimmed BRIDGES + expanded OBP_BRIDGES. + got = self._routing_projection(new_bridges, obp_bridges) + self.assertEqual(got, want) + + def test_moves_and_drops_obp_only_bridge(self): + new_bridges, obp_bridges, moved, errors, warnings = migrate_obp_rules.migrate( + copy.deepcopy(self._old_bridges()), self.OBP_SYSTEMS) + self.assertEqual(len(moved), 5) # VESTA_OBP x3 + CC_OBP x2 + self.assertNotIn('B8', new_bridges) # OBP-only bridge dropped from BRIDGES + self.assertEqual(obp_bridges['CC_OBP']['B8'], 8) # but present in the table + self.assertEqual(obp_bridges['VESTA_OBP']['B2'], 2) + + def test_ts_override_becomes_tuple(self): + old = {'BSW': [_member('VESTA_OBP', 2, 3120)]} # TS 2 (non-default) + _, obp_bridges, _, _, _ = migrate_obp_rules.migrate(old, {'VESTA_OBP'}) + self.assertEqual(obp_bridges['VESTA_OBP']['BSW'], (3120, 2)) + + def test_detects_ingress_fork(self): + old = {'B1': [_member('VESTA_OBP', 1, 2)], 'B2': [_member('VESTA_OBP', 1, 2)]} + _, _, _, errors, _ = migrate_obp_rules.migrate(old, {'VESTA_OBP'}) + self.assertTrue(errors) + + def test_detects_renumber(self): + old = {'BSW': [_member('VESTA_OBP', 1, 3120), _member('CC_OBP', 1, 8)]} + _, _, _, _, warnings = migrate_obp_rules.migrate(old, {'VESTA_OBP', 'CC_OBP'}) + self.assertTrue(warnings) + + if __name__ == '__main__': unittest.main() diff --git a/tools/migrate_obp_rules.py b/tools/migrate_obp_rules.py new file mode 100644 index 0000000..8319314 --- /dev/null +++ b/tools/migrate_obp_rules.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python +# +# migrate_obp_rules.py -- one-shot migration for the OpenBridge config-model change. +# +# HBlink3 no longer configures OpenBridge (MODE: OPENBRIDGE) systems as inline +# BRIDGES members. They now live in a per-OBP table: +# +# OBP_BRIDGES = { : { : } } # TS 1, or (TGID, TS) +# +# This tool reads your existing hblink.cfg (to learn which systems are OPENBRIDGE) +# and your existing rules.py (old inline form), then: +# * lifts every inline OBP member out of BRIDGES into OBP_BRIDGES, +# * drops any bridge left with no members (OBP_BRIDGES recreates it at load), +# * flags the two problems the new loader would reject/warn on +# - a TGID mapped to >1 bridge on one OBP (ingress fork) -> ERROR +# - a bridge with different TGIDs across OBPs (renumber) -> WARNING +# * prints a summary of what moved, and writes a new rules file. +# +# It never overwrites your rules.py: output goes to a separate file you review +# and rename. Comments and commented-out members in the old file are NOT carried +# over (the tool works on the active, executed config), so re-add any notes by hand. +# +# venv/bin/python tools/migrate_obp_rules.py [--config hblink.cfg] [--rules rules.py] [--out rules.py.migrated] +# +# Exit status is non-zero if an ingress-fork ERROR was found (the output still +# gets written so you can see and fix it). + +import argparse +import configparser +import importlib.util +import os +import sys + + +def _openbridge_systems(cfg_path): + """Section names in hblink.cfg whose MODE is OPENBRIDGE.""" + cp = configparser.ConfigParser() + if not cp.read(cfg_path): + sys.exit('ERROR: could not read config file: {}'.format(cfg_path)) + obp = set() + for section in cp.sections(): + if cp.has_option(section, 'MODE') and cp.get(section, 'MODE').strip().upper() == 'OPENBRIDGE': + obp.add(section) + return obp + + +def _load_rules(rules_path): + spec = importlib.util.spec_from_file_location('rules_old', rules_path) + mod = importlib.util.module_from_spec(spec) + try: + spec.loader.exec_module(mod) + except Exception as e: + sys.exit('ERROR: could not import rules file {}: {}'.format(rules_path, e)) + return mod + + +def _fmt_member(m): + return ("{{'SYSTEM': {!r}, 'TS': {}, 'TGID': {}, 'ACTIVE': {}, " + "'TIMEOUT': {}, 'TO_TYPE': {!r}, 'ON': {}, 'OFF': {}, 'RESET': {}}},").format( + m['SYSTEM'], m['TS'], m['TGID'], m['ACTIVE'], + m.get('TIMEOUT', 2), m.get('TO_TYPE', 'NONE'), + m.get('ON', []), m.get('OFF', []), m.get('RESET', [])) + + +def migrate(bridges, obp_systems): + """Return (new_bridges, obp_bridges, moved, errors, warnings).""" + new_bridges = {} + obp_bridges = {} # obp -> {bridge: tgid_or_(tgid, ts)} + moved = [] # (obp, bridge, tgid, ts) + errors = [] + warnings = [] + tgid_to_bridge = {} # (obp, tgid) -> bridge, ingress-fork check + bridge_obp_tgid = {} # bridge -> {obp: tgid}, renumber check + + for bridge_name, members in bridges.items(): + kept = [] + for m in members: + if m['SYSTEM'] in obp_systems: + tgid, ts = m['TGID'], m.get('TS', 1) + key = (m['SYSTEM'], tgid) + if key in tgid_to_bridge and tgid_to_bridge[key] != bridge_name: + errors.append('OBP "{}" TGID {} maps to more than one bridge ("{}" and "{}") -- ' + 'ingress fork; keep only one before starting HBlink3' + .format(m['SYSTEM'], tgid, tgid_to_bridge[key], bridge_name)) + tgid_to_bridge[key] = bridge_name + obp_bridges.setdefault(m['SYSTEM'], {})[bridge_name] = tgid if ts == 1 else (tgid, ts) + bridge_obp_tgid.setdefault(bridge_name, {})[m['SYSTEM']] = tgid + moved.append((m['SYSTEM'], bridge_name, tgid, ts)) + else: + kept.append(m) + if kept: + new_bridges[bridge_name] = kept # drop bridges left empty (OBP_BRIDGES recreates them) + + for bridge_name, obp_tgids in bridge_obp_tgid.items(): + if len(set(obp_tgids.values())) > 1: + detail = ', '.join('{}={}'.format(o, t) for o, t in sorted(obp_tgids.items())) + warnings.append('bridge "{}" renumbers TGID across OpenBridges ({}) -- legal but unusual' + .format(bridge_name, detail)) + + return new_bridges, obp_bridges, moved, errors, warnings + + +def render(new_bridges, obp_bridges, unit): + L = [] + L.append('# Generated by tools/migrate_obp_rules.py -- REVIEW before use.') + L.append('# OpenBridge members were moved from BRIDGES into OBP_BRIDGES; comments from') + L.append('# the original rules.py were not preserved.') + L.append('') + L.append('BRIDGES = {') + for bridge_name, members in new_bridges.items(): + L.append(' {!r}: ['.format(bridge_name)) + for m in members: + L.append(' ' + _fmt_member(m)) + L.append(' ],') + L.append('}') + L.append('') + L.append('OBP_BRIDGES = {') + for obp, table in obp_bridges.items(): + L.append(' {!r}: {{'.format(obp)) + for bridge_name, val in table.items(): + L.append(' {!r}: {},'.format(bridge_name, val)) + L.append(' },') + L.append('}') + L.append('') + L.append('UNIT = {!r}'.format(list(unit))) + L.append('') + return '\n'.join(L) + + +def main(): + ap = argparse.ArgumentParser(description='Migrate rules.py OpenBridge members to the OBP_BRIDGES table') + ap.add_argument('--config', default='hblink.cfg', help='hblink.cfg to read system MODEs from') + ap.add_argument('--rules', default='rules.py', help='existing (old-form) rules.py') + ap.add_argument('--out', default=None, help='output file (default: .migrated)') + args = ap.parse_args() + + out_path = args.out or (args.rules + '.migrated') + obp_systems = _openbridge_systems(args.config) + rules = _load_rules(args.rules) + bridges = getattr(rules, 'BRIDGES', {}) + unit = getattr(rules, 'UNIT', []) + + if getattr(rules, 'OBP_BRIDGES', None): + print('NOTE: {} already defines OBP_BRIDGES; this looks migrated already.'.format(args.rules)) + + new_bridges, obp_bridges, moved, errors, warnings = migrate(bridges, obp_systems) + + print('OpenBridge systems found in {}: {}'.format(args.config, ', '.join(sorted(obp_systems)) or '(none)')) + print('\n{} inline OBP member(s) moved into OBP_BRIDGES:'.format(len(moved))) + for obp, bridge_name, tgid, ts in moved: + print(' {:<12} {:<14} TGID {}{}'.format(obp, bridge_name, tgid, '' if ts == 1 else ' (TS {})'.format(ts))) + dropped = [b for b in bridges if b not in new_bridges] + if dropped: + print('\nBridges now defined only via OBP_BRIDGES (dropped from BRIDGES): {}'.format(', '.join(dropped))) + for w in warnings: + print('\nWARNING: ' + w) + for e in errors: + print('\nERROR: ' + e) + + with open(out_path, 'w', encoding='utf-8') as f: + f.write(render(new_bridges, obp_bridges, unit)) + print('\nWrote {}. Review it, then replace your rules.py with it.'.format(out_path)) + return 1 if errors else 0 + + +if __name__ == '__main__': + sys.exit(main())