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 <noreply@anthropic.com>
This commit is contained in:
Cort BuffingtonandClaude Opus 4.8 committed
1 parent
cbe405c879
commit
b43ea79c48
2 files changed
+240
-1
No files matched your search
@@ -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()
|
||||
@@ -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 = { <obp system> : { <bridge name> : <TGID> } } # 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: <rules>.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())
|
||||
Reference in new issue
Block a user