Do not inherit from built-in "dict"

This is not recommended because some type methods are implemented not
in Python but in C [1][2] and should not be overridden. Subclassing
the built-in types directly, will yield non-obvious errors that are
hard to debug, and identify at first glance [3].

[1] http://www.kr41.net/2016/03-23-dont_inherit_python_builtin_dict_type.html
[2] https://treyhunner.com/2019/04/why-you-shouldnt-inherit-from-list-and-dict-in-python/
[3] https://medium.com/bynder-tech/using-collections-in-python-36129737b5a1

Closes-Bug: #1849980

Change-Id: I08c712ff1b093370cda2ce66b93e2a0709094fe1
This commit is contained in:
Rodolfo Alonso Hernandez 2019-10-25 17:00:43 +00:00
parent 8375f4cc1c
commit 4b3baeb15a
5 changed files with 264 additions and 166 deletions

View File

@ -15,6 +15,8 @@
import abc
import collections
import copy
import itertools
import os
import re
import shutil
@ -63,12 +65,15 @@ DHCP_RELEASE_TRIES_SLEEP = 0.3
DHCP_OPT_CLIENT_ID_NUM = 61
class DictModel(dict):
class DictModel(collections.abc.MutableMapping):
"""Convert dict into an object that provides attribute access to values."""
__slots__ = ['_dictmodel_internal_storage']
def __init__(self, *args, **kwargs):
"""Convert dict values to DictModel values."""
super(DictModel, self).__init__(*args, **kwargs)
temp_dict = dict(*args)
self._dictmodel_internal_storage = {}
def needs_upgrade(item):
"""Check if `item` is a dict and needs to be changed to DictModel.
@ -82,37 +87,71 @@ class DictModel(dict):
else:
return item
for key, value in self.items():
for key, value in itertools.chain(temp_dict.items(), kwargs.items()):
if isinstance(value, (list, tuple)):
# Keep the same type but convert dicts to DictModels
self[key] = type(value)(
self._dictmodel_internal_storage[key] = type(value)(
(upgrade(item) for item in value)
)
elif needs_upgrade(value):
# Change dict instance values to DictModel instance values
self[key] = DictModel(value)
self._dictmodel_internal_storage[key] = DictModel(value)
else:
self._dictmodel_internal_storage[key] = value
def __getattr__(self, name):
try:
return self[name]
if name == '_dictmodel_internal_storage':
return super(DictModel, self).__getattr__(name)
return self.__getitem__(name)
except KeyError as e:
raise AttributeError(e)
def __setattr__(self, name, value):
self[name] = value
if name == '_dictmodel_internal_storage':
super(DictModel, self).__setattr__(name, value)
else:
self._dictmodel_internal_storage[name] = value
def __delattr__(self, name):
del self[name]
del self._dictmodel_internal_storage[name]
def __str__(self):
pairs = ['%s=%s' % (k, v) for k, v in self.items()]
pairs = ['%s=%s' % (k, v) for k, v in
self._dictmodel_internal_storage.items()]
return ', '.join(sorted(pairs))
def __getitem__(self, name):
return self._dictmodel_internal_storage[name]
def __setitem__(self, name, value):
self._dictmodel_internal_storage[name] = value
def __delitem__(self, name):
del self._dictmodel_internal_storage[name]
def __iter__(self):
return iter(self._dictmodel_internal_storage)
def __len__(self):
return len(self._dictmodel_internal_storage)
def __copy__(self):
return type(self)(self)
def __deepcopy__(self, memo):
cls = self.__class__
result = cls.__new__(cls)
memo[id(self)] = result
result._dictmodel_internal_storage = copy.deepcopy(
self._dictmodel_internal_storage)
return result
class NetModel(DictModel):
def __init__(self, d):
super(NetModel, self).__init__(d)
def __init__(self, *args, **kwargs):
super(NetModel, self).__init__(*args, **kwargs)
self._ns_name = "%s%s" % (NS_PREFIX, self.id)

View File

@ -72,6 +72,7 @@ class NeutronDebugAgent(object):
def _get_network(self, network_id):
network_dict = self.client.show_network(network_id)['network']
network = dhcp.DictModel(network_dict)
# pylint: disable=assigning-non-slot
network.external = network_dict.get('router:external')
obj_subnet = [self._get_subnet(s_id) for s_id in network.subnets]
network.subnets = obj_subnet
@ -166,6 +167,7 @@ class NeutronDebugAgent(object):
for s in network.subnets]}}
port_dict = self.client.create_port(body)['port']
port = dhcp.DictModel(port_dict)
# pylint: disable=assigning-non-slot
port.network = network
for fixed_ip in port.fixed_ips:
fixed_ip.subnet = self._get_subnet(fixed_ip.subnet_id)

View File

@ -107,18 +107,17 @@ class DHCPAgentOVSTestFramework(base.BaseSudoTestCase):
cidr = self._IP_ADDRS[ip_version]['cidr']
if prefix_override is not None:
cidr = '/'.join((cidr.split('/')[0], str(prefix_override)))
sn_dict = dhcp.DictModel({
"id": uuidutils.generate_uuid(),
"network_id": net_id,
"ip_version": ip_version,
"cidr": cidr,
"gateway_ip": (self.
_IP_ADDRS[ip_version]['gateway']),
"enable_dhcp": dhcp_enabled,
"dns_nameservers": [],
"host_routes": [],
"ipv6_ra_mode": None,
"ipv6_address_mode": None})
sn_dict = dhcp.DictModel(
id=uuidutils.generate_uuid(),
network_id=net_id,
ip_version=ip_version,
cidr=cidr,
gateway_ip=self._IP_ADDRS[ip_version]['gateway'],
enable_dhcp=dhcp_enabled,
dns_nameservers=[],
host_routes=[],
ipv6_ra_mode=None,
ipv6_address_mode=None)
if ip_version == lib_const.IP_VERSION_6:
sn_dict['ipv6_address_mode'] = lib_const.DHCPV6_STATEFUL
return sn_dict
@ -127,16 +126,15 @@ class DHCPAgentOVSTestFramework(base.BaseSudoTestCase):
ip_version=lib_const.IP_VERSION_4, ip_address=None):
ip_address = (self._IP_ADDRS[ip_version]['addr']
if not ip_address else ip_address)
port_dict = dhcp.DictModel({
"id": uuidutils.generate_uuid(),
"name": "foo",
"mac_address": mac_address,
"network_id": network_id,
"admin_state_up": True,
"device_id": uuidutils.generate_uuid(),
"device_owner": "foo",
"fixed_ips": [{"subnet_id": subnet_id,
"ip_address": ip_address}], })
port_dict = dhcp.DictModel(id=uuidutils.generate_uuid(),
name="foo",
mac_address=mac_address,
network_id=network_id,
admin_state_up=True,
device_id=uuidutils.generate_uuid(),
device_owner="foo",
fixed_ips=[{"subnet_id": subnet_id,
"ip_address": ip_address}])
return port_dict
def create_network_dict(self, net_id, subnets=None, ports=None,
@ -144,13 +142,12 @@ class DHCPAgentOVSTestFramework(base.BaseSudoTestCase):
subnets = [] if not subnets else subnets
ports = [] if not ports else ports
non_local_subnets = [] if not non_local_subnets else non_local_subnets
net_dict = dhcp.NetModel(d={
"id": net_id,
"subnets": subnets,
"non_local_subnets": non_local_subnets,
"ports": ports,
"admin_state_up": True,
"tenant_id": uuidutils.generate_uuid(), })
net_dict = dhcp.NetModel(id=net_id,
subnets=subnets,
non_local_subnets=non_local_subnets,
ports=ports,
admin_state_up=True,
tenant_id=uuidutils.generate_uuid())
return net_dict
def get_interface_name(self, network, port):

View File

@ -52,9 +52,9 @@ FAKE_TENANT_ID = 'aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa'
FAKE_PRIORITY = 6
fake_subnet1_allocation_pools = dhcp.DictModel(dict(id='', start='172.9.9.2',
end='172.9.9.254'))
fake_subnet1 = dhcp.DictModel(dict(id='bbbbbbbb-bbbb-bbbb-bbbbbbbbbbbb',
fake_subnet1_allocation_pools = dhcp.DictModel(id='', start='172.9.9.2',
end='172.9.9.254')
fake_subnet1 = dhcp.DictModel(id='bbbbbbbb-bbbb-bbbb-bbbbbbbbbbbb',
network_id=FAKE_NETWORK_UUID,
cidr='172.9.9.0/24', enable_dhcp=True, name='',
tenant_id=FAKE_TENANT_ID,
@ -62,30 +62,30 @@ fake_subnet1 = dhcp.DictModel(dict(id='bbbbbbbb-bbbb-bbbb-bbbbbbbbbbbb',
dns_nameservers=[],
ip_version=const.IP_VERSION_4,
ipv6_ra_mode=None, ipv6_address_mode=None,
allocation_pools=fake_subnet1_allocation_pools))
allocation_pools=fake_subnet1_allocation_pools)
fake_subnet2_allocation_pools = dhcp.DictModel(dict(id='', start='172.9.8.2',
end='172.9.8.254'))
fake_subnet2 = dhcp.DictModel(dict(id='dddddddd-dddd-dddd-dddddddddddd',
fake_subnet2_allocation_pools = dhcp.DictModel(id='', start='172.9.8.2',
end='172.9.8.254')
fake_subnet2 = dhcp.DictModel(id='dddddddd-dddd-dddd-dddddddddddd',
network_id=FAKE_NETWORK_UUID,
cidr='172.9.8.0/24', enable_dhcp=False, name='',
tenant_id=FAKE_TENANT_ID, gateway_ip='172.9.8.1',
host_routes=[], dns_nameservers=[],
ip_version=const.IP_VERSION_4,
allocation_pools=fake_subnet2_allocation_pools))
allocation_pools=fake_subnet2_allocation_pools)
fake_subnet3 = dhcp.DictModel(dict(id='bbbbbbbb-1111-2222-bbbbbbbbbbbb',
fake_subnet3 = dhcp.DictModel(id='bbbbbbbb-1111-2222-bbbbbbbbbbbb',
network_id=FAKE_NETWORK_UUID,
cidr='192.168.1.1/24', enable_dhcp=True,
ip_version=const.IP_VERSION_4))
ip_version=const.IP_VERSION_4)
fake_ipv6_subnet = dhcp.DictModel(dict(id='bbbbbbbb-1111-2222-bbbbbbbbbbbb',
fake_ipv6_subnet = dhcp.DictModel(id='bbbbbbbb-1111-2222-bbbbbbbbbbbb',
network_id=FAKE_NETWORK_UUID,
cidr='2001:0db8::0/64', enable_dhcp=True,
tenant_id=FAKE_TENANT_ID,
gateway_ip='2001:0db8::1',
ip_version=const.IP_VERSION_6,
ipv6_ra_mode='slaac', ipv6_address_mode=None))
ipv6_ra_mode='slaac', ipv6_address_mode=None)
fake_meta_subnet = dhcp.DictModel(dict(id='bbbbbbbb-1111-2222-bbbbbbbbbbbb',
network_id=FAKE_NETWORK_UUID,
@ -94,131 +94,124 @@ fake_meta_subnet = dhcp.DictModel(dict(id='bbbbbbbb-1111-2222-bbbbbbbbbbbb',
enable_dhcp=True,
ip_version=const.IP_VERSION_4))
fake_fixed_ip1 = dhcp.DictModel(dict(id='', subnet_id=fake_subnet1.id,
ip_address='172.9.9.9'))
fake_fixed_ip_subnet2 = dhcp.DictModel(dict(id='', subnet_id=fake_subnet2.id,
ip_address='172.9.8.9'))
fake_fixed_ip2 = dhcp.DictModel(dict(id='', subnet_id=fake_subnet1.id,
ip_address='172.9.9.10'))
fake_fixed_ipv6 = dhcp.DictModel(dict(id='', subnet_id=fake_ipv6_subnet.id,
ip_address='2001:db8::a8bb:ccff:fedd:ee99'))
fake_meta_fixed_ip = dhcp.DictModel(dict(id='', subnet=fake_meta_subnet,
ip_address='169.254.169.254'))
fake_allocation_pool_subnet1 = dhcp.DictModel(dict(id='', start='172.9.9.2',
end='172.9.9.254'))
fake_fixed_ip1 = dhcp.DictModel(id='', subnet_id=fake_subnet1.id,
ip_address='172.9.9.9')
fake_fixed_ip_subnet2 = dhcp.DictModel(id='', subnet_id=fake_subnet2.id,
ip_address='172.9.8.9')
fake_fixed_ip2 = dhcp.DictModel(id='', subnet_id=fake_subnet1.id,
ip_address='172.9.9.10')
fake_fixed_ipv6 = dhcp.DictModel(id='', subnet_id=fake_ipv6_subnet.id,
ip_address='2001:db8::a8bb:ccff:fedd:ee99')
fake_meta_fixed_ip = dhcp.DictModel(id='', subnet=fake_meta_subnet,
ip_address='169.254.169.254')
fake_allocation_pool_subnet1 = dhcp.DictModel(id='', start='172.9.9.2',
end='172.9.9.254')
fake_port1 = dhcp.DictModel(dict(id='12345678-1234-aaaa-1234567890ab',
fake_port1 = dhcp.DictModel(id='12345678-1234-aaaa-1234567890ab',
device_id='dhcp-12345678-1234-aaaa-1234567890ab',
device_owner='',
allocation_pools=fake_subnet1_allocation_pools,
mac_address='aa:bb:cc:dd:ee:ff',
network_id=FAKE_NETWORK_UUID,
fixed_ips=[fake_fixed_ip1]))
fixed_ips=[fake_fixed_ip1])
fake_dhcp_port = dhcp.DictModel(dict(id='12345678-1234-aaaa-123456789022',
fake_dhcp_port = dhcp.DictModel(
id='12345678-1234-aaaa-123456789022',
device_id='dhcp-12345678-1234-aaaa-123456789022',
device_owner=const.DEVICE_OWNER_DHCP,
allocation_pools=fake_subnet1_allocation_pools,
mac_address='aa:bb:cc:dd:ee:22',
network_id=FAKE_NETWORK_UUID,
fixed_ips=[fake_fixed_ip2]))
fixed_ips=[fake_fixed_ip2])
fake_port2 = dhcp.DictModel(dict(id='12345678-1234-aaaa-123456789000',
fake_port2 = dhcp.DictModel(id='12345678-1234-aaaa-123456789000',
device_id='dhcp-12345678-1234-aaaa-123456789000',
device_owner='',
mac_address='aa:bb:cc:dd:ee:99',
network_id=FAKE_NETWORK_UUID,
revision_number=77,
fixed_ips=[fake_fixed_ip2]))
fixed_ips=[fake_fixed_ip2])
fake_ipv6_port = dhcp.DictModel(dict(id='12345678-1234-aaaa-123456789000',
fake_ipv6_port = dhcp.DictModel(id='12345678-1234-aaaa-123456789000',
device_owner='',
mac_address='aa:bb:cc:dd:ee:99',
network_id=FAKE_NETWORK_UUID,
fixed_ips=[fake_fixed_ipv6]))
fixed_ips=[fake_fixed_ipv6])
fake_meta_port = dhcp.DictModel(dict(id='12345678-1234-aaaa-1234567890ab',
fake_meta_port = dhcp.DictModel(id='12345678-1234-aaaa-1234567890ab',
mac_address='aa:bb:cc:dd:ee:ff',
network_id=FAKE_NETWORK_UUID,
device_owner=const.DEVICE_OWNER_ROUTER_INTF,
device_id='forzanapoli',
fixed_ips=[fake_meta_fixed_ip]))
fixed_ips=[fake_meta_fixed_ip])
fake_meta_dvr_port = dhcp.DictModel(fake_meta_port.copy())
fake_meta_dvr_port = dhcp.DictModel(fake_meta_port)
fake_meta_dvr_port.device_owner = const.DEVICE_OWNER_DVR_INTERFACE
fake_dist_port = dhcp.DictModel(dict(id='12345678-1234-aaaa-1234567890ab',
fake_dist_port = dhcp.DictModel(id='12345678-1234-aaaa-1234567890ab',
mac_address='aa:bb:cc:dd:ee:ff',
network_id=FAKE_NETWORK_UUID,
device_owner=const.DEVICE_OWNER_DVR_INTERFACE,
device_id='forzanapoli',
fixed_ips=[fake_meta_fixed_ip]))
fixed_ips=[fake_meta_fixed_ip])
fake_network = dhcp.NetModel(dict(id=FAKE_NETWORK_UUID,
fake_network = dhcp.NetModel(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
admin_state_up=True,
subnets=[fake_subnet1, fake_subnet2],
ports=[fake_port1]))
ports=[fake_port1])
fake_network_ipv6 = dhcp.NetModel(dict(id=FAKE_NETWORK_UUID,
fake_network_ipv6 = dhcp.NetModel(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
admin_state_up=True,
subnets=[fake_ipv6_subnet],
ports=[fake_ipv6_port]))
ports=[fake_ipv6_port])
fake_network_ipv6_ipv4 = dhcp.NetModel(dict(id=FAKE_NETWORK_UUID,
fake_network_ipv6_ipv4 = dhcp.NetModel(
id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
admin_state_up=True,
subnets=[fake_ipv6_subnet, fake_subnet1],
ports=[fake_port1]))
ports=[fake_port1])
isolated_network = dhcp.NetModel(
dict(
id=FAKE_NETWORK_UUID,
isolated_network = dhcp.NetModel(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
admin_state_up=True,
subnets=[fake_subnet1],
ports=[fake_port1]))
ports=[fake_port1])
nonisolated_dist_network = dhcp.NetModel(
dict(
id=FAKE_NETWORK_UUID,
nonisolated_dist_network = dhcp.NetModel(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
admin_state_up=True,
subnets=[fake_subnet1],
ports=[fake_port1, fake_port2]))
ports=[fake_port1, fake_port2])
empty_network = dhcp.NetModel(
dict(
id=FAKE_NETWORK_UUID,
empty_network = dhcp.NetModel(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
admin_state_up=True,
subnets=[fake_subnet1],
ports=[]))
ports=[])
fake_meta_network = dhcp.NetModel(
dict(id=FAKE_NETWORK_UUID,
fake_meta_network = dhcp.NetModel(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
admin_state_up=True,
subnets=[fake_meta_subnet],
ports=[fake_meta_port]))
ports=[fake_meta_port])
fake_meta_dvr_network = dhcp.NetModel(fake_meta_network.copy())
fake_meta_dvr_network = dhcp.NetModel(fake_meta_network)
fake_meta_dvr_network.ports = [fake_meta_dvr_port]
fake_dist_network = dhcp.NetModel(
dict(id=FAKE_NETWORK_UUID,
fake_dist_network = dhcp.NetModel(id=FAKE_NETWORK_UUID,
tenant_id=FAKE_TENANT_ID,
admin_state_up=True,
subnets=[fake_meta_subnet],
ports=[fake_meta_port, fake_dist_port]))
ports=[fake_meta_port, fake_dist_port])
fake_down_network = dhcp.NetModel(
dict(id='12345678-dddd-dddd-1234567890ab',
fake_down_network = dhcp.NetModel(id='12345678-dddd-dddd-1234567890ab',
tenant_id=FAKE_TENANT_ID,
admin_state_up=False,
subnets=[],
ports=[]))
ports=[])
class TestDhcpAgent(base.BaseTestCase):
@ -2246,30 +2239,3 @@ class TestDeviceManager(base.BaseTestCase):
self.assertEqual(2, device.route.get_gateway.call_count)
self.assertFalse(device.route.delete_gateway.called)
device.route.add_gateway.assert_has_calls(expected)
class TestDictModel(base.BaseTestCase):
def test_basic_dict(self):
d = dict(a=1, b=2)
m = dhcp.DictModel(d)
self.assertEqual(1, m.a)
self.assertEqual(2, m.b)
def test_dict_has_sub_dict(self):
d = dict(a=dict(b=2))
m = dhcp.DictModel(d)
self.assertEqual(2, m.a.b)
def test_dict_contains_list(self):
d = dict(a=[1, 2])
m = dhcp.DictModel(d)
self.assertEqual([1, 2], m.a)
def test_dict_contains_list_of_dicts(self):
d = dict(a=[dict(b=2), dict(c=3)])
m = dhcp.DictModel(d)
self.assertEqual(2, m.a[0].b)
self.assertEqual(3, m.a[1].c)

View File

@ -13,6 +13,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import copy
import os
import mock
@ -24,6 +25,7 @@ from neutron_lib import fixture as lib_fixtures
from oslo_config import cfg
import oslo_messaging
from oslo_utils import fileutils
from oslo_utils import uuidutils
import testtools
from neutron.agent.linux import dhcp
@ -3129,6 +3131,34 @@ class TestDeviceManager(TestConfBase):
class TestDictModel(base.BaseTestCase):
def setUp(self):
super(TestDictModel, self).setUp()
self._a = uuidutils.generate_uuid()
self._b = uuidutils.generate_uuid()
self.dm = dhcp.DictModel(a=self._a, b=self._b)
def test_basic_dict(self):
d = dict(a=1, b=2)
m = dhcp.DictModel(d)
self.assertEqual(1, m.a)
self.assertEqual(2, m.b)
def test_dict_has_sub_dict(self):
d = dict(a=dict(b=2))
m = dhcp.DictModel(d)
self.assertEqual(2, m.a.b)
def test_dict_contains_list(self):
d = dict(a=[1, 2])
m = dhcp.DictModel(d)
self.assertEqual([1, 2], m.a)
def test_dict_contains_list_of_dicts(self):
d = dict(a=[dict(b=2), dict(c=3)])
m = dhcp.DictModel(d)
self.assertEqual(2, m.a[0].b)
self.assertEqual(3, m.a[1].c)
def test_string_representation_port(self):
port = dhcp.DictModel({'id': 'id', 'network_id': 'net_id'})
self.assertEqual('id=id, network_id=net_id', str(port))
@ -3136,3 +3166,67 @@ class TestDictModel(base.BaseTestCase):
def test_string_representation_network(self):
net = dhcp.DictModel({'id': 'id', 'name': 'myname'})
self.assertEqual('id=id, name=myname', str(net))
def test__init_parameters(self):
self.assertEqual(self._a, self.dm.a)
self.assertEqual(self._b, self.dm.b)
def test__init_dictmodel(self):
dm2 = dhcp.DictModel(self.dm)
self.assertEqual(self._a, dm2.a)
self.assertEqual(self._b, dm2.b)
dm2.a = 'new_value'
self.assertEqual('new_value', dm2.a)
self.assertEqual(self._a, self.dm.a)
def test__getattr(self):
self.assertEqual({'a': self._a, 'b': self._b},
self.dm._dictmodel_internal_storage)
try:
self.dm.z
except AttributeError:
pass
except Exception:
self.fail('Getting a non existing attribute from a DictModel '
'object should raise AttributeError')
def test__setattr(self):
self.dm.c = 'c_value'
self.assertEqual('c_value', self.dm.c)
def test__delattr(self):
del self.dm.a
self.assertIsNone(self.dm.get('a'))
def test__str(self):
reference = 'a=%s, b=%s' % (self._a, self._b)
self.assertEqual(reference, str(self.dm))
def test__getitem(self):
self.assertEqual(self._a, self.dm['a'])
self.assertEqual(self._b, self.dm['b'])
def test__setitem(self):
self.dm['a'] = 'a_new_value'
self.assertEqual('a_new_value', self.dm.a)
self.assertEqual('a_new_value', self.dm['a'])
self.assertEqual(self._b, self.dm.b)
def test__iter(self):
list_keys = sorted(list(self.dm))
self.assertEqual(['a', 'b'], list_keys)
def test__len(self):
self.assertEqual(2, len(self.dm))
def test__copy_and_deepcopy(self):
for method in (copy.copy, copy.deepcopy):
self.dm._tuple = (10, 11)
self.dm._list = [20, 21]
dm2 = method(self.dm)
dm2._tuple = (30, 31)
dm2._list[0] = 200
self.assertEqual((10, 11), self.dm._tuple)
self.assertEqual([20, 21], self.dm._list)
self.assertEqual((30, 31), dm2._tuple)
self.assertEqual([200, 21], dm2._list)