Merge "bgp: Introduce Main router and chassis router commands"

neutron/services/bgp/commands.py

@@ -0,0 +1,296 @@
+# Copyright 2025 Red Hat, Inc.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+
+from oslo_log import log
+from ovsdbapp.backend.ovs_idl import command as ovs_cmd
+from ovsdbapp.backend.ovs_idl import idlutils
+from ovsdbapp.backend.ovs_idl import rowview
+from ovsdbapp.schema.ovn_northbound import commands as nb_cmd
+
+from neutron.conf.services import bgp as bgp_config
+from neutron.services.bgp import constants
+from neutron.services.bgp import exceptions
+from neutron.services.bgp import helpers
+
+LOG = log.getLogger(__name__)
+
+
+def _run_idl_command(cmd, txn):
+    cmd.run_idl(txn)
+    return cmd.result
+
+
+class _LrAddCommand(nb_cmd.LrAddCommand):
+    """An idempotent command to add a logical router.
+
+    We need to subclass the LrAddCommand because it does not check if the
+    columns in the existing row are the same as the columns we are trying to
+    set.
+    """
+    def run_idl(self, txn):
+        try:
+            self.result = self.api.lookup('Logical_Router', self.router)
+        except idlutils.RowNotFound:
+            super().run_idl(txn)
+            self.result = self.api.lookup('Logical_Router', self.router)
+
+        self.set_columns(self.result, **self.columns)
+
+
+class _LrpAddCommand(nb_cmd.LrpAddCommand):
+    """An idempotent command to add a logical router port.
+
+    We need to subclass the LrpAddCommand because it does not check if the
+    columns in the existing row are the same as the columns we are trying to
+    set.
+    """
+    def __init__(
+            self, api, router_name, lrp_name, mac, networks=None, **kwargs):
+        networks = networks or []
+        super().__init__(api, router_name, lrp_name, mac, networks, **kwargs)
+
+    def run_idl(self, txn):
+        try:
+            self.result = self.api.lookup('Logical_Router_Port', self.port)
+        except idlutils.RowNotFound:
+            super().run_idl(txn)
+            self.result = self.api.lookup('Logical_Router_Port', self.port)
+        self.result.mac = self.mac
+        self.result.networks = self.networks
+        self.result.peer = self.peer
+        self.set_columns(self.result, **self.columns)
+
+
+class _HAChassisGroupAddCommand(nb_cmd.HAChassisGroupAddCommand):
+    """An idempotent command to add a HA chassis group.
+
+    We need to subclass the HAChassisGroupAddCommand because it does not check
+    if the columns in the existing row are the same as the columns we are
+    trying to set.
+    """
+    def run_idl(self, txn):
+        try:
+            hcg = self.api.lookup('HA_Chassis_Group', self.name)
+        except idlutils.RowNotFound:
+            super().run_idl(txn)
+            hcg = self.api.lookup('HA_Chassis_Group', self.name)
+
+        self.set_columns(hcg, **self.columns)
+
+        self.result = hcg.uuid
+
+
+class ReconcileRouterCommand(_LrAddCommand):
+    ROUTER_MAC_PREFIX = '00:00'
+
+    def __init__(self, api, name):
+        # We need to set policies and static_routes to empty list because IDL
+        # won't have that set until the transaction is committed
+        super().__init__(
+            api, name, may_exist=True, policies=[], static_routes=[])
+        mac_mgr = helpers.LrpMacManager.get_instance()
+        mac_mgr.register_router(name, self.router_mac_prefix)
+
+    def run_idl(self, txn):
+        super().run_idl(txn)
+
+        for key, value in self.options.items():
+            self.result.setkey('options', key, value)
+
+    @property
+    def options(self):
+        return {}
+
+    @property
+    def router_mac_prefix(self):
+        base_mac = bgp_config.get_bgp_mac_base()
+        return f'{base_mac}:{self.ROUTER_MAC_PREFIX}'
+
+
+class ReconcileMainRouterCommand(ReconcileRouterCommand):
+    ROUTER_MAC_PREFIX = '0b:96'
+
+    def __init__(self, api):
+        name = bgp_config.get_main_router_name()
+        super().__init__(api, name)
+
+    @property
+    def options(self):
+        return {
+            constants.LR_OPTIONS_DYNAMIC_ROUTING: 'true',
+            constants.LR_OPTIONS_DYNAMIC_ROUTING_REDISTRIBUTE:
+                constants.BGP_ROUTER_REDISTRIBUTE,
+            constants.LR_OPTIONS_DYNAMIC_ROUTING_VRF_ID:
+                bgp_config.get_bgp_router_tunnel_key(),
+        }
+
+
+class ReconcileChassisRouterCommand(ReconcileRouterCommand):
+    def __init__(self, api, chassis):
+        self.chassis = chassis
+        router_name = helpers.get_chassis_router_name(self.chassis.name)
+        super().__init__(api, router_name)
+
+    @property
+    def options(self):
+        return {
+            'chassis': self.chassis.name,
+        }
+
+    @property
+    def router_mac_prefix(self):
+        chassis_index = helpers.get_chassis_index(self.chassis)
+        base_mac = bgp_config.get_bgp_mac_base()
+
+        # Two bytes for chassis
+        hex_str = f"{chassis_index:0{4}x}"
+
+        return f'{base_mac}:{hex_str[0:2]}:{hex_str[2:4]}'
+
+
+class IndexAllChassis(ovs_cmd.BaseCommand):
+    def run_idl(self, txn):
+        used_indexes = set()
+        chassis_without_index = []
+
+        for chassis in self.api.tables['Chassis_Private'].rows.values():
+            try:
+                existing_index = int(
+                    chassis.external_ids[constants.OVN_BGP_CHASSIS_INDEX_KEY])
+            except (KeyError, ValueError):
+                chassis_without_index.append(chassis)
+            else:
+                used_indexes.add(existing_index)
+
+        number_of_chassis = len(self.api.tables['Chassis_Private'].rows)
+        available_indexes = set(range(number_of_chassis)) - used_indexes
+        for chassis in chassis_without_index:
+            index = available_indexes.pop()
+            chassis.setkey(
+                'external_ids',
+                constants.OVN_BGP_CHASSIS_INDEX_KEY,
+                str(index),
+            )
+
+        self.result = [
+            rowview.RowView(c)
+            for c in self.api.tables['Chassis_Private'].rows.values()]
+
+
+class ConnectChassisRouterToMainRouterCommand(ovs_cmd.BaseCommand):
+    def __init__(self, api, chassis, hcg):
+        super().__init__(api)
+        self.chassis = chassis
+        self.hcg = hcg
+        self.chassis_index = helpers.get_chassis_index(self.chassis)
+        self.router_name = helpers.get_chassis_router_name(self.chassis.name)
+
+    def validate_prerequisites(self):
+        for router_name in [
+                bgp_config.get_main_router_name(), self.router_name]:
+            try:
+                self.api.lookup('Logical_Router', router_name)
+            except idlutils.RowNotFound:
+                raise exceptions.ReconcileError(
+                    f"Router {router_name} not found")
+
+    def run_idl(self, txn):
+        self.validate_prerequisites()
+
+        mac_mgr = helpers.LrpMacManager.get_instance()
+
+        main_router_name = bgp_config.get_main_router_name()
+
+        lrp_main = helpers.get_lrp_name(main_router_name, self.router_name)
+        lrp_ch = helpers.get_lrp_name(self.router_name, main_router_name)
+
+        lrp_main_mac = mac_mgr.get_mac_address(
+            main_router_name, self.chassis_index)
+        lrp_ch_mac = mac_mgr.get_mac_address(
+            self.router_name, constants.LRP_CHASSIS_TO_MAIN_ROUTER)
+
+        _LrpAddCommand(
+            self.api,
+            self.router_name,
+            lrp_ch,
+            mac=lrp_ch_mac,
+            peer=lrp_main
+        ).run_idl(txn)
+
+        _LrpAddCommand(
+            self.api,
+            main_router_name,
+            lrp_main,
+            mac=lrp_main_mac,
+            peer=lrp_ch,
+            ha_chassis_group=self.hcg,
+            options={
+                constants.LRP_OPTIONS_DYNAMIC_ROUTING_MAINTAIN_VRF: 'true'},
+        ).run_idl(txn)
+
+
+class ReconcileChassisCommand(ovs_cmd.BaseCommand):
+    def __init__(self, api, sb_api, chassis):
+        super().__init__(api)
+        self.sb_api = sb_api
+        self.chassis = chassis
+
+    def run_idl(self, txn):
+        hcg_name = helpers.get_hcg_name(self.chassis.name)
+        hcg = _run_idl_command(_HAChassisGroupAddCommand(
+            self.api,
+            hcg_name), txn)
+
+        nb_cmd.HAChassisGroupAddChassisCommand(
+            self.api,
+            hcg,
+            self.chassis.name, constants.HA_CHASSIS_GROUP_PRIORITY
+        ).run_idl(txn)
+
+        ReconcileChassisRouterCommand(
+            self.api,
+            self.chassis,
+        ).run_idl(txn)
+
+        # Connect chassis router to the main router
+        ConnectChassisRouterToMainRouterCommand(
+            self.api,
+            self.chassis,
+            hcg,
+        ).run_idl(txn)
+
+
+class FullSyncBGPTopologyCommand(ovs_cmd.BaseCommand):
+    def __init__(self, nb_api, sb_api, chassis):
+        super().__init__(nb_api)
+        self.chassis = chassis
+        self.sb_api = sb_api
+
+    def run_idl(self, txn):
+        LOG.debug("BGP full sync topology started")
+        self.reconcile_central(txn)
+        self.reconcile_all_chassis(txn)
+        LOG.debug("BGP full sync topology completed")
+
+    def reconcile_all_chassis(self, txn):
+        for chassis in self.chassis:
+            ReconcileChassisCommand(
+                self.api, self.sb_api, chassis).run_idl(txn)
+
+    def reconcile_central(self, txn):
+        ReconcileMainRouterCommand(
+            self.api,
+        ).run_idl(txn)

neutron/services/bgp/constants.py

@@ -0,0 +1,29 @@
+# Copyright 2025 Red Hat, Inc.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+OVN_BGP_CHASSIS_INDEX_KEY = 'bgp-chassis-index'
+
+LRP_CHASSIS_TO_MAIN_ROUTER = 1
+LRP_MAIN_ROUTER_TO_CHASSIS = 2
+
+BGP_ROUTER_REDISTRIBUTE = 'connected-as-host,nat'
+
+LRP_OPTIONS_DYNAMIC_ROUTING_MAINTAIN_VRF = 'dynamic-routing-maintain-vrf'
+
+LR_OPTIONS_DYNAMIC_ROUTING = 'dynamic-routing'
+LR_OPTIONS_DYNAMIC_ROUTING_REDISTRIBUTE = 'dynamic-routing-redistribute'
+LR_OPTIONS_DYNAMIC_ROUTING_VRF_ID = 'dynamic-routing-vrf-id'
+
+HA_CHASSIS_GROUP_PRIORITY = 10

neutron/services/bgp/exceptions.py

@@ -0,0 +1,19 @@
+# Copyright 2025 Red Hat, Inc.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+class ReconcileError(Exception):
+    def __init__(self, message, resource_type=None):
+        super().__init__(message)
+        self.resource_type = resource_type

neutron/services/bgp/helpers.py

@@ -0,0 +1,125 @@
+# Copyright 2025 Red Hat, Inc.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import netaddr
+from oslo_log import log
+
+from neutron.services.bgp import constants
+from neutron.services.bgp import exceptions
+
+LOG = log.getLogger(__name__)
+
+
+class _BgpRouterMacPrefix:
+    MAC_BYTES = 6
+
+    def __init__(self, mac_prefix):
+        mac_prefix_len = len([byte for byte in mac_prefix.split(':') if byte])
+        if mac_prefix_len >= self.MAC_BYTES:
+            raise ValueError(f"MAC prefix {mac_prefix} is too long")
+        remaining_bytes = self.MAC_BYTES - mac_prefix_len
+        self.mac_prefix = mac_prefix
+        self.max_mac_index = self.calculate_max_mac_generated(remaining_bytes)
+        self.remaining_bytes = remaining_bytes
+
+    @staticmethod
+    def calculate_max_mac_generated(remaining_bytes):
+        """Calculate how many MAC address can be generated
+
+        This means depending on how many bytes are left after the MAC prefix,
+        we can generate 255^n MAC addresses, where n is the number of bytes
+        left.
+
+        For example if a mac prefix is 00:00 - it uses 2 bytes, and MAC address
+        is stored in 6 bytes. That gives us 4 bytes left and hence
+        255^4 MAC addresses.
+        """
+        return 255 ** remaining_bytes - 1
+
+
+class LrpMacManager:
+    def __init__(self):
+        self.known_routers = {}
+
+    @classmethod
+    def get_instance(cls):
+        if not hasattr(cls, "_instance"):
+            cls._instance = cls()
+        return cls._instance
+
+    def register_router(self, router_name, mac_prefix):
+        LOG.debug("Registering router %s with mac prefix %s",
+                  router_name, mac_prefix)
+        self.known_routers[router_name] = _BgpRouterMacPrefix(mac_prefix)
+
+    def get_mac_address(self, router_name, index):
+        try:
+            router = self.known_routers[router_name]
+        except KeyError:
+            raise RuntimeError(f"Router {router_name} not registered")
+
+        if index < 0 or index > router.max_mac_index:
+            raise ValueError(
+                f"Index {index} is out of range, maximum is "
+                f"{router.max_mac_index}")
+
+        # generates the hex string based on the remaining bytes
+        # example: if remaining bytes is 3, and index is 100 will be 000064
+        # because 100 in dec is 64 in hex + 4 zeros to make it 3 bytes
+        hex_str = f"{index:0{router.remaining_bytes * 2}x}"
+
+        # inserts colons between the hex bytes
+        # example: 000064 will be 00:00:64
+        hex_bytes = ':'.join(hex_str[i:i+2] for i in range(0, len(hex_str), 2))
+
+        # combines the mac prefix and the hex bytes into a valid mac address
+        # example: {00:00:00}:{00:00:64} is prefix + hex bytes
+        result = f'{router.mac_prefix}:{hex_bytes}'
+
+        try:
+            netaddr.EUI(result, version=48)
+        except netaddr.core.AddrFormatError:
+            raise ValueError(f"Invalid generated MAC address: {result}")
+
+        return result
+
+
+def get_all_chassis(sb_ovn):
+    chassis = sb_ovn.db_find_rows('Chassis').execute(check_error=True)
+    return chassis
+
+
+# Naming helper functions
+def get_lrp_name(from_name, to_name):
+    return f'bgp-lrp-{from_name}-to-{to_name}'
+
+
+def get_hcg_name(chassis_name):
+    return f'bgp-hcg-{chassis_name}'
+
+
+def get_chassis_router_name(chassis_name):
+    return f'bgp-lr-{chassis_name}'
+
+
+def get_chassis_index(chassis):
+    try:
+        return int(chassis.external_ids[constants.OVN_BGP_CHASSIS_INDEX_KEY])
+    except (KeyError, ValueError):
+        msg = (f"Chassis {chassis.name} has no index required for further "
+               "operations, such as creating chassis BGP resources")
+        LOG.error(msg)
+        # TODO(jlibosva): Use resource types for custom exceptions
+        raise exceptions.ReconcileError(msg)

neutron/services/bgp/reconciler.py

@@ -15,6 +15,8 @@
 
 from oslo_log import log
 
+from neutron.services.bgp import commands
+
 LOG = log.getLogger(__name__)
 
 
@@ -48,7 +50,14 @@ class BGPTopologyReconciler:
     def full_sync(self):
         if not self.nb_ovn.ovsdb_connection.idl.is_lock_contended:
             LOG.info("Full BGP topology synchronization started")
-            # TODO(jlibosva): Implement full sync
+            # First make sure all chassis are indexed
+            chassis = commands.IndexAllChassis(
+                self.sb_ovn).execute(check_error=True)
+            commands.FullSyncBGPTopologyCommand(
+                self.nb_ovn,
+                self.sb_ovn,
+                chassis,
+            ).execute(check_error=True)
             LOG.info(
                 "Full BGP topology synchronization completed successfully")
         else:

neutron/tests/functional/services/bgp/__init__.py

@@ -0,0 +1,110 @@
+# Copyright 2025 Red Hat, Inc.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import os
+import tempfile
+
+from oslo_config import cfg
+from oslo_utils import timeutils
+from ovsdbapp.backend.ovs_idl import connection
+from ovsdbapp import venv
+
+from neutron.conf.plugins.ml2.drivers.ovn import ovn_conf
+from neutron.conf.services import bgp as bgp_config
+from neutron.services.bgp import ovn as bgp_ovn
+from neutron.tests.functional import base as n_base
+from neutron.tests.functional.services.bgp import fixtures
+
+idl_schema_map = {
+    'OVN_Northbound': bgp_ovn.OvnNbIdl,
+    'OVN_Southbound': bgp_ovn.OvnSbIdl,
+}
+
+
+class BaseBgpIDLTestCase(n_base.BaseLoggingTestCase):
+    schemas = []
+
+    def setUp(self):
+        ovn_conf.register_opts()
+        bgp_config.register_opts(cfg.CONF)
+        super().setUp()
+        self.setup_venv()
+        self.create_idls()
+
+    def create_connection(self, schema):
+        idl = idl_schema_map[schema](self._schema_map[schema])
+        return connection.Connection(idl, timeout=10)
+
+    def setup_venv(self):
+        ovsvenv = venv.OvsOvnVenvFixture(
+            tempfile.mkdtemp(),
+            ovsdir=os.getenv('OVS_SRCDIR'),
+            ovndir=os.getenv('OVN_SRCDIR'),
+            remove=True)
+
+        self.useFixture(ovsvenv)
+
+        self._schema_map = {
+            'OVN_Northbound': ovsvenv.ovnnb_connection,
+            'OVN_Southbound': ovsvenv.ovnsb_connection,
+            'Open_vSwitch': ovsvenv.ovs_connection,
+        }
+
+    def create_idls(self):
+        for schema in self.schemas:
+            connection = self.create_connection(schema)
+            if schema == 'OVN_Northbound':
+                self.nb_api = self.useFixture(
+                    fixtures.OvnNbIdlApiFixture(connection)).obj
+            elif schema == 'OVN_Southbound':
+                self.sb_api = self.useFixture(
+                    fixtures.OvnSbIdlApiFixture(connection)).obj
+            elif schema == 'Open_vSwitch':
+                self.ovs_api = self.useFixture(
+                    fixtures.OvsApiFixture(connection)).obj
+
+
+class BaseBgpNbIdlTestCase(BaseBgpIDLTestCase):
+    schemas = ['OVN_Northbound']
+
+
+class BaseBgpSbIdlTestCase(BaseBgpIDLTestCase):
+    schemas = ['OVN_Southbound']
+
+    def setUp(self):
+        bgp_ovn.OvnSbIdl.tables = ('Chassis', 'Encap', 'Chassis_Private')
+        try:
+            super().setUp()
+        finally:
+            bgp_ovn.OvnSbIdl.tables = bgp_ovn.OVN_SB_TABLES
+
+    def add_fake_chassis(self, name, ip, external_ids=None):
+        external_ids = external_ids or {}
+        chassis = self.sb_api.chassis_add(
+            name, ['geneve'], ip).execute(check_error=True)
+
+        nb_cfg_timestamp = timeutils.utcnow_ts() * 1000
+        self.sb_api.db_create(
+            'Chassis_Private', name=name,
+            chassis=chassis.uuid, nb_cfg_timestamp=nb_cfg_timestamp,
+            external_ids=external_ids
+        ).execute(check_error=True)
+
+        return self.sb_api.db_list_rows(
+            'Chassis_Private', [name]).execute(check_error=True)[0]
+
+
+class BaseBgpTestCase(BaseBgpSbIdlTestCase):
+    schemas = ['OVN_Northbound', 'OVN_Southbound']

neutron/tests/functional/services/bgp/fixtures.py

@@ -0,0 +1,31 @@
+# Copyright 2025 Red Hat, Inc.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from ovsdbapp.schema.open_vswitch import impl_idl
+from ovsdbapp.tests.functional.schema import fixtures
+
+from neutron.services.bgp import ovn as bgp_ovn
+
+
+class OvnNbIdlApiFixture(fixtures.ApiImplFixture):
+    api_cls = bgp_ovn.BgpOvnNbIdl
+
+
+class OvnSbIdlApiFixture(fixtures.ApiImplFixture):
+    api_cls = bgp_ovn.BgpOvnSbIdl
+
+
+class OvsApiFixture(fixtures.ApiImplFixture):
+    api_cls = impl_idl.OvsdbIdl

neutron/tests/functional/services/bgp/test_commands.py

@@ -0,0 +1,642 @@
+# Copyright 2025 Red Hat, Inc.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from collections import namedtuple
+
+from oslo_config import cfg
+from oslo_utils import uuidutils
+from ovsdbapp.backend.ovs_idl import idlutils
+
+from neutron.conf.services import bgp as bgp_config
+from neutron.services.bgp import commands
+from neutron.services.bgp import constants
+from neutron.services.bgp import exceptions
+from neutron.services.bgp import helpers
+from neutron.tests.functional.services import bgp
+
+
+def _get_unique_name(prefix="test"):
+    return f"{prefix}_{uuidutils.generate_uuid()[:8]}"
+
+
+def _create_fake_chassis():
+    return _FakeChassis(_get_unique_name("chassis"))
+
+
+class _FakeChassis:
+    def __init__(self, name):
+        self.uuid = uuidutils.generate_uuid()
+        self.name = name
+        self.external_ids = {constants.OVN_BGP_CHASSIS_INDEX_KEY: '5'}
+
+
+class NbCommandsBase(bgp.BaseBgpNbIdlTestCase):
+    def setUp(self):
+        super().setUp()
+        mm = helpers.LrpMacManager.get_instance()
+        mm.known_routers.clear()
+
+
+class BgpCommandsBase(bgp.BaseBgpTestCase):
+    def setUp(self):
+        super().setUp()
+        mm = helpers.LrpMacManager.get_instance()
+        mm.known_routers.clear()
+
+
+class _AddBaseCommand:
+    table = None
+    command = None
+
+    def _assert_table_row_exists(self, name, should_exist=True):
+        try:
+            row = self.nb_api.lookup(self.table, name)
+            if should_exist:
+                self.assertIsNotNone(row)
+                self.assertEqual(row.name, name)
+                return row
+            self.fail(f"{self.table} {name} should not exist")
+        except idlutils.RowNotFound:
+            if should_exist:
+                self.fail(f"{self.table} {name} not found")
+            return None
+
+    def create_row(self, name, **kwargs):
+        self.command(self.nb_api, name, **kwargs).execute(check_error=True)
+
+    def test_create_new_row(self):
+        name = _get_unique_name()
+
+        # Verify row doesn't exist initially
+        self._assert_table_row_exists(name, should_exist=False)
+
+        self.create_row(name)
+
+        # Verify the row was created
+        created_row = self._assert_table_row_exists(name)
+        self.assertEqual(created_row.name, name)
+
+    def test_create_existing_row(self):
+        name = _get_unique_name()
+
+        # Create row first time
+        self.create_row(name)
+
+        # Verify first creation worked
+        row1 = self._assert_table_row_exists(name)
+
+        # Create same row again - should be idempotent
+        self.create_row(name)
+
+        # Lookup should not fail with duplicated name
+        row2 = self._assert_table_row_exists(name)
+        self.assertEqual(row1.uuid, row2.uuid)
+
+    def test_external_ids_update_on_create(self):
+        name = _get_unique_name()
+        self.create_row(name, external_ids={'id1': 'value1'})
+        row = self._assert_table_row_exists(name)
+        self.assertEqual(row.external_ids.get('id1'), 'value1')
+
+        # Create same row again - should be idempotent
+        self.create_row(name, external_ids={'id1': 'value2'})
+        row = self._assert_table_row_exists(name)
+        self.assertEqual(row.external_ids.get('id1'), 'value2')
+
+        # Create with different name - should create a new row
+        name = _get_unique_name()
+        self.create_row(name, external_ids={'id1': 'value3'})
+        row = self._assert_table_row_exists(name)
+
+
+class LrAddCommandTestCase(NbCommandsBase, _AddBaseCommand):
+    table = 'Logical_Router'
+    command = commands._LrAddCommand
+
+
+
+class LrpAddCommandTestCase(NbCommandsBase, _AddBaseCommand):
+    table = 'Logical_Router_Port'
+    command = commands._LrpAddCommand
+
+    def setUp(self):
+        super().setUp()
+        self.lr_name = _get_unique_name()
+        self.nb_api.lr_add(self.lr_name).execute(check_error=True)
+
+    def create_row(self, name, **kwargs):
+        if 'mac' not in kwargs:
+            kwargs['mac'] = '00:00:00:00:00:00'
+        return self.command(
+            self.nb_api, self.lr_name, name, **kwargs).execute(
+                check_error=True)
+
+    def test_create_existing_with_different_attributes(self):
+        name = _get_unique_name()
+        self.create_row(name, mac='00:00:00:00:00:00',
+                        networks=['192.168.1.0/24'], peer='lrp-peer-1')
+        lrp = self._assert_table_row_exists(name)
+        self.assertEqual(lrp.mac, '00:00:00:00:00:00')
+        self.assertEqual(lrp.networks, ['192.168.1.0/24'])
+        self.assertEqual(lrp.peer, ['lrp-peer-1'])
+
+        # Should update the MAC address
+        self.create_row(name, mac='00:00:00:00:00:01',
+                        networks=['192.168.2.0/24'], peer='lrp-peer-2')
+        lrp = self._assert_table_row_exists(name)
+        self.assertEqual(lrp.mac, '00:00:00:00:00:01')
+        self.assertEqual(lrp.networks, ['192.168.2.0/24'])
+        self.assertEqual(lrp.peer, ['lrp-peer-2'])
+
+
+class HAChassisGroupAddCommandTestCase(NbCommandsBase, _AddBaseCommand):
+    table = 'HA_Chassis_Group'
+    command = commands._HAChassisGroupAddCommand
+
+
+class ReconcileRouterCommandTestCase(NbCommandsBase):
+    def _validate_router_created(self, router_name):
+        router = self.nb_api.lr_get(router_name).execute(check_error=True)
+        self.assertEqual(router.name, router_name)
+        mm = helpers.LrpMacManager.get_instance()
+        self.assertIsNotNone(mm.known_routers.get(router_name))
+
+    def test_reconcile_new_router(self):
+        router_name = _get_unique_name()
+
+        commands.ReconcileRouterCommand(
+            self.nb_api, router_name).execute(check_error=True)
+
+        self._validate_router_created(router_name)
+
+    def test_reconcile_existing_router(self):
+        router_name = _get_unique_name()
+
+        self.nb_api.lr_add(router_name).execute(check_error=True)
+
+        commands.ReconcileRouterCommand(
+            self.nb_api, router_name).execute(check_error=True)
+
+        router = self.nb_api.lr_get(router_name).execute(check_error=True)
+        self.assertEqual(router.name, router_name)
+
+    def test_reconcile_router_updates_existing_options(self):
+        router_name = _get_unique_name()
+
+        self.nb_api.lr_add(
+            router_name, options={'wrong-option': 'value'}).execute(
+                check_error=True)
+
+        commands.ReconcileRouterCommand(
+            self.nb_api, router_name).execute(check_error=True)
+
+        router = self.nb_api.lr_get(router_name).execute(check_error=True)
+        self.assertEqual(router.name, router_name)
+
+
+class ReconcileMainRouterCommandTestCase(NbCommandsBase):
+    def setUp(self):
+        super().setUp()
+        self.router_name = _get_unique_name()
+        cfg.CONF.set_override('main_router_name', self.router_name, 'bgp')
+
+    def _validate_main_router_options(self):
+        router = self.nb_api.lr_get(self.router_name).execute(check_error=True)
+        self.assertEqual(router.name, self.router_name)
+        self.assertEqual(router.options.get('dynamic-routing'), 'true')
+        self.assertEqual(router.options.get('dynamic-routing-redistribute'),
+                         'connected-as-host,nat')
+
+    def test_reconcile_main_router_with_dynamic_routing(self):
+        commands.ReconcileMainRouterCommand(
+            self.nb_api).execute(check_error=True)
+
+        self._validate_main_router_options()
+
+    def test_reconcile_updates_existing_main_router_options(self):
+        self.nb_api.lr_add(
+            self.router_name,
+            options={'dynamic-routing': 'false', 'wrong-option': 'value'}
+        ).execute(check_error=True)
+
+        commands.ReconcileMainRouterCommand(
+            self.nb_api).execute(check_error=True)
+
+        self._validate_main_router_options()
+
+    def test_registered_mac_prefix(self):
+        cmd = commands.ReconcileMainRouterCommand(
+            self.nb_api)
+        cmd.execute(check_error=True)
+        mm = helpers.LrpMacManager.get_instance()
+        expected_prefix = cmd.router_mac_prefix
+        self.assertEqual(
+            expected_prefix,
+            mm.known_routers[self.router_name].mac_prefix)
+
+
+class ReconcileChassisRouterCommandTestCase(NbCommandsBase):
+    def test_reconcile_chassis_router(self):
+        chassis = _create_fake_chassis()
+        router_name = helpers.get_chassis_router_name(chassis.name)
+
+        commands.ReconcileChassisRouterCommand(
+            self.nb_api, chassis).execute(check_error=True)
+
+        router = self.nb_api.lr_get(router_name).execute(check_error=True)
+        self.assertEqual(router.name, router_name)
+        self.assertEqual(router.options.get('chassis'), chassis.name)
+
+    def test_reconcile_updates_chassis_router_options(self):
+        chassis = _create_fake_chassis()
+        router_name = helpers.get_chassis_router_name(chassis.name)
+
+        self.nb_api.lr_add(
+            router_name,
+            options={'chassis': 'wrong-chassis', 'other-option': 'value'}
+        ).execute(check_error=True)
+
+        commands.ReconcileChassisRouterCommand(
+            self.nb_api, chassis).execute(check_error=True)
+
+        router = self.nb_api.lr_get(router_name).execute(check_error=True)
+        self.assertEqual(router.options.get('chassis'), chassis.name)
+
+    def test_registered_mac_prefix(self):
+        chassis = _create_fake_chassis()
+        router_name = helpers.get_chassis_router_name(chassis.name)
+        cmd = commands.ReconcileChassisRouterCommand(
+            self.nb_api, chassis)
+        cmd.execute(check_error=True)
+        mm = helpers.LrpMacManager.get_instance()
+        self.assertEqual(
+            cmd.router_mac_prefix,
+            mm.known_routers[router_name].mac_prefix)
+
+
+class IndexAllChassisTestCase(bgp.BaseBgpSbIdlTestCase):
+    def test_index_all_chassis(self):
+        self.add_fake_chassis(_get_unique_name(), '192.168.1.100')
+        self.add_fake_chassis(_get_unique_name(), '192.168.1.101')
+
+        result = commands.IndexAllChassis(self.sb_api).execute(
+            check_error=True)
+
+        expected_indexes = [str(i) for i in range(2)]
+        self.assertCountEqual(
+            expected_indexes,
+            [r.external_ids.get(constants.OVN_BGP_CHASSIS_INDEX_KEY)
+             for r in result])
+
+    def test_index_all_chassis_new_chassis_added(self):
+        self.test_index_all_chassis()
+
+        self.add_fake_chassis(_get_unique_name(), '192.168.1.102')
+
+        result = commands.IndexAllChassis(self.sb_api).execute(
+            check_error=True)
+
+        expected_indexes = [str(i) for i in range(3)]
+        self.assertCountEqual(
+            expected_indexes,
+            [r.external_ids[constants.OVN_BGP_CHASSIS_INDEX_KEY]
+            for r in result])
+
+    def test_index_all_chassis_with_existing_index(self):
+        chassis_names = [_get_unique_name() for _ in range(2)]
+
+        for i, chassis_name in enumerate(chassis_names):
+            self.add_fake_chassis(chassis_name, f'192.168.1.10{i}')
+
+        commands.IndexAllChassis(self.sb_api).execute(check_error=True)
+
+        # remove chassis with index 0
+        self.sb_api.chassis_del(chassis_names[0]).execute(check_error=True)
+
+        for i in range(2):
+            self.add_fake_chassis(_get_unique_name(), f'192.168.1.11{i}')
+
+        result = commands.IndexAllChassis(self.sb_api).execute(
+            check_error=True)
+
+        expected_indexes = [str(i) for i in range(3)]
+        self.assertCountEqual(
+            expected_indexes,
+            [r.external_ids[constants.OVN_BGP_CHASSIS_INDEX_KEY]
+            for r in result])
+
+
+class ConnectChassisRouterToMainRouterCommandTestCase(NbCommandsBase):
+    def setUp(self):
+        super().setUp()
+        self.main_router_name = _get_unique_name()
+        cfg.CONF.set_override('main_router_name', self.main_router_name, 'bgp')
+
+        self.fake_chassis = _create_fake_chassis()
+        self.chassis_router_name = helpers.get_chassis_router_name(
+            self.fake_chassis.name)
+        self.chassis_index = int(
+            self.fake_chassis.external_ids[
+                constants.OVN_BGP_CHASSIS_INDEX_KEY])
+
+        hcg_name = f'bgp-hcg-{self.fake_chassis.name}'
+        self.hcg_id = self._create_hcg(hcg_name)
+
+        commands.ReconcileMainRouterCommand(
+            self.nb_api).execute(check_error=True)
+        commands.ReconcileChassisRouterCommand(
+            self.nb_api,
+            self.fake_chassis).execute(check_error=True)
+
+    def _create_hcg(self, hcg_name):
+        return commands._HAChassisGroupAddCommand(
+            self.nb_api, hcg_name).execute(check_error=True).uuid
+
+    def _validate_connection_created(self):
+        lrp_main_name = helpers.get_lrp_name(
+            self.main_router_name, self.chassis_router_name)
+        lrp_chassis_name = helpers.get_lrp_name(
+            self.chassis_router_name, self.main_router_name)
+
+        lrp_main = self.nb_api.lrp_get(lrp_main_name).execute(check_error=True)
+        lrp_chassis = self.nb_api.lrp_get(
+            lrp_chassis_name).execute(check_error=True)
+
+        # Check ports are connected
+        self.assertEqual(lrp_main.peer, [lrp_chassis_name])
+        self.assertEqual(lrp_chassis.peer, [lrp_main_name])
+
+        # Check MAC addresses
+        mm = helpers.LrpMacManager.get_instance()
+        expected_main_mac = mm.get_mac_address(
+            self.main_router_name, self.chassis_index)
+        expected_chassis_mac = mm.get_mac_address(
+            self.chassis_router_name, constants.LRP_CHASSIS_TO_MAIN_ROUTER)
+
+        self.assertEqual(expected_main_mac, lrp_main.mac)
+        self.assertEqual(expected_chassis_mac, lrp_chassis.mac)
+
+        # Verify main router LRP has HA chassis group
+        self.assertEqual(self.hcg_id, lrp_main.ha_chassis_group[0].uuid)
+
+        # Verify main router LRP has dynamic routing option
+        self.assertEqual(
+            'true', lrp_main.options.get('dynamic-routing-maintain-vrf'))
+
+        return lrp_main_name, lrp_chassis_name
+
+    def test_connect_router_to_main_router_new(self):
+        commands.ConnectChassisRouterToMainRouterCommand(
+            self.nb_api, self.fake_chassis, self.hcg_id
+        ).execute(check_error=True)
+
+        self._validate_connection_created()
+
+    def test_connect_existing_lrps_get_updated(self):
+        lrp_main_name = helpers.get_lrp_name(
+            self.main_router_name, self.chassis_router_name)
+        lrp_chassis_name = helpers.get_lrp_name(
+            self.chassis_router_name, self.main_router_name)
+
+        self.nb_api.lrp_add(
+            self.chassis_router_name, lrp_chassis_name,
+            mac='00:00:00:00:00:01',  # wrong MAC
+            networks=['10.0.0.1/24'],  # wrong IP
+            peer='wrong-peer'  # wrong peer
+        ).execute(check_error=True)
+
+        self.nb_api.lrp_add(
+            self.main_router_name, lrp_main_name,
+            mac='00:00:00:00:00:02',  # wrong MAC
+            networks=['10.0.0.2/24'],  # wrong IP
+            peer='wrong-peer'  # wrong peer
+        ).execute(check_error=True)
+
+        commands.ConnectChassisRouterToMainRouterCommand(
+            self.nb_api, self.fake_chassis, self.hcg_id
+        ).execute(check_error=True)
+
+        self._validate_connection_created()
+
+    def test_connect_router_is_idempotent(self):
+        commands.ConnectChassisRouterToMainRouterCommand(
+            self.nb_api, self.fake_chassis, self.hcg_id
+        ).execute(check_error=True)
+
+        lrp_main1, lrp_chassis1 = self._validate_connection_created()
+
+        commands.ConnectChassisRouterToMainRouterCommand(
+            self.nb_api, self.fake_chassis, self.hcg_id
+        ).execute(check_error=True)
+
+        lrp_main2, lrp_chassis2 = self._validate_connection_created()
+
+        self.assertEqual(lrp_main1, lrp_main2)
+        self.assertEqual(lrp_chassis1, lrp_chassis2)
+
+    def test_connect_router_to_non_existing_main_router(self):
+        self.nb_api.lr_del(self.main_router_name).execute(check_error=True)
+        cmd = commands.ConnectChassisRouterToMainRouterCommand(
+            self.nb_api, self.fake_chassis, self.hcg_id)
+        self.assertRaises(
+            exceptions.ReconcileError,
+            cmd.execute,
+            check_error=True
+        )
+
+    def test_connect_non_existing_router_to_main_router(self):
+        self.nb_api.lr_del(self.chassis_router_name).execute(check_error=True)
+        cmd = commands.ConnectChassisRouterToMainRouterCommand(
+            self.nb_api, self.fake_chassis, self.hcg_id)
+        self.assertRaises(
+            exceptions.ReconcileError,
+            cmd.execute,
+            check_error=True
+        )
+
+
+class ReconcileChassisCommandTestCase(BgpCommandsBase):
+    PeerConnectionAttributes = namedtuple('PeerConnectionAttributes',
+                                          ['lrp_name', 'lrp_ip', 'switch_ip'])
+
+    def setUp(self):
+        super().setUp()
+        self.main_router_name = _get_unique_name()
+        cfg.CONF.set_override('main_router_name', self.main_router_name, 'bgp')
+
+    def _create_chassis(
+            self, name=None, index=None):
+        chassis_name = name or _get_unique_name("chassis")
+
+        chassis_external_ids = {}
+        if index is not None:
+            chassis_external_ids[
+                constants.OVN_BGP_CHASSIS_INDEX_KEY] = str(index)
+
+        return self.add_fake_chassis(
+            chassis_name, f'172.24.4.{index or 0}',
+            external_ids=chassis_external_ids)
+
+    def _validate_hcg_created(self, chassis_name):
+        hcg_name = helpers.get_hcg_name(chassis_name)
+        hcg = self.nb_api.db_find(
+            'HA_Chassis_Group',
+            ('name', '=', hcg_name)
+        ).execute(check_error=True)
+        self.assertTrue(hcg)
+        return hcg[0]
+
+    def _validate_chassis_router_created(self, chassis_name):
+        router_name = helpers.get_chassis_router_name(chassis_name)
+        router = self.nb_api.lr_get(router_name).execute(check_error=True)
+        self.assertEqual(router_name, router.name)
+        self.assertEqual(chassis_name, router.options.get('chassis'))
+        return router
+
+    def _validate_main_router_connection(self, chassis_name):
+        chassis_router_name = helpers.get_chassis_router_name(chassis_name)
+
+        lrp_main_name = helpers.get_lrp_name(
+            self.main_router_name, chassis_router_name)
+        lrp_chassis_name = helpers.get_lrp_name(
+            chassis_router_name, self.main_router_name)
+
+        lrp_main = self.nb_api.lrp_get(lrp_main_name).execute(check_error=True)
+        lrp_chassis = self.nb_api.lrp_get(lrp_chassis_name).execute(
+            check_error=True)
+
+        self.assertEqual([lrp_chassis_name], lrp_main.peer)
+        self.assertEqual([lrp_main_name], lrp_chassis.peer)
+
+    def test_reconcile_chassis_basic(self):
+        chassis = self._create_chassis(index=1)
+
+        # Create main router first (prerequisite)
+        commands.ReconcileMainRouterCommand(self.nb_api).execute(
+            check_error=True)
+
+        commands.ReconcileChassisCommand(
+            self.nb_api, self.sb_api, chassis
+        ).execute(check_error=True)
+
+
+        # Validate all components were created
+        self._validate_hcg_created(chassis.name)
+        self._validate_chassis_router_created(chassis.name)
+        self._validate_main_router_connection(chassis.name)
+
+    def test_reconcile_chassis_idempotent(self):
+        chassis = self._create_chassis(index=1)
+
+        commands.ReconcileMainRouterCommand(self.nb_api).execute(
+            check_error=True)
+
+        cmd = commands.ReconcileChassisCommand(
+            self.nb_api, self.sb_api, chassis)
+        cmd.execute(check_error=True)
+        # Run again to check idempotency
+        cmd.execute(check_error=True)
+
+        self._validate_hcg_created(chassis.name)
+        self._validate_chassis_router_created(chassis.name)
+        self._validate_main_router_connection(chassis.name)
+
+    def test_reconcile_chassis_with_existing_components(self):
+        chassis = self._create_chassis(index=1)
+        hcg_name = helpers.get_hcg_name(chassis.name)
+        router_name = helpers.get_chassis_router_name(chassis.name)
+
+        # Create main router first
+        commands.ReconcileMainRouterCommand(self.nb_api).execute(
+            check_error=True)
+
+        # Pre-create HCG with different settings
+        self.nb_api.ha_chassis_group_add(hcg_name).execute(check_error=True)
+
+        # Pre-create router with wrong chassis
+        self.nb_api.lr_add(
+            router_name,
+            options={'chassis': 'wrong-chassis'}
+        ).execute(check_error=True)
+
+        # Execute command should update existing components
+        commands.ReconcileChassisCommand(
+            self.nb_api, self.sb_api, chassis
+        ).execute(check_error=True)
+
+        # Validate components were updated correctly
+        router = self.nb_api.lr_get(router_name).execute(check_error=True)
+        self.assertEqual(chassis.name, router.options.get('chassis'))
+
+    def test_reconcile_chassis_missing_main_router(self):
+        chassis = self._create_chassis(index=1)
+
+        cmd = commands.ReconcileChassisCommand(
+            self.nb_api, self.sb_api, chassis)
+        self.assertRaises(
+            exceptions.ReconcileError,
+            cmd.execute,
+            check_error=True
+        )
+
+    def test_reconcile_chassis_invalid_index(self):
+        chassis = self._create_chassis(index=1)
+        chassis.external_ids[constants.OVN_BGP_CHASSIS_INDEX_KEY] = 'invalid'
+
+        cmd = commands.ReconcileChassisCommand(
+            self.nb_api, self.sb_api, chassis)
+        self.assertRaises(
+            exceptions.ReconcileError,
+            cmd.execute,
+            check_error=True
+        )
+
+    def test_reconcile_chassis_missing_index(self):
+        chassis = self._create_chassis()
+        cmd = commands.ReconcileChassisCommand(
+            self.nb_api, self.sb_api, chassis)
+
+        self.assertRaises(
+            exceptions.ReconcileError,
+            cmd.execute,
+            check_error=True
+        )
+
+    def test_reconcile_chassis_mac_manager_registration(self):
+        chassis = self._create_chassis(index=1)
+
+        # Create main router first
+        commands.ReconcileMainRouterCommand(self.nb_api).execute(
+            check_error=True)
+
+        commands.ReconcileChassisCommand(
+            self.nb_api, self.sb_api, chassis
+        ).execute(check_error=True)
+
+        # Verify router is registered with MAC manager
+        mm = helpers.LrpMacManager.get_instance()
+        router_name = helpers.get_chassis_router_name(chassis.name)
+        self.assertIn(router_name, mm.known_routers)
+
+        # Verify MAC prefix is correct based on chassis index
+        expected_chassis_index = int(
+            chassis.external_ids[constants.OVN_BGP_CHASSIS_INDEX_KEY])
+        router_info = mm.known_routers[router_name]
+
+        # MAC prefix should be based on chassis index
+        base_mac = bgp_config.get_bgp_mac_base()
+        hex_str = f"{expected_chassis_index:0{4}x}"
+        expected_prefix = f'{base_mac}:{hex_str[0:2]}:{hex_str[2:4]}'
+        self.assertEqual(expected_prefix, router_info.mac_prefix)

neutron/tests/unit/services/bgp/test_helpers.py

@@ -0,0 +1,133 @@
+# Copyright 2025 Red Hat, Inc.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from neutron.services.bgp import helpers
+from neutron.tests import base
+
+
+class LrpMacManagerTestCase(base.BaseTestCase):
+    def setUp(self):
+        super().setUp()
+        self.manager = helpers.LrpMacManager.get_instance()
+        self.manager.known_routers.clear()
+
+    def test_singleton_instance(self):
+        instance2 = helpers.LrpMacManager.get_instance()
+        self.assertIs(self.manager, instance2)
+
+    def test_register_router_valid_prefix(self):
+        router_name = "test-router"
+        mac_prefix = "aa:bb:cc"
+
+        self.manager.register_router(router_name, mac_prefix)
+
+        self.assertIn(router_name, self.manager.known_routers)
+        router = self.manager.known_routers[router_name]
+        self.assertEqual(router.mac_prefix, mac_prefix)
+        # Should have 3 remaining bytes (6 total - 3 prefix)
+        self.assertEqual(router.remaining_bytes, 3)
+        # Max index for 3 bytes is 255^3 - 1
+        self.assertEqual(router.max_mac_index, 255 ** 3 - 1)
+
+    def test_register_router_different_prefix_lengths(self):
+        test_cases = [
+            ("aa", 5, 255 ** 5 - 1),
+            ("aa:bb", 4, 255 ** 4 - 1),
+            ("aa:bb:cc", 3, 255 ** 3 - 1),
+            ("aa:bb:cc:dd", 2, 255 ** 2 - 1),
+            ("aa:bb:cc:dd:ee", 1, 255 ** 1 - 1),
+        ]
+
+        for prefix, expected_remaining, expected_max in test_cases:
+            router_name = f"router-{prefix.replace(':', '')}"
+            self.manager.register_router(router_name, prefix)
+            router = self.manager.known_routers[router_name]
+            self.assertEqual(router.remaining_bytes, expected_remaining)
+            self.assertEqual(router.max_mac_index, expected_max)
+
+    def test_get_mac_address_valid_index(self):
+        router_name = "test-router"
+        mac_prefix = "aa:bb:cc"
+        self.manager.register_router(router_name, mac_prefix)
+
+        mac = self.manager.get_mac_address(router_name, 1)
+        self.assertEqual(mac, "aa:bb:cc:00:00:01")
+
+        mac = self.manager.get_mac_address(router_name, 256)
+        self.assertEqual(mac, "aa:bb:cc:00:01:00")
+
+        mac = self.manager.get_mac_address(router_name, 65536)
+        self.assertEqual(mac, "aa:bb:cc:01:00:00")
+
+    def test_get_mac_address_unregistered_router(self):
+        self.assertRaises(
+            RuntimeError,
+            self.manager.get_mac_address, "nonexistent-router", 1)
+
+    def test_get_mac_address_index_too_large(self):
+        router_name = "test-router"
+        mac_prefix = "aa:bb:cc:dd:ee"  # Only 1 remaining byte
+        self.manager.register_router(router_name, mac_prefix)
+
+        # Max index for 1 byte is 255
+        self.assertRaises(
+            ValueError, self.manager.get_mac_address, router_name, 256)
+
+    def test_get_mac_address_zero_index(self):
+        router_name = "test-router"
+        mac_prefix = "aa:bb:cc"
+        self.manager.register_router(router_name, mac_prefix)
+
+        mac = self.manager.get_mac_address(router_name, 0)
+        self.assertEqual(mac, "aa:bb:cc:00:00:00")
+
+    def test_get_mac_address_formatting(self):
+        router_name = "test-router"
+        mac_prefix = "aa:bb"
+        self.manager.register_router(router_name, mac_prefix)
+
+        # Test various indices to ensure proper formatting
+        test_cases = [
+            (1, "aa:bb:00:00:00:01"),
+            (255, "aa:bb:00:00:00:ff"),
+            (256, "aa:bb:00:00:01:00"),
+            (65535, "aa:bb:00:00:ff:ff"),
+            (65536, "aa:bb:00:01:00:00"),
+        ]
+
+        for index, expected in test_cases:
+            mac = self.manager.get_mac_address(router_name, index)
+            self.assertEqual(mac, expected)
+
+    def test_mac_invalid_index(self):
+        router_name = "test-router"
+        mac_prefix = "aa:bb:cc"
+        self.manager.register_router(router_name, mac_prefix)
+
+        self.assertRaises(
+            ValueError, self.manager.get_mac_address, router_name, -1)
+
+    def test_too_long_mac_prefix(self):
+        router_name = "test-router"
+        mac_prefix = "aa:bb:cc:dd:ee:ff:gg"
+        self.assertRaises(
+            ValueError, self.manager.register_router, router_name, mac_prefix)
+
+    def test_mac_prefix_without_colons(self):
+        router_name = "test-router"
+        mac_prefix = "aa"
+        self.manager.register_router(router_name, mac_prefix)
+        mac = self.manager.get_mac_address(router_name, 1)
+        self.assertEqual("aa:00:00:00:00:01", mac)