Ensure string type in provnet-network-type ext_id

When multiple segments exist, or when deleting all segments, the
provider:network_type can be None which is not a valid value
for an ovsdb external_id column, which only stores strings.

Closes-Bug: #2098815
Change-Id: I7312570520bfc0e3d5e8e68cf065577cd3dc8786
This commit is contained in:
Terry Wilson
2025-02-18 23:30:40 +00:00
parent 86f94de99a
commit c9b6679b4f
3 changed files with 38 additions and 14 deletions

View File

@@ -17,6 +17,7 @@
# when needed.
"""Utilities and helper functions."""
from collections import abc
import datetime
import functools
import hashlib
@@ -1104,3 +1105,11 @@ def ts_to_datetime(timestamp):
def datetime_to_ts(_datetime):
"""Converts datetime to timestamp in seconds"""
return int(datetime.datetime.timestamp(_datetime))
def stringmap(data: abc.Mapping[str, typing.Any],
default: str = '') -> dict[str, str]:
result = {}
for key, value in data.items():
result[key] = default if value is None else str(value)
return result

View File

@@ -2091,21 +2091,24 @@ class OVNClient:
return (ovn_const.ETHTYPE_8021ad if network.get(qinq_apidef.QINQ_FIELD)
else ovn_const.ETHTYPE_8021q)
def _gen_network_parameters(self, network):
params = {'external_ids': {
def _gen_network_parameters(self,
network: dict) -> dict[str, dict[str, str]]:
ext_ids = {
ovn_const.OVN_NETWORK_NAME_EXT_ID_KEY: network['name'],
ovn_const.OVN_NETWORK_MTU_EXT_ID_KEY: str(network['mtu']),
ovn_const.OVN_REV_NUM_EXT_ID_KEY: str(
utils.get_revision_number(network, ovn_const.TYPE_NETWORKS)),
ovn_const.OVN_NETWORK_MTU_EXT_ID_KEY: network['mtu'],
ovn_const.OVN_REV_NUM_EXT_ID_KEY:
utils.get_revision_number(network, ovn_const.TYPE_NETWORKS),
ovn_const.OVN_AZ_HINTS_EXT_ID_KEY:
','.join(common_utils.get_az_hints(network)),
# NOTE(ralonsoh): it is not considered the case of multiple
# segments.
# NOTE(twilson): in the case of multiple segments, or when all
# segments are removed, NETWORK_TYPE=None, which is invalid ovsdb
ovn_const.OVN_NETTYPE_EXT_ID_KEY: network.get(pnet.NETWORK_TYPE),
}}
}
# Enable IGMP snooping if igmp_snooping_enable is enabled in Neutron
params['other_config'] = {
other_config = {
ovn_const.MCAST_SNOOP:
ovs_conf.get_igmp_snooping_enabled(),
ovn_const.MCAST_FLOOD_UNREGISTERED:
@@ -2113,15 +2116,15 @@ class OVNClient:
ovn_const.VLAN_PASSTHRU: str(
self._get_vlan_passthru(network)).lower()}
if utils.is_provider_network(network):
params['other_config'][ovn_const.LS_OPTIONS_FDB_AGE_THRESHOLD] = (
other_config[ovn_const.LS_OPTIONS_FDB_AGE_THRESHOLD] = (
ovn_conf.get_fdb_age_threshold())
if utils.is_external_network(network):
params['other_config'][
ovn_const.LS_OPTIONS_BROADCAST_ARPS_ROUTERS] = (
'true'
if ovn_conf.is_broadcast_arps_to_all_routers_enabled() else
'false')
return params
other_config[ovn_const.LS_OPTIONS_BROADCAST_ARPS_ROUTERS] = (
'true'
if ovn_conf.is_broadcast_arps_to_all_routers_enabled() else
'false')
return {'external_ids': common_utils.stringmap(ext_ids),
'other_config': common_utils.stringmap(other_config)}
def create_network(self, context, network):
# Create a logical switch with a name equal to the Neutron network

View File

@@ -665,3 +665,15 @@ class ParsePermittedEthertypesTestCase(base.BaseTestCase):
'0x123R'),
]
mock_log.warning.assert_has_calls(calls)
class StringMapTestCase(base.BaseTestCase):
data = {'a': 1, 'b': None, 'c': 'hi'}
def test_stringmap(self):
expected = {'a': '1', 'b': '', 'c': 'hi'}
self.assertEqual(expected, utils.stringmap(self.data))
def test_stringmap_custom_default(self):
self.assertEqual('None', utils.stringmap(self.data, 'None')['b'])