test_neutron_understack_mech.py

from dataclasses import dataclass
from types import SimpleNamespace

import pytest
from neutron_lib import constants as p_const
from neutron_lib.api.definitions import portbindings
from neutron_lib.plugins.ml2 import api

from neutron_understack.neutron_understack_mech import UnderstackDriver


def _port_update_context(original, current):
    return SimpleNamespace(
        original=original,
        current=current,
        plugin_context="plugin-context",
    )


class TestUpdatePortPreCommit:
    def test_validates_when_port_becomes_router_interface(
        self, mocker, understack_driver
    ):
        original = {
            "id": "port-a",
            "network_id": "network-a",
            "device_owner": "",
            "device_id": "",
            "fixed_ips": [{"subnet_id": "subnet-a"}],
        }
        current = {
            **original,
            "device_owner": p_const.DEVICE_OWNER_ROUTER_INTF,
            "device_id": "router-a",
        }
        validate = mocker.patch(
            "neutron_understack.neutron_understack_mech."
            "svi_router.validate_svi_router_port",
            return_value=True,
        )

        understack_driver.update_port_precommit(_port_update_context(original, current))

        validate.assert_called_once_with("plugin-context", current)

    def test_validates_when_router_interface_fixed_ips_change(
        self, mocker, understack_driver
    ):
        original = {
            "id": "port-a",
            "network_id": "network-a",
            "device_owner": p_const.DEVICE_OWNER_ROUTER_INTF,
            "device_id": "router-a",
            "fixed_ips": [{"subnet_id": "subnet-a"}],
        }
        current = {
            **original,
            "fixed_ips": [
                {"subnet_id": "subnet-a"},
                {"subnet_id": "subnet-b"},
            ],
        }
        validate = mocker.patch(
            "neutron_understack.neutron_understack_mech."
            "svi_router.validate_svi_router_port",
            return_value=True,
        )

        understack_driver.update_port_precommit(_port_update_context(original, current))

        validate.assert_called_once_with("plugin-context", current)

    def test_validates_remaining_subnets_when_fixed_ip_removed(
        self, mocker, understack_driver
    ):
        original = {
            "id": "port-a",
            "network_id": "network-a",
            "device_owner": p_const.DEVICE_OWNER_ROUTER_INTF,
            "device_id": "router-a",
            "fixed_ips": [
                {"subnet_id": "subnet-a"},
                {"subnet_id": "subnet-b"},
            ],
        }
        current = {
            **original,
            "fixed_ips": [{"subnet_id": "subnet-a"}],
        }
        validate = mocker.patch(
            "neutron_understack.neutron_understack_mech."
            "svi_router.validate_svi_router_port",
            return_value=True,
        )

        understack_driver.update_port_precommit(_port_update_context(original, current))

        validate.assert_called_once_with("plugin-context", current)

    def test_skips_router_interface_without_relevant_update(
        self, mocker, understack_driver
    ):
        original = {
            "id": "port-a",
            "network_id": "network-a",
            "device_owner": p_const.DEVICE_OWNER_ROUTER_INTF,
            "device_id": "router-a",
            "fixed_ips": [{"subnet_id": "subnet-a"}],
        }
        current = {**original, "name": "new-name"}
        validate = mocker.patch(
            "neutron_understack.neutron_understack_mech."
            "svi_router.validate_svi_router_port"
        )

        understack_driver.update_port_precommit(_port_update_context(original, current))

        validate.assert_not_called()

    def test_skips_non_router_interface_update(self, mocker, understack_driver):
        original = {
            "id": "port-a",
            "network_id": "network-a",
            "device_owner": "",
            "device_id": "",
            "fixed_ips": [{"subnet_id": "subnet-a"}],
        }
        current = {**original, "fixed_ips": [{"subnet_id": "subnet-b"}]}
        validate = mocker.patch(
            "neutron_understack.neutron_understack_mech."
            "svi_router.validate_svi_router_port"
        )

        understack_driver.update_port_precommit(_port_update_context(original, current))

        validate.assert_not_called()


class TestUpdatePortPostCommit:
    def test_with_simple_port(self, understack_driver, port_context):
        understack_driver.update_port_postcommit(port_context)

        understack_driver.undersync.sync.assert_called_once()

    def test_skips_non_baremetal_port(self, understack_driver, port_context):
        port_context.current[portbindings.VNIC_TYPE] = portbindings.VNIC_NORMAL

        understack_driver.update_port_postcommit(port_context)

        understack_driver.undersync.sync.assert_not_called()


class TestDeletePortPostCommit:
    def test_skips_non_baremetal_port(self, understack_driver, port_context):
        port_context.current[portbindings.VNIC_TYPE] = portbindings.VNIC_NORMAL

        understack_driver.delete_port_postcommit(port_context)

        understack_driver.undersync.sync.assert_not_called()


@pytest.mark.usefixtures("_ironic_baremetal_port_physical_network")
class TestBindPort:
    def test_does_not_bind_vlan_only_segments(
        self,
        mocker,
        port_context,
        understack_driver,
        vlan_network_segment,
    ):
        """At level 1 understack receives only the VLAN segment and should do nothing.

        undersync is responsible for that binding.
        """
        port_context._prepare_to_bind([vlan_network_segment])
        mocker.patch.object(port_context, "continue_binding")

        understack_driver.bind_port(port_context)

        port_context.continue_binding.assert_not_called()

    def test_uses_existing_vlan_segment(
        self,
        mocker,
        port_context,
        understack_driver,
        vlan_network_segment,
    ):
        """When a VLAN segment already exists for the physnet, reuse it.

        It should not allocate a new dynamic segment.
        """
        mocker.patch.object(port_context, "allocate_dynamic_segment")
        mocker.patch.object(port_context, "continue_binding")
        mocker.patch(
            "neutron_understack.neutron_understack_mech.utils.vlan_segment_for_physnet",
            return_value=vlan_network_segment,
        )
        port_context._prepare_to_bind(port_context.network.network_segments)

        understack_driver.bind_port(port_context)

        port_context.allocate_dynamic_segment.assert_not_called()
        vxlan_segment = next(
            s
            for s in port_context.network.network_segments
            if s[api.NETWORK_TYPE] == "vxlan"
        )
        port_context.continue_binding.assert_called_once_with(
            segment_id=vxlan_segment[api.ID],
            next_segments_to_bind=[vlan_network_segment],
        )

    def test_with_no_trunk(
        self,
        mocker,
        port_context,
        understack_driver,
        understack_trunk_driver,
        vlan_network_segment,
    ):
        mocker.patch.object(
            port_context, "allocate_dynamic_segment", return_value=vlan_network_segment
        )
        mocker.patch.object(port_context, "continue_binding")
        port_context._prepare_to_bind(port_context.network.network_segments)

        understack_driver.bind_port(port_context)
        understack_driver.trunk_driver = understack_trunk_driver

        port_context.allocate_dynamic_segment.assert_called_once()
        vxlan_segment = next(
            s
            for s in port_context.network.network_segments
            if s[api.NETWORK_TYPE] == "vxlan"
        )
        port_context.continue_binding.assert_called_once_with(
            segment_id=vxlan_segment[api.ID],
            next_segments_to_bind=[vlan_network_segment],
        )

    def test_refuses_unsupported_vnic_type(
        self, mocker, port_context, understack_driver
    ):
        port_context.current[portbindings.VNIC_TYPE] = portbindings.VNIC_DIRECT
        mocker.patch.object(port_context, "continue_binding")
        port_context._prepare_to_bind(port_context.network.network_segments)

        understack_driver.bind_port(port_context)

        port_context.continue_binding.assert_not_called()

    def test_fails_when_physical_network_missing(
        self, mocker, port_context, understack_driver
    ):
        """Port binding fails when physical_network is absent from binding_profile.

        This enforces that physical_network must be present for binding to succeed.
        """
        # Remove physical_network from binding_profile to test hard requirement
        binding_profile_dict = port_context.current[portbindings.PROFILE].copy()
        binding_profile_dict.pop("physical_network", None)
        port_context.current[portbindings.PROFILE] = binding_profile_dict

        mocker.patch.object(port_context, "continue_binding")
        port_context._prepare_to_bind(port_context.network.network_segments)

        understack_driver.bind_port(port_context)

        # Binding should not proceed without physical_network
        port_context.continue_binding.assert_not_called()

    @pytest.mark.parametrize("port_dict", [{"trunk": True}], indirect=True)
    def test_with_trunk_details(
        self, mocker, understack_driver, port_context, understack_trunk_driver
    ):
        mocker.patch(
            "neutron_understack.utils.fetch_subport_network_id", return_value="112233"
        )
        mocker.patch.object(port_context, "continue_binding")
        port_context._prepare_to_bind(port_context.network.network_segments)

        understack_driver.trunk_driver = understack_trunk_driver
        mocker.patch.object(understack_driver.trunk_driver, "configure_trunk")
        understack_driver.bind_port(port_context)
        understack_driver.trunk_driver.configure_trunk.assert_called_once()
        port_context.continue_binding.assert_called_once()


class TestCreateNetworkPostCommit:
    @pytest.mark.usefixtures("_ml2_understack_conf")
    def test_vxlan_network(
        self,
        mocker,
        understack_driver,
        network_context,
        network_id,
        ucvni_group_id,
        project_id,
    ):
        @dataclass
        class FakeContext:
            current = {
                "id": "3b5f0bb1-cd53-4c71-b129-1fe7550dfdf4",
                "name": "humpback",
                "tenant_id": "f9b40d4a39c4403ab5567da17e71906a",
                "admin_state_up": True,
                "mtu": 9000,
                "status": "ACTIVE",
                "subnets": [],
                "standard_attr_id": 3926,
                "shared": False,
                "project_id": "f9b40d4a39c4403ab5567da17e71906a",
                "router:external": False,
                "provider:network_type": "vxlan",
                "provider:physical_network": None,
                "provider:segmentation_id": 200025,
                "is_default": False,
                "availability_zone_hints": [],
                "availability_zones": [],
                "ipv4_address_scope": None,
                "ipv6_address_scope": None,
                "vlan_transparent": None,
                "description": "",
                "l2_adjacency": True,
                "tags": [],
                "created_at": "2025-03-14T07:06:52Z",
                "updated_at": "2025-03-14T07:06:52Z",
                "revision_number": 1,
            }
            network_segments = [
                {
                    "id": "9e56eb8d-f9ec-47d2-ac80-3fde76087c38",
                    "network_type": "vxlan",
                    "physical_network": None,
                    "segmentation_id": 200025,
                    "network_id": "3b5f0bb1-cd53-4c71-b129-1fe7550dfdf4",
                }
            ]

        understack_driver.create_network_postcommit(FakeContext())


class TestKeystoneAuthentication:
    def test_initialize_with_keystone_auth(self, mocker):
        """Test that Undersync creates its own session using the ironic auth config."""
        mock_session_instance = mocker.MagicMock()
        mock_get_session = mocker.patch(
            "neutron_understack.config.get_session",
            return_value=mock_session_instance,
        )
        mocker.patch("neutron_understack.neutron_understack_mech.IronicClient")

        driver = UnderstackDriver()
        driver.initialize()

        mock_get_session.assert_called_once_with("ironic")
        assert driver.undersync._session == mock_session_instance