Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 35 additions & 0 deletions python/neutron-understack/neutron_understack/tests/test_routers.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
from neutron_understack.routers import add_subport_to_trunk
from neutron_understack.routers import create_port_postcommit
from neutron_understack.routers import fetch_or_create_router_segment
from neutron_understack.routers import fetch_shared_router_port
from neutron_understack.routers import handle_router_interface_removal
from neutron_understack.routers import handle_subport_removal

Expand Down Expand Up @@ -202,3 +203,37 @@ def test_no_router_on_network(self, mocker, port_context):
create_uplink_port.assert_called_once_with(
fake_segment, port_context.current["network_id"]
)


class TestFetchSharedRouterPort:
def test_no_shared_ports_returns_none_without_type_error(self, mocker):
"""LOG.error with set literal {segment, "segment"} raises TypeError.

This test catches the bug where {"segment", segment} (a set) was passed
to LOG.error instead of {"segment": segment} (a dict), causing
TypeError('format requires a mapping') and crashing create_port_postcommit.
"""
fake_segment = mocker.MagicMock()
fake_segment.__getitem__ = mocker.Mock(return_value="seg-123")

mocker.patch("neutron_lib.context.get_admin_context", return_value="admin_ctx")
mocker.patch("neutron.objects.ports.Port.get_objects", return_value=[])

result = fetch_shared_router_port(fake_segment)

assert result is None

def test_returns_first_shared_port_when_found(self, mocker, port_object):
"""Returns first port when shared ports exist."""
fake_segment = mocker.MagicMock()
fake_segment.__getitem__ = mocker.Mock(return_value="seg-123")

mocker.patch("neutron_lib.context.get_admin_context", return_value="admin_ctx")
mocker.patch(
"neutron.objects.ports.Port.get_objects",
return_value=[port_object],
)

result = fetch_shared_router_port(fake_segment)

assert result is port_object
76 changes: 76 additions & 0 deletions python/neutron-understack/neutron_understack/tests/test_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
from sqlalchemy.orm import sessionmaker

from neutron_understack import utils
from neutron_understack.utils import create_neutron_port_for_segment


class TestParentPortIsBound:
Expand Down Expand Up @@ -687,3 +688,78 @@ def test_undersync_default_to_jwt_auth(self, mocker):
assert session.headers["Content-Type"] == "application/json"
assert session.headers["Authorization"] == "Bearer test_token"
assert "X-Auth-Token" not in session.headers


class TestCreateNeutronPortForSegment:
def test_uses_project_id_not_tenant_id(self, mocker):
"""Port dict must use project_id — tenant_id is deprecated in neutron 2026.1."""
segment = {"id": "seg-123"}
project_id = "test-project-id"
context = MagicMock()
context.current = {"network_id": "net-123", "project_id": project_id}
context.plugin_context = MagicMock()

mock_core_plugin = MagicMock()
mock_core_plugin.create_port.return_value = {"id": "new-port-id"}
mocker.patch(
"neutron_understack.utils.directory.get_plugin",
return_value=mock_core_plugin,
)

create_neutron_port_for_segment(segment, context)

call_args = mock_core_plugin.create_port.call_args
port_dict = call_args[0][1]["port"]

assert "tenant_id" not in port_dict, "tenant_id is deprecated, use project_id"
assert "project_id" in port_dict

def test_port_name_uses_segment_id(self, mocker):
"""Port name must be uplink-<segment_id>."""
segment = {"id": "seg-abc-123"}
context = MagicMock()
context.current = {"network_id": "net-123", "project_id": "proj-123"}
context.plugin_context = MagicMock()

mock_core_plugin = MagicMock()
mock_core_plugin.create_port.return_value = {"id": "new-port-id"}
mocker.patch(
"neutron_understack.utils.directory.get_plugin",
return_value=mock_core_plugin,
)

create_neutron_port_for_segment(segment, context)

call_args = mock_core_plugin.create_port.call_args
port_dict = call_args[0][1]["port"]

assert port_dict["name"] == "uplink-seg-abc-123"

def test_rbac_sensitive_fields_are_not_set(self, mocker):
"""device_owner, mac_address, fixed_ips must not be explicitly set.

neutron 2026.1 tightened RBAC on create_port:device_owner,
create_port:mac_address and create_port:fixed_ips — they now require
NET_OWNER_MEMBER or ADMIN role. Passing empty values still triggers
the policy check. Let neutron set defaults instead.
"""
segment = {"id": "seg-123"}
context = MagicMock()
context.current = {"network_id": "net-123", "project_id": "proj-123"}
context.plugin_context = MagicMock()

mock_core_plugin = MagicMock()
mock_core_plugin.create_port.return_value = {"id": "new-port-id"}
mocker.patch(
"neutron_understack.utils.directory.get_plugin",
return_value=mock_core_plugin,
)

create_neutron_port_for_segment(segment, context)

call_args = mock_core_plugin.create_port.call_args
port_dict = call_args[0][1]["port"]

assert "device_owner" not in port_dict, "device_owner triggers RBAC check in 2026.1"
assert "mac_address" not in port_dict, "mac_address triggers RBAC check in 2026.1"
assert "fixed_ips" not in port_dict, "fixed_ips triggers RBAC check in 2026.1"
4 changes: 0 additions & 4 deletions python/neutron-understack/neutron_understack/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,10 +42,6 @@ def create_neutron_port_for_segment(
"port": {
"name": f"uplink-{segment['id']}",
"network_id": context.current["network_id"],
"mac_address": "",
"device_owner": "",
"device_id": "",
"fixed_ips": [],
"admin_state_up": True,
"project_id": "",
}
Expand Down
Loading