openstack/python-manilaclient
Code Issues Proposed changes

Fix neutron validation in share network creation

Browse Source 
The 'share network create' command previously
attempted to fetch the subnet directly, which
could lead to errors when the neuron network data
was invalid or missing.

This change updates the logic to properly handle
invalid neutron network or subnet IDs during share
network creation.

Test Plan:
- Added unit tests to verify valid and invalid
  neutron info.

Closes-Bug: #2051394
Change-Id: Ib233f02ad94326d5b8ffadf962fb911d417b024a
Signed-off-by: denver-baraka <denverbaraka@gmail.com>
Assisted-By: Copilot
This commit is contained in:
denver-baraka
2025-10-18 17:51:56 +03:00
parent 7ed6482c6d
commit 53ca6e9f87
5 changed files with 259 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
manilaclient/osc/v2/share_network_subnets.py
View File

@@ -110,13 +110,37 @@ class CreateShareNetworkSubnet(command.ShowOne):
"Property can be specified only with manila API "
"version >= 2.78.")
if xor(bool(parsed_args.neutron_net_id),
bool(parsed_args.neutron_subnet_id)):
neutron_client = getattr(self.app.client_manager, 'network', None)
neutron_net_id = parsed_args.neutron_net_id
neutron_subnet_id = parsed_args.neutron_subnet_id
if xor(bool(neutron_net_id),
bool(neutron_subnet_id)):
raise exceptions.CommandError(
"Both neutron_net_id and neutron_subnet_id should be "
"specified. Alternatively, neither of them should be "
"specified.")
if neutron_client and neutron_net_id:
try:
neutron_net_id = neutron_client.find_network(
neutron_net_id,
ignore_missing=False).id
except Exception:
raise exceptions.CommandError(
f"Neutron network '{neutron_net_id}'"
f" not found.")
if neutron_client and neutron_subnet_id:
try:
neutron_subnet_id = neutron_client.find_subnet(
neutron_subnet_id,
ignore_missing=False).id
except Exception:
raise exceptions.CommandError(
f"Neutron subnet '{neutron_subnet_id}'"
f" not found.")
share_network_id = oscutils.find_resource(
share_client.share_networks,
parsed_args.share_network).id
@@ -129,8 +153,8 @@ class CreateShareNetworkSubnet(command.ShowOne):
subnet_create_check = (
share_client.share_networks.share_network_subnet_create_check(
neutron_net_id=parsed_args.neutron_net_id,
neutron_subnet_id=parsed_args.neutron_subnet_id,
neutron_net_id=neutron_net_id,
neutron_subnet_id=neutron_subnet_id,
availability_zone=parsed_args.availability_zone,
reset_operation=parsed_args.restart_check,
share_network_id=share_network_id)
@@ -147,8 +171,8 @@ class CreateShareNetworkSubnet(command.ShowOne):
subnet_data[k] = dict_values
else:
share_network_subnet = share_client.share_network_subnets.create(
neutron_net_id=parsed_args.neutron_net_id,
neutron_subnet_id=parsed_args.neutron_subnet_id,
neutron_net_id=neutron_net_id,
neutron_subnet_id=neutron_subnet_id,
availability_zone=parsed_args.availability_zone,
share_network_id=share_network_id,
metadata=parsed_args.property

26
manilaclient/osc/v2/share_networks.py
View File

@@ -300,6 +300,28 @@ class CreateShareNetwork(command.ShowOne):
def take_action(self, parsed_args):
share_client = self.app.client_manager.share
neutron_client = getattr(self.app.client_manager, 'network', None)
neutron_net_id = parsed_args.neutron_net_id
neutron_subnet_id = parsed_args.neutron_subnet_id
if neutron_client and neutron_net_id:
try:
neutron_net_id = neutron_client.find_network(
neutron_net_id,
ignore_missing=False).id
except Exception:
raise exceptions.CommandError(
f"Neutron network '{parsed_args.neutron_net_id}'"
f" not found.")
if neutron_client and neutron_subnet_id:
try:
neutron_subnet_id = neutron_client.find_subnet(
neutron_subnet_id,
ignore_missing=False).id
except Exception:
raise exceptions.CommandError(
f"Neutron subnet '{parsed_args.neutron_subnet_id}'"
f" not found.")
availability_zone = None
if (parsed_args.availability_zone and
@@ -313,8 +335,8 @@ class CreateShareNetwork(command.ShowOne):
kwargs = {
"name": parsed_args.name,
"description": parsed_args.description,
"neutron_net_id": parsed_args.neutron_net_id,
"neutron_subnet_id": parsed_args.neutron_subnet_id,
"neutron_net_id": neutron_net_id,
"neutron_subnet_id": neutron_subnet_id,
}
if availability_zone:
kwargs['availability_zone'] = availability_zone

103
manilaclient/tests/unit/osc/v2/test_share_network_subnets.py
View File

@@ -93,6 +93,109 @@ class TestShareNetworkSubnetCreate(TestShareNetworkSubnet):
self.assertCountEqual(self.columns, columns)
self.assertCountEqual(self.data, data)
def test_share_network_subnet_create_valid_neutron_info(self):
fake_neutron_net_id = str(uuid.uuid4())
fake_neutron_subnet_id = str(uuid.uuid4())
neutron_client = mock.Mock()
self.app.client_manager.network = neutron_client
neutron_client.find_network.return_value = mock.Mock(
id=fake_neutron_net_id)
neutron_client.find_subnet.return_value = mock.Mock(
id=fake_neutron_subnet_id)
arglist = [
self.share_network.id,
'--neutron-net-id', fake_neutron_net_id,
'--neutron-subnet-id', fake_neutron_subnet_id,
]
verifylist = [
('share_network', self.share_network.id),
('neutron_net_id', fake_neutron_net_id),
('neutron_subnet_id', fake_neutron_subnet_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
neutron_client.find_network.assert_called_once_with(
fake_neutron_net_id,
ignore_missing=False
)
neutron_client.find_subnet.assert_called_once_with(
fake_neutron_subnet_id,
ignore_missing=False
)
self.share_subnets_mock.create.assert_called_once_with(
share_network_id=self.share_network.id,
neutron_net_id=fake_neutron_net_id,
neutron_subnet_id=fake_neutron_subnet_id,
availability_zone=None,
metadata={},
)
self.assertCountEqual(self.columns, columns)
self.assertCountEqual(self.data, data)
def test_share_network_subnet_create_invalid_neutron_network(self):
fake_neutron_net_id = str(uuid.uuid4())
fake_neutron_net_id = str(uuid.uuid4())
neutron_client = mock.Mock()
self.app.client_manager.network = neutron_client
neutron_client.find_network.side_effect = Exception(
"Network not found.")
arglist = [
self.share_network.id,
'--neutron-net-id', fake_neutron_net_id,
'--neutron-subnet-id', fake_neutron_net_id,
]
verifylist = [
('share_network', self.share_network.id),
('neutron_net_id', fake_neutron_net_id),
('neutron_subnet_id', fake_neutron_net_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.assertRaises(exceptions.CommandError,
self.cmd.take_action,
parsed_args)
neutron_client.find_network.assert_called_once_with(
fake_neutron_net_id,
ignore_missing=False
)
neutron_client.find_subnet.assert_not_called()
self.share_subnets_mock.create.assert_not_called()
def test_share_network_subnet_create_invalid_neutron_subnet(self):
fake_neutron_net_id = str(uuid.uuid4())
fake_neutron_subnet_id = str(uuid.uuid4())
neutron_client = mock.Mock()
self.app.client_manager.network = neutron_client
neutron_client.find_subnet.side_effect = Exception(
"Subnet not found.")
arglist = [
self.share_network.id,
'--neutron-net-id', fake_neutron_net_id,
'--neutron-subnet-id', fake_neutron_subnet_id,
]
verifylist = [
('share_network', self.share_network.id),
('neutron_net_id', fake_neutron_net_id),
('neutron_subnet_id', fake_neutron_subnet_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.assertRaises(exceptions.CommandError,
self.cmd.take_action,
parsed_args)
neutron_client.find_network.assert_called_once_with(
fake_neutron_net_id,
ignore_missing=False
)
neutron_client.find_subnet.assert_called_once_with(
fake_neutron_subnet_id,
ignore_missing=False
)
self.share_subnets_mock.create.assert_not_called()
def test_share_network_subnet_create_arg_group_exception(self):
fake_neutron_net_id = str(uuid.uuid4())

90
manilaclient/tests/unit/osc/v2/test_share_networks.py
View File

@@ -122,6 +122,96 @@ class TestShareNetworkCreate(TestShareNetwork):
self.assertCountEqual(self.columns, columns)
self.assertCountEqual(self.data, data)
def test_share_network_create_with_valid_neutron_info(self):
fake_neutron_net_id = str(uuid.uuid4())
fake_neutron_subnet_id = str(uuid.uuid4())
neutron_client = mock.Mock()
self.app.client_manager.network = neutron_client
neutron_client.find_network.return_value = mock.Mock(
id=fake_neutron_net_id)
neutron_client.find_subnet.return_value = mock.Mock(
id=fake_neutron_subnet_id)
arglist = [
'--neutron-net-id', fake_neutron_net_id,
'--neutron-subnet-id', fake_neutron_subnet_id,
]
verifylist = [
('neutron_net_id', fake_neutron_net_id),
('neutron_subnet_id', fake_neutron_subnet_id),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
neutron_client.find_network.assert_called_once_with(
fake_neutron_net_id, ignore_missing=False)
neutron_client.find_subnet.assert_called_once_with(
fake_neutron_subnet_id, ignore_missing=False)
self.share_networks_mock.create.assert_called_once_with(
name=None,
description=None,
neutron_net_id=fake_neutron_net_id,
neutron_subnet_id=fake_neutron_subnet_id
)
self.assertCountEqual(self.columns, columns)
self.assertCountEqual(self.data, data)
def test_share_network_create_with_invalid_neutron_network(self):
fake_neutron_net_id = str(uuid.uuid4())
neutron_client = mock.Mock()
self.app.client_manager.network = neutron_client
neutron_client.find_network.side_effect = Exception(
"Network not found")
arglist = [
'--neutron-net-id', fake_neutron_net_id
]
verifylist = [
('neutron_net_id', fake_neutron_net_id)
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.assertRaises(
exceptions.CommandError,
self.cmd.take_action,
parsed_args
)
neutron_client.find_network.assert_called_once_with(
fake_neutron_net_id, ignore_missing=False)
self.share_networks_mock.create.assert_not_called()
def test_share_network_create_with_invalid_neutron_subnet(self):
fake_neutron_subnet_id = str(uuid.uuid4())
neutron_client = mock.Mock()
self.app.client_manager.network = neutron_client
neutron_client.find_subnet.side_effect = Exception(
"Subnet not found")
arglist = [
'--neutron-subnet-id', fake_neutron_subnet_id
]
verifylist = [
('neutron_subnet_id', fake_neutron_subnet_id)
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.assertRaises(
exceptions.CommandError,
self.cmd.take_action,
parsed_args
)
neutron_client.find_subnet.assert_called_once_with(
fake_neutron_subnet_id, ignore_missing=False)
self.share_networks_mock.create.assert_not_called()
@ddt.ddt
class TestShareNetworkDelete(TestShareNetwork):

12
releasenotes/notes/bug-2051394-fix-validate-neutron-info-ebe679675111eec9.yaml Normal file
View File

@@ -0,0 +1,12 @@
fixes:
- |
The "openstack share network create" and
"openstack share network subnet create" commands
now validate Neutron network information provided
before sending the request.
Previously, validation only occured in the Manila service
which could result in client-side errors when invalid network
or subnet IDs were supplied. The client now performs
proper validation and handles missing or invalid
network data gracefully.