Statistics
| Branch: | Tag: | Revision:

root / snf-cyclades-app / synnefo / api / tests / ports.py @ a3b8aaf6

History | View | Annotate | Download (9.4 kB)

1
# Copyright 2011-2013 GRNET S.A. All rights reserved.
2
#
3
# Redistribution and use in source and binary forms, with or
4
# without modification, are permitted provided that the following
5
# conditions are met:
6
#
7
#   1. Redistributions of source code must retain the above
8
#      copyright notice, this list of conditions and the following
9
#      disclaimer.
10
#
11
#   2. Redistributions in binary form must reproduce the above
12
#      copyright notice, this list of conditions and the following
13
#      disclaimer in the documentation and/or other materials
14
#      provided with the distribution.
15
#
16
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
17
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
18
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
19
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
20
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
21
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
22
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
23
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
24
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
25
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
26
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
27
# POSSIBILITY OF SUCH DAMAGE.
28
#
29
# The views and conclusions contained in the software and
30
# documentation are those of the authors and should not be
31
# interpreted as representing official policies, either expressed
32
# or implied, of GRNET S.A.i
33

    
34
from snf_django.utils.testing import (BaseAPITest, override_settings)
35
from snf_django.utils.testing import BaseAPITest
36
from django.utils import simplejson as json
37
from synnefo.cyclades_settings import cyclades_services
38
from synnefo.lib.services import get_service_path
39
from synnefo.lib import join_urls
40
import synnefo.db.models_factory as dbmf
41

    
42
COMPUTE_URL = get_service_path(cyclades_services, 'compute',
43
                               version='v2.0')
44
PORTS_URL = join_urls(COMPUTE_URL, "ports/")
45

    
46

    
47
class PortTest(BaseAPITest):
48
    def test_get_ports(self):
49
        url = join_urls(PORTS_URL)
50
        response = self.get(url)
51
        self.assertEqual(response.status_code, 200)
52
        ports = json.loads(response.content)
53
        self.assertEqual(ports, {"ports": []})
54

    
55
    def test_get_port_unfound(self):
56
        url = join_urls(PORTS_URL, "123")
57
        response = self.get(url)
58
        self.assertEqual(response.status_code, 404)
59

    
60
    def test_get_port(self):
61
        nic = dbmf.NetworkInterfaceFactory.create()
62
        url = join_urls(PORTS_URL, str(nic.id))
63
        response = self.get(url, user=nic.network.userid)
64
        self.assertEqual(response.status_code, 200)
65

    
66
    def test_delete_port(self):
67
        nic = dbmf.NetworkInterfaceFactory.create(device_owner='vm')
68
        url = join_urls(PORTS_URL, str(nic.id))
69
        response = self.delete(url, user=nic.network.userid)
70
        self.assertEqual(response.status_code, 204)
71

    
72
    def test_remove_nic_malformed(self):
73
        url = join_urls(PORTS_URL, "123")
74
        response = self.delete(url)
75
        self.assertItemNotFound(response)
76

    
77
    def test_update_port_name(self):
78
        nic = dbmf.NetworkInterfaceFactory.create(device_owner='vm')
79
        url = join_urls(PORTS_URL, str(nic.id))
80
        request = {'port': {"name": "test-name"}}
81
        response = self.put(url, params=json.dumps(request),
82
                            user=nic.network.userid)
83
        self.assertEqual(response.status_code, 200)
84
        res = json.loads(response.content)
85
        self.assertEqual(res['port']['name'], "test-name")
86

    
87
    def test_update_port_sg_unfound(self):
88
        sg1 = dbmf.SecurityGroupFactory.create()
89
        nic = dbmf.NetworkInterfaceFactory.create(device_owner='vm')
90
        nic.security_groups.add(sg1)
91
        nic.save()
92
        url = join_urls(PORTS_URL, str(nic.id))
93
        request = {'port': {"security_groups": ["123"]}}
94
        response = self.put(url, params=json.dumps(request),
95
                            user=nic.network.userid)
96
        self.assertEqual(response.status_code, 404)
97

    
98
    def test_update_port_sg(self):
99
        sg1 = dbmf.SecurityGroupFactory.create()
100
        sg2 = dbmf.SecurityGroupFactory.create()
101
        sg3 = dbmf.SecurityGroupFactory.create()
102
        nic = dbmf.NetworkInterfaceFactory.create(device_owner='vm')
103
        nic.security_groups.add(sg1)
104
        nic.save()
105
        url = join_urls(PORTS_URL, str(nic.id))
106
        request = {'port': {"security_groups": [str(sg2.id), str(sg3.id)]}}
107
        response = self.put(url, params=json.dumps(request),
108
                            user=nic.network.userid)
109
        res = json.loads(response.content)
110
        self.assertEqual(res['port']['security_groups'],
111
                         [str(sg2.id), str(sg3.id)])
112

    
113
    def test_create_port_no_network(self):
114
        request = {
115
            "port": {
116
                "device_id": "123",
117
                "name": "port1",
118
                "network_id": "123"
119
            }
120
        }
121
        response = self.post(PORTS_URL, params=json.dumps(request))
122
        self.assertEqual(response.status_code, 404)
123

    
124
    def test_create_port(self):
125
        net = dbmf.NetworkFactory.create()
126
        subnet1 = dbmf.IPv4SubnetFactory.create(network=net)
127
        subnet2 = dbmf.IPv6SubnetFactory.create(network=net)
128
        sg1 = dbmf.SecurityGroupFactory.create()
129
        sg2 = dbmf.SecurityGroupFactory.create()
130
        vm = dbmf.VirtualMachineFactory.create(userid=net.userid)
131
        request = {
132
            "port": {
133
                "name": "port1",
134
                "network_id": str(net.id),
135
                "device_id": str(vm.id),
136
                "security_groups": [str(sg1.id), str(sg2.id)]
137
            }
138
        }
139
        response = self.post(PORTS_URL, params=json.dumps(request),
140
                             user=net.userid)
141
        self.assertEqual(response.status_code, 201)
142

    
143
    def test_add_nic_to_deleted_network(self):
144
        user = 'userr'
145
        vm = dbmf.VirtualMachineFactory(name='yo', userid=user,
146
                                            operstate="ACTIVE")
147
        net = dbmf.NetworkFactory(state='ACTIVE', userid=user,
148
                                      deleted=True)
149
        request = {
150
            "port": {
151
                "device_id": str(vm.id),
152
                "name": "port1",
153
                "network_id": str(net.id)
154
            }
155
        }
156
        response = self.post(PORTS_URL, params=json.dumps(request),
157
                             user=net.userid)
158
        self.assertBadRequest(response)
159

    
160
    def test_add_nic_to_public_network(self):
161
        user = 'userr'
162
        vm = dbmf.VirtualMachineFactory(name='yo', userid=user)
163
        net = dbmf.NetworkFactory(state='ACTIVE', userid=user, public=True)
164
        request = {
165
            "port": {
166
                "device_id": str(vm.id),
167
                "name": "port1",
168
                "network_id": str(net.id)
169
            }
170
        }
171
        response = self.post(PORTS_URL, params=json.dumps(request),
172
                             user=net.userid)
173
        self.assertFault(response, 403, 'forbidden')
174

    
175
    def test_add_nic_malformed_1(self):
176
        user = 'userr'
177
        vm = dbmf.VirtualMachineFactory(name='yo', userid=user)
178
        net = dbmf.NetworkFactory(state='ACTIVE', userid=user)
179
        request = {
180
            "port": {
181
                "name": "port1",
182
                "network_id": str(net.id)
183
            }
184
        }
185
        response = self.post(PORTS_URL, params=json.dumps(request),
186
                             user=net.userid)
187
        self.assertBadRequest(response)
188

    
189
    def test_add_nic_malformed_2(self):
190
        user = 'userr'
191
        vm = dbmf.VirtualMachineFactory(name='yo', userid=user)
192
        net = dbmf.NetworkFactory(state='ACTIVE', userid=user)
193
        request = {
194
            "port": {
195
                "device_id": str(vm.id),
196
                "name": "port1"
197
            }
198
        }
199
        response = self.post(PORTS_URL, params=json.dumps(request),
200
                             user=net.userid)
201
        self.assertBadRequest(response)
202

    
203
    def test_add_nic_not_active(self):
204
        """Test connecting VM to non-active network"""
205
        user = 'dummy'
206
        vm = dbmf.VirtualMachineFactory(name='yo', userid=user)
207
        net = dbmf.NetworkFactory(state='PENDING', userid=user)
208
        request = {
209
            "port": {
210
                "device_id": str(vm.id),
211
                "name": "port1",
212
                "network_id": str(net.id)
213
            }
214
        }
215
        response = self.post(PORTS_URL, params=json.dumps(request),
216
                             user=net.userid)
217
        # Test that returns BuildInProgress
218
        self.assertEqual(response.status_code, 409)
219

    
220
#    def test_add_nic_full_network(self, mrapi):
221
#        """Test connecting VM to a full network"""
222
#        user = 'userr'
223
#        vm = dbmf.VirtualMachineFactory(name='yo', userid=user,
224
#                                            operstate="STARTED")
225
#        net = dbmf.NetworkFactory(state='ACTIVE', subnet='10.0.0.0/30',
226
#                                      userid=user, dhcp=True)
227
#        pool = net.get_pool()
228
#        while not pool.empty():
229
#            pool.get()
230
#        pool.save()
231
#        pool = net.get_pool()
232
#        self.assertTrue(pool.empty())
233
#        request = {'add': {'serverRef': vm.id}}
234
#        response = self.mypost('networks/%d/action' % net.id,
235
#                               net.userid, json.dumps(request), 'json')
236
#        # Test that returns OverLimit
237
#        self.assertEqual(response.status_code, 413)
238
#        self.assertFalse(mrapi.called)