Coverage for manila/tests/test_network.py: 78%

1# Copyright 2015 Mirantis Inc. 

2# All Rights Reserved. 

3# 

4# Licensed under the Apache License, Version 2.0 (the "License"); you may 

5# not use this file except in compliance with the License. You may obtain 

6# a copy of the License at 

7# 

8# http://www.apache.org/licenses/LICENSE-2.0 

9# 

10# Unless required by applicable law or agreed to in writing, software 

11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 

12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 

13# License for the specific language governing permissions and limitations 

14# under the License. 

15 

16import ddt 

17from oslo_config import cfg 

18from oslo_utils import importutils 

19 

20from manila import exception 

21from manila import network 

22from manila import test 

23 

24CONF = cfg.CONF 

25 

26 

27@ddt.ddt 

28class APITestCase(test.TestCase): 

29 

30 def setUp(self): 

31 super(APITestCase, self).setUp() 

32 self.mock_object(importutils, 'import_class') 

33 

34 def test_init_api_with_default_config_group_name(self): 

35 network.API() 

36 

37 importutils.import_class.assert_called_once_with( 

38 CONF.network_api_class) 

39 importutils.import_class.return_value.assert_called_once_with( 

40 config_group_name=None, label='user') 

41 

42 def test_init_api_with_custom_config_group_name(self): 

43 group_name = 'FOO_GROUP_NAME' 

44 

45 network.API(config_group_name=group_name) 

46 

47 importutils.import_class.assert_called_once_with( 

48 getattr(CONF, group_name).network_api_class) 

49 importutils.import_class.return_value.assert_called_once_with( 

50 config_group_name=group_name, label='user') 

51 

52 def test_init_api_with_custom_config_group_name_and_label(self): 

53 group_name = 'FOO_GROUP_NAME' 

54 label = 'custom_label' 

55 

56 network.API(config_group_name=group_name, label=label) 

57 

58 importutils.import_class.assert_called_once_with( 

59 getattr(CONF, group_name).network_api_class) 

60 importutils.import_class.return_value.assert_called_once_with( 

61 config_group_name=group_name, label=label) 

62 

63 

64@ddt.ddt 

65class NetworkBaseAPITestCase(test.TestCase): 

66 

67 def setUp(self): 

68 super(NetworkBaseAPITestCase, self).setUp() 

69 self.db_driver = 'fake_driver' 

70 self.mock_object(importutils, 'import_module') 

71 

72 def test_inherit_network_base_api_no_redefinitions(self): 

73 class FakeNetworkAPI(network.NetworkBaseAPI): 

74 pass 

75 

76 self.assertRaises(TypeError, FakeNetworkAPI) 

77 

78 def test_inherit_network_base_api_deallocate_not_redefined(self): 

79 class FakeNetworkAPI(network.NetworkBaseAPI): 

80 def allocate_network(self, *args, **kwargs): 

81 pass 

82 

83 def manage_network_allocations( 

84 self, context, allocations, share_server, 

85 share_network=None): 

86 pass 

87 

88 def unmanage_network_allocations(self, context, share_server_id): 

89 pass 

90 

91 def include_network_info(self, share_network_subnet): 

92 pass 

93 

94 self.assertRaises(TypeError, FakeNetworkAPI) 

95 

96 def test_inherit_network_base_api_allocate_not_redefined(self): 

97 class FakeNetworkAPI(network.NetworkBaseAPI): 

98 def deallocate_network(self, *args, **kwargs): 

99 pass 

100 

101 def manage_network_allocations( 

102 self, context, allocations, share_server, 

103 share_network=None): 

104 pass 

105 

106 def unmanage_network_allocations(self, context, share_server_id): 

107 pass 

108 

109 def include_network_info(self, share_network_subnet): 

110 pass 

111 

112 self.assertRaises(TypeError, FakeNetworkAPI) 

113 

114 def test_inherit_network_base_api(self): 

115 class FakeNetworkAPI(network.NetworkBaseAPI): 

116 def allocate_network(self, *args, **kwargs): 

117 pass 

118 

119 def deallocate_network(self, *args, **kwargs): 

120 pass 

121 

122 def manage_network_allocations( 

123 self, context, allocations, share_server, 

124 share_network=None): 

125 pass 

126 

127 def unmanage_network_allocations(self, context, share_server_id): 

128 pass 

129 

130 def include_network_info(self, share_network_subnet): 

131 pass 

132 

133 result = FakeNetworkAPI() 

134 

135 self.assertTrue(hasattr(result, '_verify_share_network')) 

136 self.assertTrue(hasattr(result, 'allocate_network')) 

137 self.assertTrue(hasattr(result, 'deallocate_network')) 

138 

139 def test__verify_share_network_ok(self): 

140 class FakeNetworkAPI(network.NetworkBaseAPI): 

141 def allocate_network(self, *args, **kwargs): 

142 pass 

143 

144 def deallocate_network(self, *args, **kwargs): 

145 pass 

146 

147 def manage_network_allocations( 

148 self, context, allocations, share_server, 

149 share_network=None): 

150 pass 

151 

152 def unmanage_network_allocations(self, context, share_server_id): 

153 pass 

154 

155 def include_network_info(self, share_network_subnet): 

156 pass 

157 

158 result = FakeNetworkAPI() 

159 

160 result._verify_share_network('foo_id', {'id': 'bar_id'}) 

161 

162 def test__verify_share_network_fail(self): 

163 class FakeNetworkAPI(network.NetworkBaseAPI): 

164 def allocate_network(self, *args, **kwargs): 

165 pass 

166 

167 def deallocate_network(self, *args, **kwargs): 

168 pass 

169 

170 def manage_network_allocations( 

171 self, context, allocations, share_server, 

172 share_network=None): 

173 pass 

174 

175 def unmanage_network_allocations(self, context, share_server_id): 

176 pass 

177 

178 def include_network_info(self, share_network_subnet): 

179 pass 

180 

181 result = FakeNetworkAPI() 

182 

183 self.assertRaises( 

184 exception.NetworkBadConfigurationException, 

185 result._verify_share_network, 'foo_id', None) 

186 

187 @ddt.data((True, False, set([6])), (False, True, set([4])), 

188 (True, True, set([4, 6])), (False, False, set())) 

189 @ddt.unpack 

190 def test_enabled_ip_versions(self, network_plugin_ipv6_enabled, 

191 network_plugin_ipv4_enabled, 

192 enable_ip_versions): 

193 class FakeNetworkAPI(network.NetworkBaseAPI): 

194 def allocate_network(self, *args, **kwargs): 

195 pass 

196 

197 def deallocate_network(self, *args, **kwargs): 

198 pass 

199 

200 def manage_network_allocations( 

201 self, context, allocations, share_server, 

202 share_network=None): 

203 pass 

204 

205 def unmanage_network_allocations(self, context, share_server_id): 

206 pass 

207 

208 def include_network_info(self, share_network_subnet): 

209 pass 

210 

211 network.CONF.set_default( 

212 'network_plugin_ipv6_enabled', network_plugin_ipv6_enabled) 

213 network.CONF.set_default( 

214 'network_plugin_ipv4_enabled', network_plugin_ipv4_enabled) 

215 

216 result = FakeNetworkAPI() 

217 

218 if enable_ip_versions: 

219 self.assertTrue(hasattr(result, 'enabled_ip_versions')) 

220 self.assertEqual(enable_ip_versions, result.enabled_ip_versions) 

221 else: 

222 self.assertRaises(exception.NetworkBadConfigurationException, 

223 getattr, result, 'enabled_ip_versions')