← Back to branch summary

~hudson-openstack/nova/trunk

« back to all changes in this revision

Viewing changes to nova/tests/network_unittest.py

  • Committer: Tarmac
  • Author(s): Vishvananda Ishaya
  • Date: 2010-08-14 21:13:44 UTC
  • mfrom: (198.4.31 network-refactor)
  • Revision ID: hudson@openstack.org-20100814211344-hqv3g8zfnf22251b
Improves pep8 compliance and pylint score in network code.

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/tests/network_unittest.py
15
15
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
16
16
#    License for the specific language governing permissions and limitations
17
17
#    under the License.
18
 
 
 
18
"""
 
19
Unit Tests for network code
 
20
"""
19
21
import IPy
20
22
import os
21
23
import logging
31
33
 
32
34
FLAGS = flags.FLAGS
33
35
 
 
36
 
34
37
class NetworkTestCase(test.TrialTestCase):
35
 
    def setUp(self):
 
38
    """Test cases for network code"""
 
39
    def setUp(self):  # pylint: disable=C0103
36
40
        super(NetworkTestCase, self).setUp()
37
41
        # NOTE(vish): if you change these flags, make sure to change the
38
42
        #             flags in the corresponding section in nova-dhcpbridge
43
47
                   network_size=32)
44
48
        logging.getLogger().setLevel(logging.DEBUG)
45
49
        self.manager = manager.AuthManager()
46
 
        self.dnsmasq = FakeDNSMasq()
47
50
        self.user = self.manager.create_user('netuser', 'netuser', 'netuser')
48
51
        self.projects = []
49
52
        self.projects.append(self.manager.create_project('netuser',
54
57
            self.projects.append(self.manager.create_project(name,
55
58
                                                             'netuser',
56
59
                                                             name))
57
 
        self.network = model.PublicNetworkController()
 
60
            vpn.NetworkData.create(self.projects[i].id)
58
61
        self.service = service.VlanNetworkService()
59
62
 
60
 
    def tearDown(self):
 
63
    def tearDown(self):  # pylint: disable=C0103
61
64
        super(NetworkTestCase, self).tearDown()
62
65
        for project in self.projects:
63
66
            self.manager.delete_project(project)
64
67
        self.manager.delete_user(self.user)
65
68
 
66
69
    def test_public_network_allocation(self):
 
70
        """Makes sure that we can allocaate a public ip"""
67
71
        pubnet = IPy.IP(flags.FLAGS.public_range)
68
 
        address = self.network.allocate_ip(self.user.id, self.projects[0].id, "public")
 
72
        address = self.service.allocate_elastic_ip(self.user.id,
 
73
                                           self.projects[0].id)
69
74
        self.assertTrue(IPy.IP(address) in pubnet)
70
 
        self.assertTrue(IPy.IP(address) in self.network.network)
71
75
 
72
76
    def test_allocate_deallocate_fixed_ip(self):
73
 
        result  = yield self.service.allocate_fixed_ip(
 
77
        """Makes sure that we can allocate and deallocate a fixed ip"""
 
78
        result = self.service.allocate_fixed_ip(
74
79
                self.user.id, self.projects[0].id)
75
80
        address = result['private_dns_name']
76
81
        mac = result['mac_address']
77
 
        logging.debug("Was allocated %s" % (address))
78
82
        net = model.get_project_network(self.projects[0].id, "default")
79
83
        self.assertEqual(True, is_in_project(address, self.projects[0].id))
80
84
        hostname = "test-host"
81
 
        self.dnsmasq.issue_ip(mac, address, hostname, net.bridge_name)
82
 
        rv = self.service.deallocate_fixed_ip(address)
 
85
        issue_ip(mac, address, hostname, net.bridge_name)
 
86
        self.service.deallocate_fixed_ip(address)
83
87
 
84
88
        # Doesn't go away until it's dhcp released
85
89
        self.assertEqual(True, is_in_project(address, self.projects[0].id))
86
90
 
87
 
        self.dnsmasq.release_ip(mac, address, hostname, net.bridge_name)
 
91
        release_ip(mac, address, hostname, net.bridge_name)
88
92
        self.assertEqual(False, is_in_project(address, self.projects[0].id))
89
93
 
90
 
    def test_range_allocation(self):
91
 
        hostname = "test-host"
92
 
        result = yield self.service.allocate_fixed_ip(
93
 
                    self.user.id, self.projects[0].id)
 
94
    def test_side_effects(self):
 
95
        """Ensures allocating and releasing has no side effects"""
 
96
        hostname = "side-effect-host"
 
97
        result = self.service.allocate_fixed_ip(self.user.id,
 
98
                                                self.projects[0].id)
94
99
        mac = result['mac_address']
95
100
        address = result['private_dns_name']
96
 
        result = yield self.service.allocate_fixed_ip(
97
 
                self.user, self.projects[1].id)
 
101
        result = self.service.allocate_fixed_ip(self.user,
 
102
                                                self.projects[1].id)
98
103
        secondmac = result['mac_address']
99
104
        secondaddress = result['private_dns_name']
100
105
 
102
107
        secondnet = model.get_project_network(self.projects[1].id, "default")
103
108
 
104
109
        self.assertEqual(True, is_in_project(address, self.projects[0].id))
105
 
        self.assertEqual(True, is_in_project(secondaddress, self.projects[1].id))
 
110
        self.assertEqual(True, is_in_project(secondaddress,
 
111
                                             self.projects[1].id))
106
112
        self.assertEqual(False, is_in_project(address, self.projects[1].id))
107
113
 
108
114
        # Addresses are allocated before they're issued
109
 
        self.dnsmasq.issue_ip(mac, address, hostname, net.bridge_name)
110
 
        self.dnsmasq.issue_ip(secondmac, secondaddress,
111
 
                                hostname, secondnet.bridge_name)
 
115
        issue_ip(mac, address, hostname, net.bridge_name)
 
116
        issue_ip(secondmac, secondaddress, hostname, secondnet.bridge_name)
112
117
 
113
 
        rv = self.service.deallocate_fixed_ip(address)
114
 
        self.dnsmasq.release_ip(mac, address, hostname, net.bridge_name)
 
118
        self.service.deallocate_fixed_ip(address)
 
119
        release_ip(mac, address, hostname, net.bridge_name)
115
120
        self.assertEqual(False, is_in_project(address, self.projects[0].id))
116
121
 
117
122
        # First address release shouldn't affect the second
118
 
        self.assertEqual(True, is_in_project(secondaddress, self.projects[1].id))
 
123
        self.assertEqual(True, is_in_project(secondaddress,
 
124
                                             self.projects[1].id))
119
125
 
120
 
        rv = self.service.deallocate_fixed_ip(secondaddress)
121
 
        self.dnsmasq.release_ip(secondmac, secondaddress,
122
 
                                hostname, secondnet.bridge_name)
123
 
        self.assertEqual(False, is_in_project(secondaddress, self.projects[1].id))
 
126
        self.service.deallocate_fixed_ip(secondaddress)
 
127
        release_ip(secondmac, secondaddress, hostname, secondnet.bridge_name)
 
128
        self.assertEqual(False, is_in_project(secondaddress,
 
129
                                              self.projects[1].id))
124
130
 
125
131
    def test_subnet_edge(self):
126
 
        result = yield self.service.allocate_fixed_ip(self.user.id,
 
132
        """Makes sure that private ips don't overlap"""
 
133
        result = self.service.allocate_fixed_ip(self.user.id,
127
134
                                                       self.projects[0].id)
128
135
        firstaddress = result['private_dns_name']
129
136
        hostname = "toomany-hosts"
130
 
        for i in range(1,5):
 
137
        for i in range(1, 5):
131
138
            project_id = self.projects[i].id
132
 
            result = yield self.service.allocate_fixed_ip(
 
139
            result = self.service.allocate_fixed_ip(
133
140
                    self.user, project_id)
134
141
            mac = result['mac_address']
135
142
            address = result['private_dns_name']
136
 
            result = yield self.service.allocate_fixed_ip(
 
143
            result = self.service.allocate_fixed_ip(
137
144
                    self.user, project_id)
138
145
            mac2 = result['mac_address']
139
146
            address2 = result['private_dns_name']
140
 
            result = yield self.service.allocate_fixed_ip(
 
147
            result = self.service.allocate_fixed_ip(
141
148
                   self.user, project_id)
142
149
            mac3 = result['mac_address']
143
150
            address3 = result['private_dns_name']
144
 
            self.assertEqual(False, is_in_project(address, self.projects[0].id))
145
 
            self.assertEqual(False, is_in_project(address2, self.projects[0].id))
146
 
            self.assertEqual(False, is_in_project(address3, self.projects[0].id))
147
 
            rv = self.service.deallocate_fixed_ip(address)
148
 
            rv = self.service.deallocate_fixed_ip(address2)
149
 
            rv = self.service.deallocate_fixed_ip(address3)
150
151
            net = model.get_project_network(project_id, "default")
151
 
            self.dnsmasq.release_ip(mac, address, hostname, net.bridge_name)
152
 
            self.dnsmasq.release_ip(mac2, address2, hostname, net.bridge_name)
153
 
            self.dnsmasq.release_ip(mac3, address3, hostname, net.bridge_name)
 
152
            issue_ip(mac, address, hostname, net.bridge_name)
 
153
            issue_ip(mac2, address2, hostname, net.bridge_name)
 
154
            issue_ip(mac3, address3, hostname, net.bridge_name)
 
155
            self.assertEqual(False, is_in_project(address,
 
156
                                                  self.projects[0].id))
 
157
            self.assertEqual(False, is_in_project(address2,
 
158
                                                  self.projects[0].id))
 
159
            self.assertEqual(False, is_in_project(address3,
 
160
                                                  self.projects[0].id))
 
161
            self.service.deallocate_fixed_ip(address)
 
162
            self.service.deallocate_fixed_ip(address2)
 
163
            self.service.deallocate_fixed_ip(address3)
 
164
            release_ip(mac, address, hostname, net.bridge_name)
 
165
            release_ip(mac2, address2, hostname, net.bridge_name)
 
166
            release_ip(mac3, address3, hostname, net.bridge_name)
154
167
        net = model.get_project_network(self.projects[0].id, "default")
155
 
        rv = self.service.deallocate_fixed_ip(firstaddress)
156
 
        self.dnsmasq.release_ip(mac, firstaddress, hostname, net.bridge_name)
 
168
        self.service.deallocate_fixed_ip(firstaddress)
 
169
        release_ip(mac, firstaddress, hostname, net.bridge_name)
157
170
 
158
 
    def test_212_vpn_ip_and_port_looks_valid(self):
159
 
        vpn.NetworkData.create(self.projects[0].id)
 
171
    def test_vpn_ip_and_port_looks_valid(self):
 
172
        """Ensure the vpn ip and port are reasonable"""
160
173
        self.assert_(self.projects[0].vpn_ip)
161
174
        self.assert_(self.projects[0].vpn_port >= FLAGS.vpn_start_port)
162
175
        self.assert_(self.projects[0].vpn_port <= FLAGS.vpn_end_port)
163
176
 
164
177
    def test_too_many_vpns(self):
 
178
        """Ensure error is raised if we run out of vpn ports"""
165
179
        vpns = []
166
180
        for i in xrange(vpn.NetworkData.num_ports_for_ip(FLAGS.vpn_ip)):
167
181
            vpns.append(vpn.NetworkData.create("vpnuser%s" % i))
169
183
        for network_datum in vpns:
170
184
            network_datum.destroy()
171
185
 
172
 
    def test_release_before_deallocate(self):
173
 
        pass
174
 
 
175
 
    def test_deallocate_before_issued(self):
176
 
        pass
177
 
 
178
 
    def test_too_many_addresses(self):
179
 
        """
180
 
        Here, we test that a proper NoMoreAddresses exception is raised.
181
 
 
182
 
        However, the number of available IP addresses depends on the test
 
186
    def test_ips_are_reused(self):
 
187
        """Makes sure that ip addresses that are deallocated get reused"""
 
188
        result = self.service.allocate_fixed_ip(
 
189
                    self.user.id, self.projects[0].id)
 
190
        mac = result['mac_address']
 
191
        address = result['private_dns_name']
 
192
 
 
193
        hostname = "reuse-host"
 
194
        net = model.get_project_network(self.projects[0].id, "default")
 
195
 
 
196
        issue_ip(mac, address, hostname, net.bridge_name)
 
197
        self.service.deallocate_fixed_ip(address)
 
198
        release_ip(mac, address, hostname, net.bridge_name)
 
199
 
 
200
        result = self.service.allocate_fixed_ip(
 
201
                self.user, self.projects[0].id)
 
202
        secondmac = result['mac_address']
 
203
        secondaddress = result['private_dns_name']
 
204
        self.assertEqual(address, secondaddress)
 
205
        self.service.deallocate_fixed_ip(secondaddress)
 
206
        issue_ip(secondmac, secondaddress, hostname, net.bridge_name)
 
207
        release_ip(secondmac, secondaddress, hostname, net.bridge_name)
 
208
 
 
209
    def test_available_ips(self):
 
210
        """Make sure the number of available ips for the network is correct
 
211
 
 
212
        The number of available IP addresses depends on the test
183
213
        environment's setup.
184
214
 
185
215
        Network size is set in test fixture's setUp method.
186
216
 
187
 
        There are FLAGS.cnt_vpn_clients addresses reserved for VPN (NUM_RESERVED_VPN_IPS)
188
 
 
189
 
        And there are NUM_STATIC_IPS that are always reserved by Nova for the necessary
190
 
        services (gateway, CloudPipe, etc)
191
 
 
192
 
        So we should get flags.network_size - (NUM_STATIC_IPS +
193
 
                                               NUM_PREALLOCATED_IPS +
194
 
                                               NUM_RESERVED_VPN_IPS)
195
 
        usable addresses
 
217
        There are ips reserved at the bottom and top of the range.
 
218
        services (network, gateway, CloudPipe, broadcast)
196
219
        """
197
220
        net = model.get_project_network(self.projects[0].id, "default")
198
 
 
199
 
        # Determine expected number of available IP addresses
200
 
        num_static_ips = net.num_static_ips
201
221
        num_preallocated_ips = len(net.hosts.keys())
202
 
        num_reserved_vpn_ips = flags.FLAGS.cnt_vpn_clients
203
 
        num_available_ips = flags.FLAGS.network_size - (num_static_ips +
204
 
                                                        num_preallocated_ips +
205
 
                                                        num_reserved_vpn_ips)
 
222
        net_size = flags.FLAGS.network_size
 
223
        num_available_ips = net_size - (net.num_bottom_reserved_ips +
 
224
                                        num_preallocated_ips +
 
225
                                        net.num_top_reserved_ips)
 
226
        self.assertEqual(num_available_ips, len(list(net.available)))
 
227
 
 
228
    def test_too_many_addresses(self):
 
229
        """Test for a NoMoreAddresses exception when all fixed ips are used.
 
230
        """
 
231
        net = model.get_project_network(self.projects[0].id, "default")
206
232
 
207
233
        hostname = "toomany-hosts"
208
234
        macs = {}
209
235
        addresses = {}
210
 
        for i in range(0, (num_available_ips - 1)):
211
 
            result = yield self.service.allocate_fixed_ip(self.user.id, self.projects[0].id)
 
236
        # Number of availaible ips is len of the available list
 
237
        num_available_ips = len(list(net.available))
 
238
        for i in range(num_available_ips):
 
239
            result = self.service.allocate_fixed_ip(self.user.id,
 
240
                                                    self.projects[0].id)
212
241
            macs[i] = result['mac_address']
213
242
            addresses[i] = result['private_dns_name']
214
 
            self.dnsmasq.issue_ip(macs[i], addresses[i], hostname, net.bridge_name)
215
 
 
216
 
        self.assertFailure(self.service.allocate_fixed_ip(self.user.id, self.projects[0].id), NoMoreAddresses)
217
 
 
218
 
        for i in range(0, (num_available_ips - 1)):
219
 
            rv = self.service.deallocate_fixed_ip(addresses[i])
220
 
            self.dnsmasq.release_ip(macs[i], addresses[i], hostname, net.bridge_name)
 
243
            issue_ip(macs[i], addresses[i], hostname, net.bridge_name)
 
244
 
 
245
        self.assertEqual(len(list(net.available)), 0)
 
246
        self.assertRaises(NoMoreAddresses, self.service.allocate_fixed_ip,
 
247
                          self.user.id, self.projects[0].id)
 
248
 
 
249
        for i in range(len(addresses)):
 
250
            self.service.deallocate_fixed_ip(addresses[i])
 
251
            release_ip(macs[i], addresses[i], hostname, net.bridge_name)
 
252
        self.assertEqual(len(list(net.available)), num_available_ips)
 
253
 
221
254
 
222
255
def is_in_project(address, project_id):
 
256
    """Returns true if address is in specified project"""
223
257
    return address in model.get_project_network(project_id).list_addresses()
224
258
 
225
 
def _get_project_addresses(project_id):
226
 
    project_addresses = []
227
 
    for addr in model.get_project_network(project_id).list_addresses():
228
 
        project_addresses.append(addr)
229
 
    return project_addresses
230
259
 
231
260
def binpath(script):
 
261
    """Returns the absolute path to a script in bin"""
232
262
    return os.path.abspath(os.path.join(__file__, "../../../bin", script))
233
263
 
234
 
class FakeDNSMasq(object):
235
 
    def issue_ip(self, mac, ip, hostname, interface):
236
 
        cmd = "%s add %s %s %s" % (binpath('nova-dhcpbridge'),
237
 
                                   mac, ip, hostname)
238
 
        env = {'DNSMASQ_INTERFACE': interface,
239
 
               'TESTING' : '1',
240
 
               'FLAGFILE' : FLAGS.dhcpbridge_flagfile}
241
 
        (out, err) = utils.execute(cmd, addl_env=env)
242
 
        logging.debug("ISSUE_IP: %s, %s " % (out, err))
243
 
 
244
 
    def release_ip(self, mac, ip, hostname, interface):
245
 
        cmd = "%s del %s %s %s" % (binpath('nova-dhcpbridge'),
246
 
                                   mac, ip, hostname)
247
 
        env = {'DNSMASQ_INTERFACE': interface,
248
 
               'TESTING' : '1',
249
 
               'FLAGFILE' : FLAGS.dhcpbridge_flagfile}
250
 
        (out, err) = utils.execute(cmd, addl_env=env)
251
 
        logging.debug("RELEASE_IP: %s, %s " % (out, err))
252
 
 
 
264
 
 
265
def issue_ip(mac, private_ip, hostname, interface):
 
266
    """Run add command on dhcpbridge"""
 
267
    cmd = "%s add %s %s %s" % (binpath('nova-dhcpbridge'),
 
268
                               mac, private_ip, hostname)
 
269
    env = {'DNSMASQ_INTERFACE': interface,
 
270
           'TESTING': '1',
 
271
           'FLAGFILE': FLAGS.dhcpbridge_flagfile}
 
272
    (out, err) = utils.execute(cmd, addl_env=env)
 
273
    logging.debug("ISSUE_IP: %s, %s ", out, err)
 
274
 
 
275
def release_ip(mac, private_ip, hostname, interface):
 
276
    """Run del command on dhcpbridge"""
 
277
    cmd = "%s del %s %s %s" % (binpath('nova-dhcpbridge'),
 
278
                               mac, private_ip, hostname)
 
279
    env = {'DNSMASQ_INTERFACE': interface,
 
280
           'TESTING': '1',
 
281
           'FLAGFILE': FLAGS.dhcpbridge_flagfile}
 
282
    (out, err) = utils.execute(cmd, addl_env=env)
 
283
    logging.debug("RELEASE_IP: %s, %s ", out, err)