diff --git a/src/openstack_mcp_server/tools/network_tools.py b/src/openstack_mcp_server/tools/network_tools.py index 8e5c337..5f67469 100644 --- a/src/openstack_mcp_server/tools/network_tools.py +++ b/src/openstack_mcp_server/tools/network_tools.py @@ -1,10 +1,15 @@ from fastmcp import FastMCP from .base import get_openstack_conn +from .request.network import ( + ExternalGatewayInfo, + Route, +) from .response.network import ( FloatingIP, Network, Port, + Router, Subnet, ) @@ -42,6 +47,11 @@ def register_tools(self, mcp: FastMCP): mcp.tool()(self.update_floating_ip) mcp.tool()(self.create_floating_ips_bulk) mcp.tool()(self.assign_first_available_floating_ip) + mcp.tool()(self.get_routers) + mcp.tool()(self.create_router) + mcp.tool()(self.get_router_detail) + mcp.tool()(self.update_router) + mcp.tool()(self.delete_router) def get_networks( self, @@ -63,9 +73,9 @@ def get_networks( filters["status"] = status_filter.upper() if shared_only: - filters["shared"] = True + filters["is_shared"] = True - networks = conn.list_networks(filters=filters) + networks = conn.network.networks(**filters) return [ self._convert_to_network_model(network) for network in networks @@ -80,6 +90,7 @@ def create_network( provider_network_type: str | None = None, provider_physical_network: str | None = None, provider_segmentation_id: int | None = None, + project_id: str | None = None, ) -> Network: """ Create a new Network. @@ -107,6 +118,9 @@ def create_network( if provider_network_type: network_args["provider_network_type"] = provider_network_type + if project_id: + network_args["project_id"] = project_id + if provider_physical_network: network_args["provider_physical_network"] = ( provider_physical_network @@ -245,7 +259,7 @@ def get_subnets( filters["project_id"] = project_id if is_dhcp_enabled is not None: filters["enable_dhcp"] = is_dhcp_enabled - subnets = conn.list_subnets(filters=filters) + subnets = conn.network.subnets(**filters) if has_gateway is not None: subnets = [ s for s in subnets if (s.gateway_ip is not None) == has_gateway @@ -440,7 +454,9 @@ def get_ports( filters["device_id"] = device_id if network_id: filters["network_id"] = network_id - ports = conn.list_ports(filters=filters) + + ports = conn.network.ports(**filters) + return [self._convert_to_port_model(port) for port in ports] def get_port_allowed_address_pairs(self, port_id: str) -> list[dict]: @@ -860,3 +876,204 @@ def _convert_to_floating_ip_model(self, openstack_ip) -> FloatingIP: port_id=openstack_ip.port_id, router_id=openstack_ip.router_id, ) + + def get_routers( + self, + status_filter: str | None = None, + project_id: str | None = None, + is_admin_state_up: bool | None = None, + ) -> list[Router]: + """ + Get the list of Routers with optional filtering. + :param status_filter: Filter by router status (e.g., `ACTIVE`, `DOWN`) + :param project_id: Filter by project ID + :param is_admin_state_up: Filter by admin state + :return: List of Router objects + """ + conn = get_openstack_conn() + filters: dict = {} + if status_filter: + filters["status"] = status_filter.upper() + if project_id: + filters["project_id"] = project_id + if is_admin_state_up is not None: + filters["admin_state_up"] = is_admin_state_up + # Do not pass unsupported filters (e.g., status) to the server. + server_filters = self._sanitize_server_filters(filters) + routers = conn.network.routers(**server_filters) + + router_models = [self._convert_to_router_model(r) for r in routers] + if status_filter: + status_upper = status_filter.upper() + router_models = [ + r + for r in router_models + if (r.status or "").upper() == status_upper + ] + return router_models + + def create_router( + self, + name: str | None = None, + description: str | None = None, + is_admin_state_up: bool = True, + is_distributed: bool | None = None, + project_id: str | None = None, + external_gateway_info: ExternalGatewayInfo | None = None, + ) -> Router: + """ + Create a new Router. + Typical use-cases: + - Create basic router: name="r1" (defaults to admin_state_up=True) + - Create distributed router: is_distributed=True + - Create with external gateway for north-south traffic: + external_gateway_info={"network_id": "ext-net", "enable_snat": True, + "external_fixed_ips": [{"subnet_id": "ext-subnet", "ip_address": "203.0.113.10"}]} + - Create with project ownership: project_id="proj-1" + Notes: + - external_gateway_info should follow Neutron schema: at minimum include + "network_id"; optional keys include "enable_snat" and "external_fixed_ips". + :param name: Router name + :param description: Router description + :param is_admin_state_up: Administrative state + :param is_distributed: Distributed router flag + :param project_id: Project ownership + :param external_gateway_info: External gateway info dict + :return: Created Router object + """ + conn = get_openstack_conn() + router_args: dict = {"admin_state_up": is_admin_state_up} + if name is not None: + router_args["name"] = name + if description is not None: + router_args["description"] = description + if is_distributed is not None: + router_args["distributed"] = is_distributed + if project_id is not None: + router_args["project_id"] = project_id + if external_gateway_info is not None: + router_args["external_gateway_info"] = ( + external_gateway_info.model_dump(exclude_none=True) + ) + router = conn.network.create_router(**router_args) + return self._convert_to_router_model(router) + + def get_router_detail(self, router_id: str) -> Router: + """ + Get detailed information about a specific Router. + :param router_id: ID of the router to retrieve + :return: Router details + """ + conn = get_openstack_conn() + router = conn.network.get_router(router_id) + return self._convert_to_router_model(router) + + def update_router( + self, + router_id: str, + name: str | None = None, + description: str | None = None, + is_admin_state_up: bool | None = None, + is_distributed: bool | None = None, + external_gateway_info: ExternalGatewayInfo | None = None, + clear_external_gateway: bool = False, + routes: list[Route] | None = None, + ) -> Router: + """ + Update Router attributes atomically. Only provided parameters are changed; + omitted parameters remain untouched. + Typical use-cases: + - Rename and change description: name="r-new", description="d". + - Toggle admin state: read current via get_router_detail(); pass inverted bool to is_admin_state_up. + - Set distributed flag: is_distributed=True or False. + - Set external gateway: external_gateway_info={"network_id": "ext-net", "enable_snat": True, "external_fixed_ips": [...]}. + - Clear external gateway: clear_external_gateway=True (takes precedence over external_gateway_info). + - Replace static routes: routes=[{"destination": "192.0.2.0/24", "nexthop": "10.0.0.1"}]. Pass [] to remove all routes. + Notes: + - For list-typed fields (routes), the provided list replaces the entire list on the server. + - To clear external gateway, use clear_external_gateway=True. If both provided, clear_external_gateway takes precedence. + :param router_id: ID of the router to update + :param name: New router name + :param description: New router description + :param is_admin_state_up: Administrative state + :param is_distributed: Distributed router flag + :param external_gateway_info: External gateway info dict to set + :param clear_external_gateway: If True, clear external gateway (set to None) + :param routes: Static routes (replaces entire list) + :return: Updated Router object + """ + conn = get_openstack_conn() + update_args: dict = {} + if name is not None: + update_args["name"] = name + if description is not None: + update_args["description"] = description + if is_admin_state_up is not None: + update_args["admin_state_up"] = is_admin_state_up + if is_distributed is not None: + update_args["distributed"] = is_distributed + if clear_external_gateway: + update_args["external_gateway_info"] = None + elif external_gateway_info is not None: + update_args["external_gateway_info"] = ( + external_gateway_info.model_dump(exclude_none=True) + ) + if routes is not None: + update_args["routes"] = [ + r.model_dump(exclude_none=True) for r in routes + ] + if not update_args: + current = conn.network.get_router(router_id) + return self._convert_to_router_model(current) + router = conn.network.update_router(router_id, **update_args) + return self._convert_to_router_model(router) + + def delete_router(self, router_id: str) -> None: + """ + Delete a Router. + :param router_id: ID of the router to delete + :return: None + """ + conn = get_openstack_conn() + conn.network.delete_router(router_id, ignore_missing=False) + return None + + def _convert_to_router_model(self, openstack_router) -> Router: + """ + Convert an OpenStack Router object to a Router pydantic model. + :param openstack_router: OpenStack router object + :return: Pydantic Router model + """ + return Router( + id=openstack_router.id, + name=getattr(openstack_router, "name", None), + status=getattr(openstack_router, "status", None), + description=getattr(openstack_router, "description", None), + project_id=getattr(openstack_router, "project_id", None), + is_admin_state_up=getattr( + openstack_router, "is_admin_state_up", None + ), + external_gateway_info=getattr( + openstack_router, "external_gateway_info", None + ), + is_distributed=getattr(openstack_router, "is_distributed", None), + is_ha=getattr(openstack_router, "is_ha", None), + routes=getattr(openstack_router, "routes", None), + ) + + def _sanitize_server_filters(self, filters: dict) -> dict: + """ + Remove unsupported query params before sending to Neutron. + + Currently removed keys: + - "status": not universally supported for server-side filtering + + :param filters: original filter dict + :return: cleaned filter dict safe for server query + """ + if not filters: + return {} + attrs = dict(filters) + # Remove client-only or unsupported filters + attrs.pop("status", None) + return attrs diff --git a/src/openstack_mcp_server/tools/request/network.py b/src/openstack_mcp_server/tools/request/network.py new file mode 100644 index 0000000..c04182b --- /dev/null +++ b/src/openstack_mcp_server/tools/request/network.py @@ -0,0 +1,26 @@ +from pydantic import BaseModel + + +class Route(BaseModel): + """Static route for a router.""" + + destination: str + nexthop: str + + +class ExternalFixedIP(BaseModel): + """External fixed IP assignment for router gateway.""" + + subnet_id: str | None = None + ip_address: str | None = None + + +class ExternalGatewayInfo(BaseModel): + """External gateway information for a router. + At minimum include `network_id`. Optionally include `enable_snat` and + `external_fixed_ips`. + """ + + network_id: str + enable_snat: bool | None = None + external_fixed_ips: list[ExternalFixedIP] | None = None diff --git a/tests/tools/test_network_tools.py b/tests/tools/test_network_tools.py index c41f2e3..3b06996 100644 --- a/tests/tools/test_network_tools.py +++ b/tests/tools/test_network_tools.py @@ -1,10 +1,15 @@ from unittest.mock import Mock from openstack_mcp_server.tools.network_tools import NetworkTools +from openstack_mcp_server.tools.request.network import ( + ExternalGatewayInfo, + Route, +) from openstack_mcp_server.tools.response.network import ( FloatingIP, Network, Port, + Router, Subnet, ) @@ -49,7 +54,10 @@ def test_get_networks_success( mock_network2.provider_segmentation_id = None mock_network2.project_id = "proj-admin-000" - mock_conn.list_networks.return_value = [mock_network1, mock_network2] + mock_conn.network.networks.return_value = [ + mock_network1, + mock_network2, + ] network_tools = self.get_network_tools() result = network_tools.get_networks() @@ -86,7 +94,7 @@ def test_get_networks_success( assert result[0] == expected_network1 assert result[1] == expected_network2 - mock_conn.list_networks.assert_called_once_with(filters={}) + mock_conn.network.networks.assert_called_once_with() def test_get_networks_empty_list( self, @@ -95,14 +103,14 @@ def test_get_networks_empty_list( """Test getting openstack networks when no networks exist.""" mock_conn = mock_openstack_connect_network - mock_conn.list_networks.return_value = [] + mock_conn.network.networks.return_value = [] network_tools = self.get_network_tools() result = network_tools.get_networks() assert result == [] - mock_conn.list_networks.assert_called_once_with(filters={}) + mock_conn.network.networks.assert_called_once_with() def test_get_networks_with_status_filter( self, @@ -137,7 +145,7 @@ def test_get_networks_with_status_filter( mock_network2.provider_segmentation_id = None mock_network2.project_id = None - mock_conn.list_networks.return_value = [ + mock_conn.network.networks.return_value = [ mock_network1, ] # Only ACTIVE network network_tools = self.get_network_tools() @@ -147,8 +155,8 @@ def test_get_networks_with_status_filter( assert result[0].id == "net-active" assert result[0].status == "ACTIVE" - mock_conn.list_networks.assert_called_once_with( - filters={"status": "ACTIVE"}, + mock_conn.network.networks.assert_called_once_with( + status="ACTIVE", ) def test_get_networks_shared_only( @@ -184,7 +192,7 @@ def test_get_networks_shared_only( mock_network2.provider_segmentation_id = None mock_network2.project_id = None - mock_conn.list_networks.return_value = [ + mock_conn.network.networks.return_value = [ mock_network2, ] # Only shared network @@ -195,10 +203,38 @@ def test_get_networks_shared_only( assert result[0].id == "net-shared" assert result[0].is_shared is True - mock_conn.list_networks.assert_called_once_with( - filters={"shared": True}, + mock_conn.network.networks.assert_called_once_with( + is_shared=True, ) + def test_get_networks_status_filter_case_insensitive( + self, + mock_openstack_connect_network, + ): + mock_conn = mock_openstack_connect_network + + mock_network = Mock() + mock_network.id = "net-active" + mock_network.name = "active-network" + mock_network.status = "ACTIVE" + mock_network.description = None + mock_network.is_admin_state_up = True + mock_network.is_shared = False + mock_network.mtu = None + mock_network.provider_network_type = None + mock_network.provider_physical_network = None + mock_network.provider_segmentation_id = None + mock_network.project_id = None + + mock_conn.network.networks.return_value = [mock_network] + + tools = self.get_network_tools() + res = tools.get_networks(status_filter="active") + + assert len(res) == 1 + assert res[0].status == "ACTIVE" + mock_conn.network.networks.assert_called_once_with(status="ACTIVE") + def test_create_network_success(self, mock_openstack_connect_network): """Test creating a network successfully.""" mock_conn = mock_openstack_connect_network @@ -482,7 +518,7 @@ def test_get_ports_with_filters(self, mock_openstack_connect_network): port.fixed_ips = [{"subnet_id": "subnet-1", "ip_address": "10.0.0.10"}] port.security_group_ids = ["sg-1", "sg-2"] - mock_conn.list_ports.return_value = [port] + mock_conn.network.ports.return_value = [port] tools = self.get_network_tools() result = tools.get_ports( @@ -510,12 +546,10 @@ def test_get_ports_with_filters(self, mock_openstack_connect_network): ), ] - mock_conn.list_ports.assert_called_once_with( - filters={ - "status": "ACTIVE", - "device_id": "device-1", - "network_id": "net-1", - }, + mock_conn.network.ports.assert_called_once_with( + status="ACTIVE", + device_id="device-1", + network_id="net-1", ) def test_create_port_success(self, mock_openstack_connect_network): @@ -565,6 +599,34 @@ def test_create_port_success(self, mock_openstack_connect_network): mock_conn.network.create_port.assert_called_once() + def test_get_ports_status_filter_only( + self, mock_openstack_connect_network + ): + mock_conn = mock_openstack_connect_network + + port = Mock() + port.id = "port-1" + port.name = "p1" + port.status = "DOWN" + port.description = None + port.project_id = None + port.network_id = "net-1" + port.admin_state_up = True + port.is_admin_state_up = True + port.device_id = None + port.device_owner = None + port.mac_address = "fa:16:3e:00:00:03" + port.fixed_ips = [] + port.security_group_ids = None + + mock_conn.network.ports.return_value = [port] + + tools = self.get_network_tools() + res = tools.get_ports(status_filter="down") + assert len(res) == 1 + assert res[0].status == "DOWN" + mock_conn.network.ports.assert_called_once_with(status="DOWN") + def test_get_port_detail_success(self, mock_openstack_connect_network): mock_conn = mock_openstack_connect_network @@ -843,7 +905,7 @@ def test_get_subnets_filters_and_has_gateway_true( subnet2.dns_nameservers = [] subnet2.host_routes = [] - mock_conn.list_subnets.return_value = [subnet1, subnet2] + mock_conn.network.subnets.return_value = [subnet1, subnet2] tools = self.get_network_tools() result = tools.get_subnets( @@ -871,13 +933,11 @@ def test_get_subnets_filters_and_has_gateway_true( host_routes=[], ) - mock_conn.list_subnets.assert_called_once_with( - filters={ - "network_id": "net-1", - "ip_version": 4, - "project_id": "proj-1", - "enable_dhcp": True, - }, + mock_conn.network.subnets.assert_called_once_with( + network_id="net-1", + ip_version=4, + project_id="proj-1", + enable_dhcp=True, ) def test_get_subnets_has_gateway_false( @@ -918,7 +978,7 @@ def test_get_subnets_has_gateway_false( subnet2.dns_nameservers = [] subnet2.host_routes = [] - mock_conn.list_subnets.return_value = [subnet1, subnet2] + mock_conn.network.subnets.return_value = [subnet1, subnet2] tools = self.get_network_tools() result = tools.get_subnets( @@ -1305,3 +1365,207 @@ def test_update_reassign_bulk_and_auto_assign_floating_ip( mock_conn.network.update_ip.return_value = exists auto = tools.assign_first_available_floating_ip("ext-net", "port-9") assert isinstance(auto, FloatingIP) + + def test_get_routers_with_filters(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + + r = Mock() + r.id = "router-1" + r.name = "r1" + r.status = "ACTIVE" + r.description = "desc" + r.project_id = "proj-1" + r.is_admin_state_up = True + r.external_gateway_info = None + r.is_distributed = False + r.is_ha = False + r.routes = [] + + mock_conn.network.routers.return_value = [r] + + tools = self.get_network_tools() + res = tools.get_routers( + status_filter="ACTIVE", + project_id="proj-1", + is_admin_state_up=True, + ) + + assert res == [ + Router( + id="router-1", + name="r1", + status="ACTIVE", + description="desc", + project_id="proj-1", + is_admin_state_up=True, + external_gateway_info=None, + is_distributed=False, + is_ha=False, + routes=[], + ), + ] + + mock_conn.network.routers.assert_called_once_with( + project_id="proj-1", + admin_state_up=True, + ) + + def test_create_router_success(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + + r = Mock() + r.id = "router-1" + r.name = "r1" + r.status = "ACTIVE" + r.description = "desc" + r.project_id = "proj-1" + r.is_admin_state_up = True + r.external_gateway_info = None + r.is_distributed = True + r.is_ha = None + r.routes = [] + mock_conn.network.create_router.return_value = r + + tools = self.get_network_tools() + res = tools.create_router( + name="r1", + description="desc", + is_admin_state_up=True, + is_distributed=True, + project_id="proj-1", + external_gateway_info=ExternalGatewayInfo(network_id="ext-net"), + ) + + assert isinstance(res, Router) + mock_conn.network.create_router.assert_called_once_with( + admin_state_up=True, + name="r1", + description="desc", + distributed=True, + project_id="proj-1", + external_gateway_info={"network_id": "ext-net"}, + ) + + def test_create_router_minimal(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + + r = Mock() + r.id = "router-2" + r.name = None + r.status = "DOWN" + r.description = None + r.project_id = None + r.is_admin_state_up = True + r.external_gateway_info = None + r.is_distributed = None + r.is_ha = None + r.routes = None + mock_conn.network.create_router.return_value = r + + tools = self.get_network_tools() + res = tools.create_router() + assert isinstance(res, Router) + mock_conn.network.create_router.assert_called_once_with( + admin_state_up=True, + ) + + def test_get_router_detail_success(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + + r = Mock() + r.id = "router-3" + r.name = "r3" + r.status = "ACTIVE" + r.description = None + r.project_id = "proj-1" + r.is_admin_state_up = True + r.external_gateway_info = None + r.is_distributed = False + r.is_ha = False + r.routes = [] + mock_conn.network.get_router.return_value = r + + tools = self.get_network_tools() + res = tools.get_router_detail("router-3") + assert res.id == "router-3" + mock_conn.network.get_router.assert_called_once_with("router-3") + + def test_update_router_success(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + + r = Mock() + r.id = "router-4" + r.name = "r4-new" + r.status = "ACTIVE" + r.description = "d-new" + r.project_id = "proj-1" + r.is_admin_state_up = False + r.external_gateway_info = None + r.is_distributed = True + r.is_ha = False + r.routes = [] + mock_conn.network.update_router.return_value = r + + tools = self.get_network_tools() + res = tools.update_router( + router_id="router-4", + name="r4-new", + description="d-new", + is_admin_state_up=False, + is_distributed=True, + external_gateway_info=ExternalGatewayInfo( + network_id="ext-net", enable_snat=True + ), + routes=[ + Route(destination="198.51.100.0/24", nexthop="10.0.0.254") + ], + ) + assert res.name == "r4-new" + mock_conn.network.update_router.assert_called_once_with( + "router-4", + name="r4-new", + description="d-new", + admin_state_up=False, + distributed=True, + external_gateway_info={ + "network_id": "ext-net", + "enable_snat": True, + }, + routes=[ + {"destination": "198.51.100.0/24", "nexthop": "10.0.0.254"} + ], + ) + + def test_update_router_no_fields_returns_current( + self, mock_openstack_connect_network + ): + mock_conn = mock_openstack_connect_network + + current = Mock() + current.id = "router-5" + current.name = "r5" + current.status = "ACTIVE" + current.description = None + current.project_id = None + current.is_admin_state_up = True + current.external_gateway_info = None + current.is_distributed = None + current.is_ha = None + current.routes = None + mock_conn.network.get_router.return_value = current + + tools = self.get_network_tools() + res = tools.update_router("router-5") + assert res.id == "router-5" + + def test_delete_router_success(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + mock_conn.network.delete_router.return_value = None + + tools = self.get_network_tools() + result = tools.delete_router("router-6") + assert result is None + mock_conn.network.delete_router.assert_called_once_with( + "router-6", + ignore_missing=False, + )