Skip to content

Commit 518ef4b

Browse files
committed
fix(network): refactor code
- remove custom raise - remove test case relation removing raise
1 parent 5ff8e7a commit 518ef4b

File tree

2 files changed

+22
-175
lines changed

2 files changed

+22
-175
lines changed

src/openstack_mcp_server/tools/network_tools.py

Lines changed: 20 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -157,9 +157,6 @@ def get_network_detail(self, network_id: str) -> Network:
157157
conn = get_openstack_conn()
158158

159159
network = conn.network.get_network(network_id)
160-
if not network:
161-
raise Exception(f"Network with ID {network_id} not found")
162-
163160
return self._convert_to_network_model(network)
164161

165162
def update_network(
@@ -201,10 +198,9 @@ def update_network(
201198
update_args["shared"] = is_shared
202199

203200
if not update_args:
204-
raise Exception("No update parameters provided")
205-
201+
current = conn.network.get_network(network_id)
202+
return self._convert_to_network_model(current)
206203
network = conn.network.update_network(network_id, **update_args)
207-
208204
return self._convert_to_network_model(network)
209205

210206
def delete_network(self, network_id: str) -> None:
@@ -218,11 +214,6 @@ def delete_network(self, network_id: str) -> None:
218214
:raises Exception: If the network is not found
219215
"""
220216
conn = get_openstack_conn()
221-
222-
network = conn.network.get_network(network_id)
223-
if not network:
224-
raise Exception(f"Network with ID {network_id} not found")
225-
226217
conn.network.delete_network(network_id, ignore_missing=False)
227218

228219
return None
@@ -289,14 +280,9 @@ def get_subnets(
289280
filters["enable_dhcp"] = is_dhcp_enabled
290281
subnets = conn.list_subnets(filters=filters)
291282
if has_gateway is not None:
292-
if has_gateway:
293-
subnets = [
294-
s for s in subnets if getattr(s, "gateway_ip", None)
295-
]
296-
else:
297-
subnets = [
298-
s for s in subnets if not getattr(s, "gateway_ip", None)
299-
]
283+
subnets = [
284+
s for s in subnets if (s.gateway_ip is not None) == has_gateway
285+
]
300286
return [self._convert_to_subnet_model(subnet) for subnet in subnets]
301287

302288
def create_subnet(
@@ -372,8 +358,6 @@ def get_subnet_detail(self, subnet_id: str) -> Subnet:
372358
"""
373359
conn = get_openstack_conn()
374360
subnet = conn.network.get_subnet(subnet_id)
375-
if not subnet:
376-
raise Exception(f"Subnet with ID {subnet_id} not found")
377361
return self._convert_to_subnet_model(subnet)
378362

379363
def update_subnet(
@@ -427,7 +411,8 @@ def update_subnet(
427411
if host_routes is not None:
428412
update_args["host_routes"] = host_routes
429413
if not update_args:
430-
raise Exception("No update parameters provided")
414+
current = conn.network.get_subnet(subnet_id)
415+
return self._convert_to_subnet_model(current)
431416
subnet = conn.network.update_subnet(subnet_id, **update_args)
432417
return self._convert_to_subnet_model(subnet)
433418

@@ -442,9 +427,6 @@ def delete_subnet(self, subnet_id: str) -> None:
442427
:raises Exception: If the subnet is not found
443428
"""
444429
conn = get_openstack_conn()
445-
subnet = conn.network.get_subnet(subnet_id)
446-
if not subnet:
447-
raise Exception(f"Subnet with ID {subnet_id} not found")
448430
conn.network.delete_subnet(subnet_id, ignore_missing=False)
449431
return None
450432

@@ -503,11 +485,9 @@ def toggle_subnet_dhcp(self, subnet_id: str) -> Subnet:
503485
"""
504486
conn = get_openstack_conn()
505487
current = conn.network.get_subnet(subnet_id)
506-
if not current:
507-
raise Exception(f"Subnet with ID {subnet_id} not found")
508488
subnet = conn.network.update_subnet(
509489
subnet_id,
510-
enable_dhcp=not bool(current.enable_dhcp),
490+
enable_dhcp=not current.enable_dhcp,
511491
)
512492
return self._convert_to_subnet_model(subnet)
513493

@@ -586,8 +566,6 @@ def add_port_fixed_ip(
586566
"""
587567
conn = get_openstack_conn()
588568
port = conn.network.get_port(port_id)
589-
if not port:
590-
raise Exception(f"Port with ID {port_id} not found")
591569
fixed_ips = list(port.fixed_ips or [])
592570
entry: dict = {}
593571
if subnet_id is not None:
@@ -619,8 +597,6 @@ def remove_port_fixed_ip(
619597
"""
620598
conn = get_openstack_conn()
621599
port = conn.network.get_port(port_id)
622-
if not port:
623-
raise Exception(f"Port with ID {port_id} not found")
624600
current = list(port.fixed_ips or [])
625601
if not current:
626602
return self._convert_to_port_model(port)
@@ -648,9 +624,7 @@ def get_port_allowed_address_pairs(self, port_id: str) -> list[dict]:
648624
"""
649625
conn = get_openstack_conn()
650626
port = conn.network.get_port(port_id)
651-
if not port:
652-
raise Exception(f"Port with ID {port_id} not found")
653-
return list(getattr(port, "allowed_address_pairs", []) or [])
627+
return list(port.allowed_address_pairs or [])
654628

655629
def add_port_allowed_address_pair(
656630
self,
@@ -673,13 +647,13 @@ def add_port_allowed_address_pair(
673647
"""
674648
conn = get_openstack_conn()
675649
port = conn.network.get_port(port_id)
676-
if not port:
677-
raise Exception(f"Port with ID {port_id} not found")
678-
pairs = list(getattr(port, "allowed_address_pairs", []) or [])
650+
651+
pairs = list(port.allowed_address_pairs or [])
679652
entry = {"ip_address": ip_address}
680653
if mac_address is not None:
681654
entry["mac_address"] = mac_address
682655
pairs.append(entry)
656+
683657
updated = conn.network.update_port(
684658
port_id,
685659
allowed_address_pairs=pairs,
@@ -707,9 +681,7 @@ def remove_port_allowed_address_pair(
707681
"""
708682
conn = get_openstack_conn()
709683
port = conn.network.get_port(port_id)
710-
if not port:
711-
raise Exception(f"Port with ID {port_id} not found")
712-
pairs = list(getattr(port, "allowed_address_pairs", []) or [])
684+
pairs = list(port.allowed_address_pairs or [])
713685

714686
def keep(p: dict) -> bool:
715687
if mac_address is None:
@@ -757,7 +729,8 @@ def set_port_binding(
757729
if profile is not None:
758730
update_args["binding_profile"] = profile
759731
if not update_args:
760-
raise Exception("No update parameters provided")
732+
current = conn.network.get_port(port_id)
733+
return self._convert_to_port_model(current)
761734
updated = conn.network.update_port(port_id, **update_args)
762735
return self._convert_to_port_model(updated)
763736

@@ -795,11 +768,9 @@ def toggle_port_admin_state(self, port_id: str) -> Port:
795768
"""
796769
conn = get_openstack_conn()
797770
current = conn.network.get_port(port_id)
798-
if not current:
799-
raise Exception(f"Port with ID {port_id} not found")
800771
updated = conn.network.update_port(
801772
port_id,
802-
admin_state_up=not bool(current.admin_state_up),
773+
admin_state_up=not current.admin_state_up,
803774
)
804775
return self._convert_to_port_model(updated)
805776

@@ -863,8 +834,6 @@ def get_port_detail(self, port_id: str) -> Port:
863834
"""
864835
conn = get_openstack_conn()
865836
port = conn.network.get_port(port_id)
866-
if not port:
867-
raise Exception(f"Port with ID {port_id} not found")
868837
return self._convert_to_port_model(port)
869838

870839
def update_port(
@@ -908,7 +877,8 @@ def update_port(
908877
if security_group_ids is not None:
909878
update_args["security_groups"] = security_group_ids
910879
if not update_args:
911-
raise Exception("No update parameters provided")
880+
current = conn.network.get_port(port_id)
881+
return self._convert_to_port_model(current)
912882
port = conn.network.update_port(port_id, **update_args)
913883
return self._convert_to_port_model(port)
914884

@@ -923,9 +893,6 @@ def delete_port(self, port_id: str) -> None:
923893
:raises Exception: If the port is not found
924894
"""
925895
conn = get_openstack_conn()
926-
port = conn.network.get_port(port_id)
927-
if not port:
928-
raise Exception(f"Port with ID {port_id} not found")
929896
conn.network.delete_port(port_id, ignore_missing=False)
930897
return None
931898

@@ -991,7 +958,7 @@ def get_floating_ips(
991958
filters["floating_network_id"] = floating_network_id
992959
ips = list(conn.network.ips(**filters))
993960
if unassigned_only:
994-
ips = [i for i in ips if not getattr(i, "port_id", None)]
961+
ips = [i for i in ips if not i.port_id]
995962
return [self._convert_to_floating_ip_model(ip) for ip in ips]
996963

997964
def create_floating_ip(
@@ -1104,9 +1071,6 @@ def delete_floating_ip(self, floating_ip_id: str) -> None:
11041071
:raises Exception: If the floating IP is not found
11051072
"""
11061073
conn = get_openstack_conn()
1107-
ip = conn.network.get_ip(floating_ip_id)
1108-
if not ip:
1109-
raise Exception(f"Floating IP with ID {floating_ip_id} not found")
11101074
conn.network.delete_ip(floating_ip_id, ignore_missing=False)
11111075
return None
11121076

@@ -1199,7 +1163,7 @@ def assign_first_available_floating_ip(
11991163
conn.network.ips(floating_network_id=floating_network_id),
12001164
)
12011165
available = next(
1202-
(i for i in existing if not getattr(i, "port_id", None)),
1166+
(i for i in existing if not i.port_id),
12031167
None,
12041168
)
12051169
if available is None:

0 commit comments

Comments
 (0)