diff --git a/src/openstack_mcp_server/tools/network_tools.py b/src/openstack_mcp_server/tools/network_tools.py index f873fad..8e5c337 100644 --- a/src/openstack_mcp_server/tools/network_tools.py +++ b/src/openstack_mcp_server/tools/network_tools.py @@ -1,18 +1,22 @@ from fastmcp import FastMCP -from openstack_mcp_server.tools.response.network import Network - from .base import get_openstack_conn +from .response.network import ( + FloatingIP, + Network, + Port, + Subnet, +) class NetworkTools: """ - A class to encapsulate Neutron-related tools and utilities. + A class to encapsulate Network-related tools and utilities. """ def register_tools(self, mcp: FastMCP): """ - Register Neutron-related tools with the FastMCP instance. + Register Network-related tools with the FastMCP instance. """ mcp.tool()(self.get_networks) @@ -20,6 +24,24 @@ def register_tools(self, mcp: FastMCP): mcp.tool()(self.get_network_detail) mcp.tool()(self.update_network) mcp.tool()(self.delete_network) + mcp.tool()(self.get_subnets) + mcp.tool()(self.create_subnet) + mcp.tool()(self.get_subnet_detail) + mcp.tool()(self.update_subnet) + mcp.tool()(self.delete_subnet) + mcp.tool()(self.get_ports) + mcp.tool()(self.create_port) + mcp.tool()(self.get_port_detail) + mcp.tool()(self.update_port) + mcp.tool()(self.delete_port) + mcp.tool()(self.get_port_allowed_address_pairs) + mcp.tool()(self.set_port_binding) + mcp.tool()(self.get_floating_ips) + mcp.tool()(self.create_floating_ip) + mcp.tool()(self.delete_floating_ip) + mcp.tool()(self.update_floating_ip) + mcp.tool()(self.create_floating_ips_bulk) + mcp.tool()(self.assign_first_available_floating_ip) def get_networks( self, @@ -27,14 +49,11 @@ def get_networks( shared_only: bool = False, ) -> list[Network]: """ - Get the list of Neutron networks with optional filtering. - - Args: - status_filter: Filter networks by status (e.g., 'ACTIVE', 'DOWN') - shared_only: If True, only show shared networks + Get the list of Networks with optional filtering. - Returns: - List of Network objects + :param status_filter: Filter networks by status (e.g., `ACTIVE`, `DOWN`) + :param shared_only: If True, only show shared networks + :return: List of Network objects """ conn = get_openstack_conn() @@ -63,19 +82,16 @@ def create_network( provider_segmentation_id: int | None = None, ) -> Network: """ - Create a new Neutron network. - - Args: - name: Network name - description: Network description - is_admin_state_up: Administrative state - is_shared: Whether the network is shared - provider_network_type: Provider network type (e.g., 'vlan', 'flat', 'vxlan') - provider_physical_network: Physical network name - provider_segmentation_id: Segmentation ID for VLAN/VXLAN - - Returns: - Created Network object + Create a new Network. + + :param name: Network name + :param description: Network description + :param is_admin_state_up: Administrative state + :param is_shared: Whether the network is shared + :param provider_network_type: Provider network type (e.g., 'vlan', 'flat', 'vxlan') + :param provider_physical_network: Physical network name + :param provider_segmentation_id: Segmentation ID for VLAN/VXLAN + :return: Created Network object """ conn = get_openstack_conn() @@ -105,20 +121,14 @@ def create_network( def get_network_detail(self, network_id: str) -> Network: """ - Get detailed information about a specific Neutron network. - - Args: - network_id: ID of the network to retrieve + Get detailed information about a specific Network. - Returns: - Network object with detailed information + :param network_id: ID of the network to retrieve + :return: Network details """ conn = get_openstack_conn() network = conn.network.get_network(network_id) - if not network: - raise Exception(f"Network with ID {network_id} not found") - return self._convert_to_network_model(network) def update_network( @@ -130,17 +140,14 @@ def update_network( is_shared: bool | None = None, ) -> Network: """ - Update an existing Neutron network. - - Args: - network_id: ID of the network to update - name: New network name - description: New network description - is_admin_state_up: New administrative state - is_shared: New shared state - - Returns: - Updated Network object + Update an existing Network. + + :param network_id: ID of the network to update + :param name: New network name + :param description: New network description + :param is_admin_state_up: New administrative state + :param is_shared: New shared state + :return: Updated Network object """ conn = get_openstack_conn() @@ -156,28 +163,19 @@ def update_network( update_args["shared"] = is_shared if not update_args: - raise Exception("No update parameters provided") - + current = conn.network.get_network(network_id) + return self._convert_to_network_model(current) network = conn.network.update_network(network_id, **update_args) - return self._convert_to_network_model(network) def delete_network(self, network_id: str) -> None: """ - Delete a Neutron network. - - Args: - network_id: ID of the network to delete + Delete a Network. - Returns: - None + :param network_id: ID of the network to delete + :return: None """ conn = get_openstack_conn() - - network = conn.network.get_network(network_id) - if not network: - raise Exception(f"Network with ID {network_id} not found") - conn.network.delete_network(network_id, ignore_missing=False) return None @@ -186,11 +184,8 @@ def _convert_to_network_model(self, openstack_network) -> Network: """ Convert an OpenStack network object to a Network pydantic model. - Args: - openstack_network: OpenStack network object - - Returns: - Network pydantic model + :param openstack_network: OpenStack network object + :return: Pydantic Network model """ return Network( id=openstack_network.id, @@ -208,3 +203,660 @@ def _convert_to_network_model(self, openstack_network) -> Network: or None, project_id=openstack_network.project_id or None, ) + + def get_subnets( + self, + network_id: str | None = None, + ip_version: int | None = None, + project_id: str | None = None, + has_gateway: bool | None = None, + is_dhcp_enabled: bool | None = None, + ) -> list[Subnet]: + """ + Get the list of Subnets with optional filtering. + + Use this to narrow results by network, project, IP version, gateway presence, and + DHCP-enabled state. + + Notes: + - has_gateway is applied client-side after retrieval and checks whether `gateway_ip` is set. + - `is_dhcp_enabled` maps to Neutron's `enable_dhcp` filter. + - Combining filters further restricts the result (logical AND). + + Examples: + - All IPv4 subnets in a network: `network_id="net-1"`, `ip_version=4` + - Only subnets with a gateway: `has_gateway=True` + - DHCP-enabled subnets for a project: `project_id="proj-1"`, `is_dhcp_enabled=True` + + :param network_id: Filter by network ID + :param ip_version: Filter by IP version (e.g., 4, 6) + :param project_id: Filter by project ID + :param has_gateway: True for subnets with a gateway, False for no gateway + :param is_dhcp_enabled: True for DHCP-enabled subnets, False for disabled + :return: List of Subnet objects + """ + conn = get_openstack_conn() + filters: dict = {} + if network_id: + filters["network_id"] = network_id + if ip_version is not None: + filters["ip_version"] = ip_version + if project_id: + filters["project_id"] = project_id + if is_dhcp_enabled is not None: + filters["enable_dhcp"] = is_dhcp_enabled + subnets = conn.list_subnets(filters=filters) + if has_gateway is not None: + subnets = [ + s for s in subnets if (s.gateway_ip is not None) == has_gateway + ] + return [self._convert_to_subnet_model(subnet) for subnet in subnets] + + def create_subnet( + self, + network_id: str, + cidr: str, + name: str | None = None, + ip_version: int = 4, + gateway_ip: str | None = None, + is_dhcp_enabled: bool = True, + description: str | None = None, + dns_nameservers: list[str] | None = None, + allocation_pools: list[dict] | None = None, + host_routes: list[dict] | None = None, + ) -> Subnet: + """ + Create a new Subnet. + + :param network_id: ID of the parent network + :param cidr: Subnet CIDR + :param name: Subnet name + :param ip_version: IP version + :param gateway_ip: Gateway IP address + :param is_dhcp_enabled: Whether DHCP is enabled + :param description: Subnet description + :param dns_nameservers: DNS nameserver list + :param allocation_pools: Allocation pool list + :param host_routes: Static host routes + :return: Created Subnet object + """ + conn = get_openstack_conn() + subnet_args: dict = { + "network_id": network_id, + "cidr": cidr, + "ip_version": ip_version, + "enable_dhcp": is_dhcp_enabled, + } + if name is not None: + subnet_args["name"] = name + if description is not None: + subnet_args["description"] = description + if gateway_ip is not None: + subnet_args["gateway_ip"] = gateway_ip + if dns_nameservers is not None: + subnet_args["dns_nameservers"] = dns_nameservers + if allocation_pools is not None: + subnet_args["allocation_pools"] = allocation_pools + if host_routes is not None: + subnet_args["host_routes"] = host_routes + subnet = conn.network.create_subnet(**subnet_args) + return self._convert_to_subnet_model(subnet) + + def get_subnet_detail(self, subnet_id: str) -> Subnet: + """ + Get detailed information about a specific Subnet. + + :param subnet_id: ID of the subnet to retrieve + :return: Subnet details + """ + conn = get_openstack_conn() + subnet = conn.network.get_subnet(subnet_id) + return self._convert_to_subnet_model(subnet) + + def update_subnet( + self, + subnet_id: str, + name: str | None = None, + description: str | None = None, + gateway_ip: str | None = None, + clear_gateway: bool = False, + is_dhcp_enabled: bool | None = None, + dns_nameservers: list[str] | None = None, + allocation_pools: list[dict] | None = None, + host_routes: list[dict] | None = None, + ) -> Subnet: + """ + Update subnet attributes atomically. Only provided parameters are changed; omitted + parameters remain untouched. + + Typical use-cases: + - Set gateway: `gateway_ip="10.0.0.1"`. + - Clear gateway: `clear_gateway=True`. + - Enable/disable DHCP: `is_dhcp_enabled=True or False`. + - Batch updates: update name/description and DNS nameservers together. + + Notes: + - `clear_gateway=True` explicitly clears `gateway_ip` (sets to None). If both `gateway_ip` + and `clear_gateway=True` are provided, `clear_gateway` takes precedence. + - For list-typed fields (`dns_nameservers`, `allocation_pools`, `host_routes`), the provided + list replaces the entire list on the server. Pass `[]` to remove all entries. + - For a DHCP toggle, read the current value via `get_subnet_detail()` and pass the inverted + boolean to `is_dhcp_enabled`. + + Examples: + - Clear the gateway and disable DHCP: `clear_gateway=True`, `is_dhcp_enabled=False` + - Replace DNS servers: `dns_nameservers=["8.8.8.8", "1.1.1.1"]` + + :param subnet_id: ID of the subnet to update + :param name: New subnet name + :param description: New subnet description + :param gateway_ip: New gateway IP + :param clear_gateway: If True, clear the gateway IP (sets to None) + :param is_dhcp_enabled: DHCP enabled state + :param dns_nameservers: DNS nameserver list (replaces entire list) + :param allocation_pools: Allocation pool list (replaces entire list) + :param host_routes: Static host routes (replaces entire list) + :return: Updated Subnet 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 clear_gateway: + update_args["gateway_ip"] = None + elif gateway_ip is not None: + update_args["gateway_ip"] = gateway_ip + if is_dhcp_enabled is not None: + update_args["enable_dhcp"] = is_dhcp_enabled + if dns_nameservers is not None: + update_args["dns_nameservers"] = dns_nameservers + if allocation_pools is not None: + update_args["allocation_pools"] = allocation_pools + if host_routes is not None: + update_args["host_routes"] = host_routes + if not update_args: + current = conn.network.get_subnet(subnet_id) + return self._convert_to_subnet_model(current) + subnet = conn.network.update_subnet(subnet_id, **update_args) + return self._convert_to_subnet_model(subnet) + + def delete_subnet(self, subnet_id: str) -> None: + """ + Delete a Subnet. + + :param subnet_id: ID of the subnet to delete + :return: None + """ + conn = get_openstack_conn() + conn.network.delete_subnet(subnet_id, ignore_missing=False) + return None + + def _convert_to_subnet_model(self, openstack_subnet) -> Subnet: + """ + Convert an OpenStack subnet object to a Subnet pydantic model. + + :param openstack_subnet: OpenStack subnet object + :return: Pydantic Subnet model + """ + return Subnet( + id=openstack_subnet.id, + name=openstack_subnet.name, + status=getattr(openstack_subnet, "status", None), + description=openstack_subnet.description, + project_id=openstack_subnet.project_id, + network_id=openstack_subnet.network_id, + cidr=openstack_subnet.cidr, + ip_version=openstack_subnet.ip_version, + gateway_ip=openstack_subnet.gateway_ip, + is_dhcp_enabled=openstack_subnet.is_dhcp_enabled, + allocation_pools=getattr( + openstack_subnet, "allocation_pools", None + ), + dns_nameservers=getattr(openstack_subnet, "dns_nameservers", None), + host_routes=getattr(openstack_subnet, "host_routes", None), + ) + + def get_ports( + self, + status_filter: str | None = None, + device_id: str | None = None, + network_id: str | None = None, + ) -> list[Port]: + """ + Get the list of Ports with optional filtering. + + :param status_filter: Filter by port status (e.g., `ACTIVE`, `DOWN`) + :param device_id: Filter by device ID + :param network_id: Filter by network ID + :return: List of Port objects + """ + conn = get_openstack_conn() + filters: dict = {} + if status_filter: + filters["status"] = status_filter.upper() + if device_id: + filters["device_id"] = device_id + if network_id: + filters["network_id"] = network_id + ports = conn.list_ports(filters=filters) + return [self._convert_to_port_model(port) for port in ports] + + def get_port_allowed_address_pairs(self, port_id: str) -> list[dict]: + """ + Get allowed address pairs configured on a port. + + :param port_id: Port ID + :return: Allowed address pairs + """ + conn = get_openstack_conn() + port = conn.network.get_port(port_id) + return list(port.allowed_address_pairs or []) + + def set_port_binding( + self, + port_id: str, + host_id: str | None = None, + vnic_type: str | None = None, + profile: dict | None = None, + ) -> Port: + """ + Set binding attributes for a port. + + :param port_id: Port ID + :param host_id: Binding host ID + :param vnic_type: VNIC type + :param profile: Binding profile + :return: Updated Port object + """ + conn = get_openstack_conn() + update_args: dict = {} + if host_id is not None: + update_args["binding_host_id"] = host_id + if vnic_type is not None: + update_args["binding_vnic_type"] = vnic_type + if profile is not None: + update_args["binding_profile"] = profile + if not update_args: + current = conn.network.get_port(port_id) + return self._convert_to_port_model(current) + updated = conn.network.update_port(port_id, **update_args) + return self._convert_to_port_model(updated) + + def create_port( + self, + network_id: str, + name: str | None = None, + description: str | None = None, + is_admin_state_up: bool = True, + device_id: str | None = None, + fixed_ips: list[dict] | None = None, + security_group_ids: list[str] | None = None, + ) -> Port: + """ + Create a new Port. + + :param network_id: ID of the parent network + :param name: Port name + :param description: Port description + :param is_admin_state_up: Administrative state + :param device_id: Device ID + :param fixed_ips: Fixed IP list + :param security_group_ids: Security group ID list + :return: Created Port object + """ + conn = get_openstack_conn() + port_args: dict = { + "network_id": network_id, + "admin_state_up": is_admin_state_up, + } + if name is not None: + port_args["name"] = name + if description is not None: + port_args["description"] = description + if device_id is not None: + port_args["device_id"] = device_id + if fixed_ips is not None: + port_args["fixed_ips"] = fixed_ips + if security_group_ids is not None: + port_args["security_groups"] = security_group_ids + port = conn.network.create_port(**port_args) + return self._convert_to_port_model(port) + + def get_port_detail(self, port_id: str) -> Port: + """ + Get detailed information about a specific Port. + + :param port_id: ID of the port to retrieve + :return: Port details + """ + conn = get_openstack_conn() + port = conn.network.get_port(port_id) + return self._convert_to_port_model(port) + + def update_port( + self, + port_id: str, + name: str | None = None, + description: str | None = None, + is_admin_state_up: bool | None = None, + device_id: str | None = None, + security_group_ids: list[str] | None = None, + allowed_address_pairs: list[dict] | None = None, + fixed_ips: list[dict] | None = None, + ) -> Port: + """ + Update an existing Port. Only provided parameters are changed; omitted parameters remain untouched. + + Typical use-cases: + - Set admin state down: is_admin_state_up=False + - Toggle admin state: read current via get_port_detail(); pass inverted value + - Replace security groups: security_group_ids=["sg-1", "sg-2"] + - Replace allowed address pairs: + 1) current = get_port_allowed_address_pairs(port_id) + 2) edit the list (append/remove dicts) + 3) update_port(port_id, allowed_address_pairs=current) + - Replace fixed IPs: + 1) current = get_port_detail(port_id).fixed_ips + 2) edit the list + 3) update_port(port_id, fixed_ips=current) + + Notes: + - List-typed fields (security groups, allowed address pairs, fixed IPs) replace the entire list + with the provided value. Pass [] to remove all entries. + - For fixed IPs, each dict typically includes keys like "subnet_id" and/or "ip_address". + + Examples: + - Add a fixed IP: read current, append a new {"subnet_id": "subnet-2", "ip_address": "10.0.1.10"}, + then pass fixed_ips=[...] + - Clear all security groups: security_group_ids=[] + + :param port_id: ID of the port to update + :param name: New port name + :param description: New port description + :param is_admin_state_up: Administrative state + :param device_id: Device ID + :param security_group_ids: Security group ID list (replaces entire list) + :param allowed_address_pairs: Allowed address pairs (replaces entire list) + :param fixed_ips: Fixed IP assignments (replaces entire list) + :return: Updated Port 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 device_id is not None: + update_args["device_id"] = device_id + if security_group_ids is not None: + update_args["security_groups"] = security_group_ids + if allowed_address_pairs is not None: + update_args["allowed_address_pairs"] = allowed_address_pairs + if fixed_ips is not None: + update_args["fixed_ips"] = fixed_ips + if not update_args: + current = conn.network.get_port(port_id) + return self._convert_to_port_model(current) + port = conn.network.update_port(port_id, **update_args) + return self._convert_to_port_model(port) + + def delete_port(self, port_id: str) -> None: + """ + Delete a Port. + + :param port_id: ID of the port to delete + :return: None + """ + conn = get_openstack_conn() + conn.network.delete_port(port_id, ignore_missing=False) + return None + + def _convert_to_port_model(self, openstack_port) -> Port: + """ + Convert an OpenStack Port object to a Port pydantic model. + + :param openstack_port: OpenStack port object + :return: Pydantic Port model + """ + return Port( + id=openstack_port.id, + name=openstack_port.name, + status=openstack_port.status, + description=openstack_port.description, + project_id=openstack_port.project_id, + network_id=openstack_port.network_id, + is_admin_state_up=openstack_port.is_admin_state_up, + device_id=openstack_port.device_id, + device_owner=openstack_port.device_owner, + mac_address=openstack_port.mac_address, + fixed_ips=openstack_port.fixed_ips, + security_group_ids=openstack_port.security_group_ids + if hasattr(openstack_port, "security_group_ids") + else None, + ) + + def get_floating_ips( + self, + status_filter: str | None = None, + project_id: str | None = None, + port_id: str | None = None, + floating_network_id: str | None = None, + unassigned_only: bool | None = None, + ) -> list[FloatingIP]: + """ + Get the list of Floating IPs with optional filtering. + + :param status_filter: Filter by IP status (e.g., `ACTIVE`) + :param project_id: Filter by project ID + :param port_id: Filter by attached port ID + :param floating_network_id: Filter by external network ID + :param unassigned_only: If True, return only unassigned IPs + :return: List of FloatingIP objects + """ + conn = get_openstack_conn() + filters: dict = {} + if status_filter: + filters["status"] = status_filter.upper() + if project_id: + filters["project_id"] = project_id + if port_id: + filters["port_id"] = port_id + if floating_network_id: + filters["floating_network_id"] = floating_network_id + ips = list(conn.network.ips(**filters)) + if unassigned_only: + ips = [i for i in ips if not i.port_id] + return [self._convert_to_floating_ip_model(ip) for ip in ips] + + def create_floating_ip( + self, + floating_network_id: str, + description: str | None = None, + fixed_ip_address: str | None = None, + port_id: str | None = None, + project_id: str | None = None, + ) -> FloatingIP: + """ + Create a new Floating IP. + + Typical use-cases: + - Allocate in a pool and attach immediately: provide port_id (and optionally fixed_ip_address). + - Allocate for later use: omit port_id (unassigned state). + - Add metadata: provide description. + + :param floating_network_id: External (floating) network ID + :param description: Floating IP description (omit to keep empty) + :param fixed_ip_address: Internal fixed IP to map when attaching to a port + :param port_id: Port ID to attach (omit for unassigned allocation) + :param project_id: Project ID to assign ownership + :return: Created FloatingIP object + """ + conn = get_openstack_conn() + ip_args: dict = {"floating_network_id": floating_network_id} + if description is not None: + ip_args["description"] = description + if fixed_ip_address is not None: + ip_args["fixed_ip_address"] = fixed_ip_address + if port_id is not None: + ip_args["port_id"] = port_id + if project_id is not None: + ip_args["project_id"] = project_id + ip = conn.network.create_ip(**ip_args) + return self._convert_to_floating_ip_model(ip) + + def attach_floating_ip_to_port( + self, + floating_ip_id: str, + port_id: str, + fixed_ip_address: str | None = None, + ) -> FloatingIP: + """ + Attach a Floating IP to a Port. + + :param floating_ip_id: Floating IP ID + :param port_id: Port ID to attach + :param fixed_ip_address: Specific fixed IP on the port (optional) + :return: Updated Floating IP object + """ + conn = get_openstack_conn() + update_args: dict = {"port_id": port_id} + if fixed_ip_address is not None: + update_args["fixed_ip_address"] = fixed_ip_address + ip = conn.network.update_ip(floating_ip_id, **update_args) + return self._convert_to_floating_ip_model(ip) + + def update_floating_ip( + self, + floating_ip_id: str, + description: str | None = None, + port_id: str | None = None, + fixed_ip_address: str | None = None, + clear_port: bool = False, + ) -> FloatingIP: + """ + Update Floating IP attributes. Only provided parameters are changed; omitted + parameters remain untouched. + + Typical use-cases: + - Attach to a port: port_id="port-1" (optionally fixed_ip_address="10.0.0.10"). + - Detach from its port: clear_port=True and omit port_id (sets port_id=None). + - Keep current port: clear_port=False and omit port_id. + - Update description: description="new desc" or clear with description=None. + - Reassign to another port: port_id="new-port" (optionally with fixed_ip_address). + + Notes: + - Passing None for description clears it. + - clear_port controls whether to detach when no port_id is provided. + - fixed_ip_address is optional and can be provided alongside port_id. + + :param floating_ip_id: Floating IP ID to update + :param description: New description (omit to keep unchanged, None to clear) + :param port_id: Port ID to attach; omit to keep or detach depending on clear_port + :param clear_port: If True and port_id is omitted, detach (set port_id=None); if False and + port_id is omitted, keep current attachment + :param fixed_ip_address: Specific fixed IP to map when attaching + :return: Updated FloatingIP object + """ + conn = get_openstack_conn() + update_args: dict = {} + if description is not None: + update_args["description"] = description + if port_id is not None: + update_args["port_id"] = port_id + if fixed_ip_address is not None: + update_args["fixed_ip_address"] = fixed_ip_address + else: + if clear_port: + update_args["port_id"] = None + if not update_args: + current = conn.network.get_ip(floating_ip_id) + return self._convert_to_floating_ip_model(current) + ip = conn.network.update_ip(floating_ip_id, **update_args) + return self._convert_to_floating_ip_model(ip) + + def delete_floating_ip(self, floating_ip_id: str) -> None: + """ + Delete a Floating IP. + + :param floating_ip_id: Floating IP ID to delete + :return: None + """ + conn = get_openstack_conn() + conn.network.delete_ip(floating_ip_id, ignore_missing=False) + return None + + def create_floating_ips_bulk( + self, + floating_network_id: str, + count: int, + ) -> list[FloatingIP]: + """ + Create multiple floating IPs on the specified external network. + + :param floating_network_id: External network ID + :param count: Number of floating IPs to create (negative treated as 0) + :return: List of created FloatingIP objects + """ + conn = get_openstack_conn() + created = [] + for _ in range(max(0, count)): + ip = conn.network.create_ip( + floating_network_id=floating_network_id, + ) + created.append(self._convert_to_floating_ip_model(ip)) + return created + + def assign_first_available_floating_ip( + self, + floating_network_id: str, + port_id: str, + ) -> FloatingIP: + """ + Assign the first available floating IP from a network to a port. + If none are available, create a new one and assign it. + + :param floating_network_id: External network ID + :param port_id: Target port ID + :return: Updated FloatingIP object + """ + conn = get_openstack_conn() + existing = list( + conn.network.ips(floating_network_id=floating_network_id), + ) + available = next( + (i for i in existing if not i.port_id), + None, + ) + if available is None: + created = conn.network.create_ip( + floating_network_id=floating_network_id, + ) + target_id = created.id + else: + target_id = available.id + ip = conn.network.update_ip(target_id, port_id=port_id) + return self._convert_to_floating_ip_model(ip) + + def _convert_to_floating_ip_model(self, openstack_ip) -> FloatingIP: + """ + Convert an OpenStack floating IP object to a FloatingIP pydantic model. + + :param openstack_ip: OpenStack floating IP object + :return: Pydantic FloatingIP model + """ + return FloatingIP( + id=openstack_ip.id, + name=openstack_ip.name, + status=openstack_ip.status, + description=openstack_ip.description, + project_id=openstack_ip.project_id, + floating_ip_address=openstack_ip.floating_ip_address, + floating_network_id=openstack_ip.floating_network_id, + fixed_ip_address=openstack_ip.fixed_ip_address, + port_id=openstack_ip.port_id, + router_id=openstack_ip.router_id, + ) diff --git a/src/openstack_mcp_server/tools/response/network.py b/src/openstack_mcp_server/tools/response/network.py index e8747b9..6894bd5 100644 --- a/src/openstack_mcp_server/tools/response/network.py +++ b/src/openstack_mcp_server/tools/response/network.py @@ -17,35 +17,81 @@ class Network(BaseModel): class Subnet(BaseModel): id: str - name: str - status: str + name: str | None = None + status: str | None = None + description: str | None = None + project_id: str | None = None + network_id: str | None = None + cidr: str | None = None + ip_version: int | None = None + gateway_ip: str | None = None + is_dhcp_enabled: bool | None = None + allocation_pools: list[dict] | None = None + dns_nameservers: list[str] | None = None + host_routes: list[dict] | None = None class Port(BaseModel): id: str - name: str - status: str + name: str | None = None + status: str | None = None + description: str | None = None + project_id: str | None = None + network_id: str | None = None + is_admin_state_up: bool | None = None + device_id: str | None = None + device_owner: str | None = None + mac_address: str | None = None + fixed_ips: list[dict] | None = None + security_group_ids: list[str] | None = None class Router(BaseModel): id: str - name: str - status: str + name: str | None = None + status: str | None = None + description: str | None = None + project_id: str | None = None + is_admin_state_up: bool | None = None + external_gateway_info: dict | None = None + is_distributed: bool | None = None + is_ha: bool | None = None + routes: list[dict] | None = None class SecurityGroup(BaseModel): id: str - name: str - status: str + name: str | None = None + status: str | None = None + description: str | None = None + project_id: str | None = None + security_group_rule_ids: list[str] | None = None class SecurityGroupRule(BaseModel): id: str - name: str - status: str + name: str | None = None + status: str | None = None + description: str | None = None + project_id: str | None = None + direction: str | None = None + ethertype: str | None = None + protocol: str | None = None + port_range_min: int | None = None + port_range_max: int | None = None + remote_ip_prefix: str | None = None + remote_group_id: str | None = None + security_group_id: str | None = None class FloatingIP(BaseModel): id: str - name: str - status: str + name: str | None = None + status: str | None = None + description: str | None = None + project_id: str | None = None + floating_ip_address: str | None = None + floating_network_id: str | None = None + fixed_ip_address: str | None = None + port_id: str | None = None + router_id: str | None = None diff --git a/tests/tools/test_network_tools.py b/tests/tools/test_network_tools.py index c659a17..c41f2e3 100644 --- a/tests/tools/test_network_tools.py +++ b/tests/tools/test_network_tools.py @@ -1,9 +1,12 @@ from unittest.mock import Mock -import pytest - from openstack_mcp_server.tools.network_tools import NetworkTools -from openstack_mcp_server.tools.response.network import Network +from openstack_mcp_server.tools.response.network import ( + FloatingIP, + Network, + Port, + Subnet, +) class TestNetworkTools: @@ -338,23 +341,6 @@ def test_get_network_detail_success(self, mock_openstack_connect_network): mock_conn.network.get_network.assert_called_once_with("net-detail-123") - def test_get_network_detail_not_found( - self, - mock_openstack_connect_network, - ): - """Test getting network detail when network not found.""" - mock_conn = mock_openstack_connect_network - - mock_conn.network.get_network.return_value = None - - network_tools = self.get_network_tools() - - with pytest.raises( - Exception, - match="Network with ID nonexistent-net not found", - ): - network_tools.get_network_detail("nonexistent-net") - def test_update_network_success(self, mock_openstack_connect_network): """Test updating a network successfully.""" mock_conn = mock_openstack_connect_network @@ -459,20 +445,6 @@ def test_update_network_partial_update( **expected_args, ) - def test_update_network_no_parameters( - self, - mock_openstack_connect_network, - ): - """Test updating a network with no parameters provided.""" - mock_conn = mock_openstack_connect_network - - network_tools = self.get_network_tools() - - with pytest.raises(Exception, match="No update parameters provided"): - network_tools.update_network("net-123") - - mock_conn.network.update_network.assert_not_called() - def test_delete_network_success(self, mock_openstack_connect_network): """Test deleting a network successfully.""" mock_conn = mock_openstack_connect_network @@ -480,7 +452,6 @@ def test_delete_network_success(self, mock_openstack_connect_network): mock_network = Mock() mock_network.name = "network-to-delete" - mock_conn.network.get_network.return_value = mock_network mock_conn.network.delete_network.return_value = None network_tools = self.get_network_tools() @@ -488,24 +459,849 @@ def test_delete_network_success(self, mock_openstack_connect_network): assert result is None - mock_conn.network.get_network.assert_called_once_with("net-delete-123") mock_conn.network.delete_network.assert_called_once_with( "net-delete-123", ignore_missing=False, ) - def test_delete_network_not_found(self, mock_openstack_connect_network): - """Test deleting a network when network not found.""" + def test_get_ports_with_filters(self, mock_openstack_connect_network): mock_conn = mock_openstack_connect_network - mock_conn.network.get_network.return_value = None + port = Mock() + port.id = "port-1" + port.name = "p1" + port.status = "ACTIVE" + port.description = None + port.project_id = "proj-1" + port.network_id = "net-1" + port.admin_state_up = True + port.is_admin_state_up = True + port.device_id = "device-1" + port.device_owner = "compute:nova" + port.mac_address = "fa:16:3e:00:00:01" + 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] + + tools = self.get_network_tools() + result = tools.get_ports( + status_filter="ACTIVE", + device_id="device-1", + network_id="net-1", + ) - network_tools = self.get_network_tools() + assert result == [ + Port( + id="port-1", + name="p1", + status="ACTIVE", + description=None, + project_id="proj-1", + network_id="net-1", + is_admin_state_up=True, + device_id="device-1", + device_owner="compute:nova", + mac_address="fa:16:3e:00:00:01", + fixed_ips=[ + {"subnet_id": "subnet-1", "ip_address": "10.0.0.10"}, + ], + security_group_ids=["sg-1", "sg-2"], + ), + ] + + mock_conn.list_ports.assert_called_once_with( + filters={ + "status": "ACTIVE", + "device_id": "device-1", + "network_id": "net-1", + }, + ) + + def test_create_port_success(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 = "desc" + port.project_id = "proj-1" + 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:02" + port.fixed_ips = [] + port.security_group_ids = ["sg-1"] + + mock_conn.network.create_port.return_value = port + + tools = self.get_network_tools() + result = tools.create_port( + network_id="net-1", + name="p1", + description="desc", + is_admin_state_up=True, + fixed_ips=[], + security_group_ids=["sg-1"], + ) + + assert result == Port( + id="port-1", + name="p1", + status="DOWN", + description="desc", + project_id="proj-1", + network_id="net-1", + is_admin_state_up=True, + device_id=None, + device_owner=None, + mac_address="fa:16:3e:00:00:02", + fixed_ips=[], + security_group_ids=["sg-1"], + ) + + mock_conn.network.create_port.assert_called_once() + + def test_get_port_detail_success(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + + port = Mock() + port.id = "port-1" + port.name = "p1" + port.status = "ACTIVE" + 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.get_port.return_value = port + + tools = self.get_network_tools() + result = tools.get_port_detail("port-1") + assert result.id == "port-1" + mock_conn.network.get_port.assert_called_once_with("port-1") + + def test_update_port_success(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + + port = Mock() + port.id = "port-1" + port.name = "p-new" + port.status = "ACTIVE" + port.description = "d-new" + port.project_id = None + port.network_id = "net-1" + port.admin_state_up = False + port.is_admin_state_up = False + port.device_id = "dev-2" + port.device_owner = None + port.mac_address = "fa:16:3e:00:00:04" + port.fixed_ips = [] + port.security_group_ids = ["sg-2"] + + mock_conn.network.update_port.return_value = port + + tools = self.get_network_tools() + res = tools.update_port( + port_id="port-1", + name="p-new", + description="d-new", + is_admin_state_up=False, + device_id="dev-2", + security_group_ids=["sg-2"], + ) + assert res.name == "p-new" + mock_conn.network.update_port.assert_called_once_with( + "port-1", + name="p-new", + description="d-new", + admin_state_up=False, + device_id="dev-2", + security_groups=["sg-2"], + ) - with pytest.raises( - Exception, - match="Network with ID nonexistent-net not found", - ): - network_tools.delete_network("nonexistent-net") + def test_delete_port_success(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + + port = Mock() + port.id = "port-1" + mock_conn.network.delete_port.return_value = None + + tools = self.get_network_tools() + result = tools.delete_port("port-1") + assert result is None + mock_conn.network.delete_port.assert_called_once_with( + "port-1", + ignore_missing=False, + ) + + def test_add_port_fixed_ip(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + + current = Mock() + current.fixed_ips = [ + {"subnet_id": "subnet-1", "ip_address": "10.0.0.10"}, + ] + mock_conn.network.get_port.return_value = current + + updated = Mock() + updated.id = "port-1" + updated.name = "p1" + updated.status = "ACTIVE" + updated.description = None + updated.project_id = None + updated.network_id = "net-1" + updated.admin_state_up = True + updated.is_admin_state_up = True + updated.device_id = None + updated.device_owner = None + updated.mac_address = "fa:16:3e:00:00:05" + updated.fixed_ips = [ + {"subnet_id": "subnet-1", "ip_address": "10.0.0.10"}, + {"subnet_id": "subnet-2", "ip_address": "10.0.1.10"}, + ] + updated.security_group_ids = None + mock_conn.network.update_port.return_value = updated + + tools = self.get_network_tools() + new_fixed = list(current.fixed_ips) + new_fixed.append({"subnet_id": "subnet-2", "ip_address": "10.0.1.10"}) + res = tools.update_port("port-1", fixed_ips=new_fixed) + assert len(res.fixed_ips or []) == 2 + + def test_remove_port_fixed_ip(self, mock_openstack_connect_network): + mock_conn = mock_openstack_connect_network + + current = Mock() + current.fixed_ips = [ + {"subnet_id": "subnet-1", "ip_address": "10.0.0.10"}, + {"subnet_id": "subnet-2", "ip_address": "10.0.1.10"}, + ] + mock_conn.network.get_port.return_value = current + + updated = Mock() + updated.id = "port-1" + updated.name = "p1" + updated.status = "ACTIVE" + updated.description = None + updated.project_id = None + updated.network_id = "net-1" + updated.admin_state_up = True + updated.is_admin_state_up = True + updated.device_id = None + updated.device_owner = None + updated.mac_address = "fa:16:3e:00:00:06" + updated.fixed_ips = [ + {"subnet_id": "subnet-1", "ip_address": "10.0.0.10"}, + ] + updated.security_group_ids = None + mock_conn.network.update_port.return_value = updated + + tools = self.get_network_tools() + filtered = [ + fi for fi in current.fixed_ips if fi["ip_address"] != "10.0.1.10" + ] + res = tools.update_port("port-1", fixed_ips=filtered) + assert len(res.fixed_ips or []) == 1 + + def test_get_and_update_allowed_address_pairs( + self, + mock_openstack_connect_network, + ): + mock_conn = mock_openstack_connect_network + + port = Mock() + port.allowed_address_pairs = [] + mock_conn.network.get_port.return_value = port + + tools = self.get_network_tools() + lst = tools.get_port_allowed_address_pairs("port-1") + assert lst == [] + + updated = Mock() + updated.id = "port-1" + updated.name = "p1" + updated.status = "ACTIVE" + updated.description = None + updated.project_id = None + updated.network_id = "net-1" + updated.admin_state_up = True + updated.is_admin_state_up = True + updated.device_id = None + updated.device_owner = None + updated.mac_address = "fa:16:3e:00:00:07" + updated.fixed_ips = [] + updated.security_group_ids = None + mock_conn.network.update_port.return_value = updated + + pairs = [] + pairs.append( + {"ip_address": "192.0.2.5", "mac_address": "aa:bb:cc:dd:ee:ff"} + ) + res_add = tools.update_port("port-1", allowed_address_pairs=pairs) + assert isinstance(res_add, Port) + + filtered = [ + p + for p in pairs + if not ( + p["ip_address"] == "192.0.2.5" + and p["mac_address"] == "aa:bb:cc:dd:ee:ff" + ) + ] + res_remove = tools.update_port( + "port-1", allowed_address_pairs=filtered + ) + assert isinstance(res_remove, Port) + + def test_set_port_binding_and_admin_state( + self, + mock_openstack_connect_network, + ): + mock_conn = mock_openstack_connect_network + + updated = Mock() + updated.id = "port-1" + updated.name = "p1" + updated.status = "ACTIVE" + updated.description = None + updated.project_id = None + updated.network_id = "net-1" + updated.is_admin_state_up = False + updated.device_id = None + updated.device_owner = None + updated.mac_address = "fa:16:3e:00:00:08" + updated.fixed_ips = [] + updated.security_group_ids = None + mock_conn.network.update_port.return_value = updated + + tools = self.get_network_tools() + res_bind = tools.set_port_binding( + "port-1", + host_id="host-1", + vnic_type="normal", + profile={"key": "val"}, + ) + assert isinstance(res_bind, Port) + + res_set = tools.update_port("port-1", is_admin_state_up=False) + assert res_set.is_admin_state_up is False + + current = Mock() + current.is_admin_state_up = False + mock_conn.network.get_port.return_value = current + updated.is_admin_state_up = True + res_toggle = tools.update_port( + "port-1", is_admin_state_up=not current.admin_state_up + ) + assert res_toggle.is_admin_state_up is True + + def test_get_subnets_filters_and_has_gateway_true( + self, + mock_openstack_connect_network, + ): + mock_conn = mock_openstack_connect_network + + subnet1 = Mock() + subnet1.id = "subnet-1" + subnet1.name = "s1" + subnet1.status = "ACTIVE" + subnet1.description = None + subnet1.project_id = "proj-1" + subnet1.network_id = "net-1" + subnet1.cidr = "10.0.0.0/24" + subnet1.ip_version = 4 + subnet1.gateway_ip = "10.0.0.1" + subnet1.enable_dhcp = True + subnet1.is_dhcp_enabled = True + subnet1.allocation_pools = [] + subnet1.dns_nameservers = [] + subnet1.host_routes = [] + + subnet2 = Mock() + subnet2.id = "subnet-2" + subnet2.name = "s2" + subnet2.status = "ACTIVE" + subnet2.description = None + subnet2.project_id = "proj-2" + subnet2.network_id = "net-1" + subnet2.cidr = "10.0.1.0/24" + subnet2.ip_version = 4 + subnet2.gateway_ip = None + subnet2.enable_dhcp = False + subnet2.is_dhcp_enabled = False + subnet2.allocation_pools = [] + subnet2.dns_nameservers = [] + subnet2.host_routes = [] + + mock_conn.list_subnets.return_value = [subnet1, subnet2] + + tools = self.get_network_tools() + result = tools.get_subnets( + network_id="net-1", + ip_version=4, + project_id="proj-1", + has_gateway=True, + is_dhcp_enabled=True, + ) + + assert len(result) == 1 + assert result[0] == Subnet( + id="subnet-1", + name="s1", + status="ACTIVE", + description=None, + project_id="proj-1", + network_id="net-1", + cidr="10.0.0.0/24", + ip_version=4, + gateway_ip="10.0.0.1", + is_dhcp_enabled=True, + allocation_pools=[], + dns_nameservers=[], + 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, + }, + ) + + def test_get_subnets_has_gateway_false( + self, + mock_openstack_connect_network, + ): + mock_conn = mock_openstack_connect_network + + subnet1 = Mock() + subnet1.id = "subnet-1" + subnet1.name = "s1" + subnet1.status = "ACTIVE" + subnet1.description = None + subnet1.project_id = None + subnet1.network_id = "net-1" + subnet1.cidr = "10.0.0.0/24" + subnet1.ip_version = 4 + subnet1.gateway_ip = "10.0.0.1" + subnet1.enable_dhcp = True + subnet1.is_dhcp_enabled = True + subnet1.allocation_pools = [] + subnet1.dns_nameservers = [] + subnet1.host_routes = [] + + subnet2 = Mock() + subnet2.id = "subnet-2" + subnet2.name = "s2" + subnet2.status = "ACTIVE" + subnet2.description = None + subnet2.project_id = None + subnet2.network_id = "net-1" + subnet2.cidr = "10.0.1.0/24" + subnet2.ip_version = 4 + subnet2.gateway_ip = None + subnet2.enable_dhcp = False + subnet2.is_dhcp_enabled = False + subnet2.allocation_pools = [] + subnet2.dns_nameservers = [] + subnet2.host_routes = [] + + mock_conn.list_subnets.return_value = [subnet1, subnet2] + + tools = self.get_network_tools() + result = tools.get_subnets( + network_id="net-1", + has_gateway=False, + ) + + assert len(result) == 1 + assert result[0].id == "subnet-2" + + def test_create_subnet_success( + self, + mock_openstack_connect_network, + ): + mock_conn = mock_openstack_connect_network + + subnet = Mock() + subnet.id = "subnet-new" + subnet.name = "s-new" + subnet.status = "ACTIVE" + subnet.description = "desc" + subnet.project_id = "proj-1" + subnet.network_id = "net-1" + subnet.cidr = "10.0.0.0/24" + subnet.ip_version = 4 + subnet.gateway_ip = "10.0.0.1" + subnet.enable_dhcp = True + subnet.is_dhcp_enabled = True + subnet.allocation_pools = [{"start": "10.0.0.10", "end": "10.0.0.20"}] + subnet.dns_nameservers = ["8.8.8.8"] + subnet.host_routes = [] + + mock_conn.network.create_subnet.return_value = subnet + + tools = self.get_network_tools() + result = tools.create_subnet( + network_id="net-1", + cidr="10.0.0.0/24", + name="s-new", + gateway_ip="10.0.0.1", + is_dhcp_enabled=True, + description="desc", + dns_nameservers=["8.8.8.8"], + allocation_pools=[{"start": "10.0.0.10", "end": "10.0.0.20"}], + host_routes=[], + ) + + assert result == Subnet( + id="subnet-new", + name="s-new", + status="ACTIVE", + description="desc", + project_id="proj-1", + network_id="net-1", + cidr="10.0.0.0/24", + ip_version=4, + gateway_ip="10.0.0.1", + is_dhcp_enabled=True, + allocation_pools=[{"start": "10.0.0.10", "end": "10.0.0.20"}], + dns_nameservers=["8.8.8.8"], + host_routes=[], + ) + + mock_conn.network.create_subnet.assert_called_once() + + def test_get_subnet_detail_success( + self, + mock_openstack_connect_network, + ): + mock_conn = mock_openstack_connect_network + + subnet = Mock() + subnet.id = "subnet-1" + subnet.name = "s1" + subnet.status = "ACTIVE" + subnet.description = None + subnet.project_id = "proj-1" + subnet.network_id = "net-1" + subnet.cidr = "10.0.0.0/24" + subnet.ip_version = 4 + subnet.gateway_ip = "10.0.0.1" + subnet.enable_dhcp = True + subnet.is_dhcp_enabled = True + subnet.allocation_pools = [] + subnet.dns_nameservers = [] + subnet.host_routes = [] + + mock_conn.network.get_subnet.return_value = subnet + + tools = self.get_network_tools() + result = tools.get_subnet_detail("subnet-1") + + assert result.id == "subnet-1" + mock_conn.network.get_subnet.assert_called_once_with("subnet-1") + + def test_update_subnet_success( + self, + mock_openstack_connect_network, + ): + mock_conn = mock_openstack_connect_network + + subnet = Mock() + subnet.id = "subnet-1" + subnet.name = "s1-new" + subnet.status = "ACTIVE" + subnet.description = "d-new" + subnet.project_id = "proj-1" + subnet.network_id = "net-1" + subnet.cidr = "10.0.0.0/24" + subnet.ip_version = 4 + subnet.gateway_ip = "10.0.0.254" + subnet.enable_dhcp = False + subnet.is_dhcp_enabled = False + subnet.allocation_pools = [] + subnet.dns_nameservers = [] + subnet.host_routes = [] + + mock_conn.network.update_subnet.return_value = subnet + + tools = self.get_network_tools() + result = tools.update_subnet( + subnet_id="subnet-1", + name="s1-new", + description="d-new", + gateway_ip="10.0.0.254", + is_dhcp_enabled=False, + ) + + assert result.name == "s1-new" + mock_conn.network.update_subnet.assert_called_once_with( + "subnet-1", + name="s1-new", + description="d-new", + gateway_ip="10.0.0.254", + enable_dhcp=False, + ) + + def test_delete_subnet_success( + self, + mock_openstack_connect_network, + ): + mock_conn = mock_openstack_connect_network + + subnet = Mock() + subnet.id = "subnet-1" + mock_conn.network.delete_subnet.return_value = None + + tools = self.get_network_tools() + result = tools.delete_subnet("subnet-1") + + assert result is None + mock_conn.network.delete_subnet.assert_called_once_with( + "subnet-1", + ignore_missing=False, + ) + + def test_set_and_clear_subnet_gateway( + self, + mock_openstack_connect_network, + ): + mock_conn = mock_openstack_connect_network + + updated = Mock() + updated.id = "subnet-1" + updated.name = "s1" + updated.status = "ACTIVE" + updated.description = None + updated.project_id = None + updated.network_id = "net-1" + updated.cidr = "10.0.0.0/24" + updated.ip_version = 4 + updated.gateway_ip = "10.0.0.254" + updated.enable_dhcp = True + updated.is_dhcp_enabled = True + updated.allocation_pools = [] + updated.dns_nameservers = [] + updated.host_routes = [] + + mock_conn.network.update_subnet.return_value = updated + + tools = self.get_network_tools() + res1 = tools.update_subnet("subnet-1", gateway_ip="10.0.0.254") + assert res1.gateway_ip == "10.0.0.254" + + updated.gateway_ip = None + res2 = tools.update_subnet("subnet-1", clear_gateway=True) + assert res2.gateway_ip is None + + def test_set_and_toggle_subnet_dhcp( + self, + mock_openstack_connect_network, + ): + mock_conn = mock_openstack_connect_network + + updated = Mock() + updated.id = "subnet-1" + updated.name = "s1" + updated.status = "ACTIVE" + updated.description = None + updated.project_id = None + updated.network_id = "net-1" + updated.cidr = "10.0.0.0/24" + updated.ip_version = 4 + updated.gateway_ip = "10.0.0.1" + updated.enable_dhcp = True + updated.is_dhcp_enabled = True + updated.allocation_pools = [] + updated.dns_nameservers = [] + updated.host_routes = [] + + mock_conn.network.update_subnet.return_value = updated + + tools = self.get_network_tools() + res1 = tools.update_subnet("subnet-1", is_dhcp_enabled=True) + assert res1.is_dhcp_enabled is True + + updated.enable_dhcp = False + updated.is_dhcp_enabled = False + res2 = tools.update_subnet("subnet-1", is_dhcp_enabled=False) + assert res2.is_dhcp_enabled is False + + def test_get_floating_ips_with_filters_and_unassigned( + self, + mock_openstack_connect_network, + ): + mock_conn = mock_openstack_connect_network + + f1 = Mock() + f1.id = "fip-1" + f1.name = None + f1.status = "DOWN" + f1.description = None + f1.project_id = "proj-1" + f1.floating_ip_address = "203.0.113.10" + f1.floating_network_id = "ext-net" + f1.fixed_ip_address = None + f1.port_id = None + f1.router_id = None + + f2 = Mock() + f2.id = "fip-2" + f2.name = None + f2.status = "ACTIVE" + f2.description = None + f2.project_id = "proj-1" + f2.floating_ip_address = "203.0.113.11" + f2.floating_network_id = "ext-net" + f2.fixed_ip_address = "10.0.0.10" + f2.port_id = "port-1" + f2.router_id = None + + mock_conn.network.ips.return_value = [f1, f2] + + tools = self.get_network_tools() + result = tools.get_floating_ips( + status_filter="ACTIVE", + project_id="proj-1", + floating_network_id="ext-net", + unassigned_only=True, + ) + assert result == [ + FloatingIP( + id="fip-1", + name=None, + status="DOWN", + description=None, + project_id="proj-1", + floating_ip_address="203.0.113.10", + floating_network_id="ext-net", + fixed_ip_address=None, + port_id=None, + router_id=None, + ), + ] + + def test_create_attach_detach_delete_floating_ip( + self, + mock_openstack_connect_network, + ): + mock_conn = mock_openstack_connect_network + + fip = Mock() + fip.id = "fip-1" + fip.name = None + fip.status = "DOWN" + fip.description = "d" + fip.project_id = "proj-1" + fip.floating_ip_address = "203.0.113.10" + fip.floating_network_id = "ext-net" + fip.fixed_ip_address = None + fip.port_id = None + fip.router_id = None + mock_conn.network.create_ip.return_value = fip + + tools = self.get_network_tools() + created = tools.create_floating_ip("ext-net", description="d") + assert created.floating_network_id == "ext-net" + + updated = Mock() + updated.id = "fip-1" + updated.name = None + updated.status = "ACTIVE" + updated.description = "d" + updated.project_id = "proj-1" + updated.floating_ip_address = "203.0.113.10" + updated.floating_network_id = "ext-net" + updated.fixed_ip_address = "10.0.0.10" + updated.port_id = "port-1" + updated.router_id = None + mock_conn.network.update_ip.return_value = updated + + attached = tools.update_floating_ip( + "fip-1", + port_id="port-1", + fixed_ip_address="10.0.0.10", + ) + assert attached.port_id == "port-1" + + updated.port_id = None + detached = tools.update_floating_ip("fip-1", clear_port=True) + assert detached.port_id is None + + mock_conn.network.get_ip.return_value = updated + tools.delete_floating_ip("fip-1") + mock_conn.network.delete_ip.assert_called_once_with( + "fip-1", + ignore_missing=False, + ) + + def test_update_reassign_bulk_and_auto_assign_floating_ip( + self, + mock_openstack_connect_network, + ): + mock_conn = mock_openstack_connect_network - mock_conn.network.delete_network.assert_not_called() + updated = Mock() + updated.id = "fip-1" + updated.name = None + updated.status = "DOWN" + updated.description = "new desc" + updated.project_id = None + updated.floating_ip_address = "203.0.113.10" + updated.floating_network_id = "ext-net" + updated.fixed_ip_address = None + updated.port_id = None + updated.router_id = None + mock_conn.network.update_ip.return_value = updated + + tools = self.get_network_tools() + res_desc = tools.update_floating_ip("fip-1", description="new desc") + assert res_desc.description == "new desc" + + updated.port_id = "port-2" + res_reassign = tools.update_floating_ip("fip-1", port_id="port-2") + assert res_reassign.port_id == "port-2" + + f1 = Mock() + f1.id = "fip-a" + f1.name = None + f1.status = "DOWN" + f1.description = None + f1.project_id = None + f1.floating_ip_address = "203.0.113.20" + f1.floating_network_id = "ext-net" + f1.fixed_ip_address = None + f1.port_id = None + f1.router_id = None + mock_conn.network.create_ip.side_effect = [f1] + bulk = tools.create_floating_ips_bulk("ext-net", 1) + assert len(bulk) == 1 + + exists = Mock() + exists.id = "fip-b" + exists.name = None + exists.status = "DOWN" + exists.description = None + exists.project_id = None + exists.floating_ip_address = "203.0.113.21" + exists.floating_network_id = "ext-net" + exists.fixed_ip_address = None + exists.port_id = None + exists.router_id = None + mock_conn.network.ips.return_value = [exists] + mock_conn.network.update_ip.return_value = exists + auto = tools.assign_first_available_floating_ip("ext-net", "port-9") + assert isinstance(auto, FloatingIP)