/
test_vnc_plugin_db.py
115 lines (93 loc) · 4.31 KB
/
test_vnc_plugin_db.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
import unittest
import uuid
from flexmock import flexmock
from vnc_openstack import neutron_plugin_db as db
class MockDbInterface(db.DBInterface):
def __init__(self):
pass
class TestDbInterface(unittest.TestCase):
_tenant_ids = ['tenant_id_1',
'tenant_id_2']
def _list_resource(self, resource, ret_count=0):
def _list_others(parent_id, count):
self.assertEqual(count, True)
self.assertTrue(parent_id in self._tenant_ids)
self.assertTrue(resource in ['virtual_networks',
'virtual_machine_interfaces',
'logical_routers',
'network_policys',
'network_ipams',
'route_tables'])
r = resource.replace("_", "-")
return {r: {'count': ret_count}}
def _list_fip(back_ref_id, count):
self.assertEqual(count, True)
self.assertTrue(back_ref_id in self._tenant_ids)
r = resource.replace("_", "-")
return {r: {'count': ret_count}}
if resource == "floating_ips":
return _list_fip
return _list_others
def _test_for(self, resource):
dbi = MockDbInterface()
kwargs={"operational": True,
resource + "_list": self._list_resource(resource, 1),
}
dbi._vnc_lib = flexmock(**kwargs)
ret = dbi._resource_count_optimized(resource,
filters={'tenant_id': self._tenant_ids[0]})
self.assertEqual(ret, 1)
ret = dbi._resource_count_optimized(resource,
filters={'tenant_id': self._tenant_ids})
self.assertEqual(ret, 2)
def test_resource_count_optimized(self):
dbi = MockDbInterface()
ret = dbi._resource_count_optimized('virtual-networks',
filters={'f': 'some-filter'})
self.assertEqual(ret, None)
ret = dbi._resource_count_optimized('virtual-networks',
filters={'tenant_id': 'some-id',
'f': 'some_filter'})
self.assertEqual(ret, None)
self._test_for("virtual_networks")
self._test_for("virtual_machine_interfaces")
self._test_for("floating_ips")
self._test_for("logical_routers")
self._test_for("network_policys")
self._test_for("network_ipams")
self._test_for("route_tables")
def test_floating_show_router_id(self):
dbi = MockDbInterface()
vmi_obj = None
def fake_virtual_machine_read(id, fq_name=None, fields=None,
parent_id=None):
if id == 'fip_port_uuid1':
net_uuid = 'match_vn_uuid'
elif id == 'fip_port_uuid2':
net_uuid = 'miss_vn_uuid'
elif id == 'router_port_uuid':
net_uuid = 'match_vn_uuid'
return flexmock(uuid=id,
get_virtual_network_refs=\
lambda: [{'uuid': net_uuid}])
dbi._vnc_lib = flexmock(
fq_name_to_id=lambda res, name: 'fip_pool_uuid',
virtual_machine_interface_read=fake_virtual_machine_read,
logical_routers_list=lambda parent_id, detail: [
flexmock(uuid='router_uuid',
get_virtual_machine_interface_refs=\
lambda: [{'uuid': 'router_port_uuid'}])])
fip_obj = flexmock(
uuid = 'fip_uuid',
get_fq_name=lambda: ['domain', 'project', 'fip'],
get_project_refs=lambda: [{'uuid': str(uuid.uuid4())}],
get_floating_ip_address=lambda: 'fip_ip',
get_floating_ip_fixed_ip_address= lambda: 'fip_port_ip')
fip_obj.get_virtual_machine_interface_refs = \
lambda: [{'uuid': 'fip_port_uuid1'}]
fip_neutron = dbi._floatingip_vnc_to_neutron(fip_obj)
self.assertEqual(fip_neutron['router_id'], 'router_uuid')
fip_obj.get_virtual_machine_interface_refs = \
lambda: [{'uuid': 'fip_port_uuid2'}]
fip_neutron = dbi._floatingip_vnc_to_neutron(fip_obj)
self.assertIsNone(fip_neutron['router_id'])