Statistics
| Branch: | Tag: | Revision:

root / snf-cyclades-app / synnefo / helpdesk / tests.py @ 88329243

History | View | Annotate | Download (12.5 kB)

1
# Copyright 2011 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.
33
#
34

    
35
import mock
36

    
37
from django.test import TestCase, Client
38
from django.conf import settings
39
from django.core.urlresolvers import reverse
40

    
41

    
42
USER1 = "5edcb5aa-1111-4146-a8ed-2b6287824353"
43
USER2 = "5edcb5aa-2222-4146-a8ed-2b6287824353"
44

    
45
USERS_UUIDS = {}
46
USERS_UUIDS[USER1] = {'displayname': 'testuser@test.com'}
47
USERS_UUIDS[USER2] = {'displayname': 'testuser2@test.com'}
48

    
49
USERS_DISPLAYNAMES = dict(map(lambda k: (k[1]['displayname'], {'uuid': k[0]}),
50
                          USERS_UUIDS.iteritems()))
51

    
52
from synnefo.db import models_factory as mfactory
53

    
54

    
55
class AstakosClientMock():
56
    def __init__(*args, **kwargs):
57
        pass
58

    
59
    def get_username(self, token, uuid):
60
        try:
61
            return USERS_UUIDS.get(uuid)['displayname']
62
        except TypeError:
63
            return None
64

    
65
    def get_uuid(self, token, display_name):
66
        try:
67
            return USERS_DISPLAYNAMES.get(display_name)['uuid']
68
        except TypeError:
69
            return None
70

    
71

    
72
class AuthClient(Client):
73

    
74
    def request(self, **request):
75
        token = request.pop('user_token', '0000')
76
        if token:
77
            request['HTTP_X_AUTH_TOKEN'] = token
78
        return super(AuthClient, self).request(**request)
79

    
80

    
81
def get_user_mock(request, *args, **kwargs):
82
    request.user_uniq = None
83
    request.user = None
84
    if request.META.get('HTTP_X_AUTH_TOKEN', None) == '0000':
85
        request.user_uniq = 'test'
86
        request.user = {'uniq': 'test', 'auth_token': '0000'}
87
    if request.META.get('HTTP_X_AUTH_TOKEN', None) == '0001':
88
        request.user_uniq = 'test'
89
        request.user = {'uniq': 'test', 'groups': ['default',
90
                                                   'helpdesk'],
91
                        'auth_token': '0001'}
92

    
93

    
94
@mock.patch("astakosclient.AstakosClient", new=AstakosClientMock)
95
@mock.patch("snf_django.lib.astakos.get_user", new=get_user_mock)
96
class HelpdeskTests(TestCase):
97
    """
98
    Helpdesk tests. Test correctness of permissions and returned data.
99
    """
100

    
101
    def setUp(self):
102
        settings.SKIP_SSH_VALIDATION = True
103
        settings.HELPDESK_ENABLED = True
104
        self.client = AuthClient()
105

    
106
        # init models
107
        vm1u1 = mfactory.VirtualMachineFactory(userid=USER1, name="user1 vm",
108
                                               pk=1001)
109
        vm1u2 = mfactory.VirtualMachineFactory(userid=USER2, name="user2 vm1",
110
                                               pk=1002)
111
        vm2u2 = mfactory.VirtualMachineFactory(userid=USER2, name="user2 vm2",
112
                                               pk=1003)
113

    
114
        nic1 = mfactory.NetworkInterfaceFactory(machine=vm1u2,
115
                                                userid=vm1u2.userid,
116
                                                network__public=False,
117
                                                network__userid=USER1)
118
        ip2 = mfactory.IPv4AddressFactory(nic__machine=vm1u1,
119
                                          userid=vm1u1.userid,
120
                                          network__public=True,
121
                                          network__userid=None,
122
                                          address="195.251.222.211")
123

    
124
    def test_enabled_setting(self):
125
        settings.HELPDESK_ENABLED = False
126

    
127
        # helpdesk is disabled
128
        r = self.client.get(reverse('helpdesk-index'), user_token="0001")
129
        self.assertEqual(r.status_code, 404)
130
        r = self.client.get(reverse('helpdesk-details',
131
                                    args=['testuser@test.com']),
132
                            user_token="0001")
133
        self.assertEqual(r.status_code, 404)
134

    
135
    def test_ip_lookup(self):
136
        # ip does not exist, proper message gets displayed
137
        r = self.client.get(reverse('helpdesk-details',
138
                            args=["195.251.221.122"]), user_token='0001')
139
        self.assertContains(r, 'User with IP')
140

    
141
        # ip exists, 'test' account discovered
142
        r = self.client.get(reverse('helpdesk-details',
143
                            args=["195.251.222.211"]), user_token='0001')
144
        self.assertEqual(r.context['account'], USER1)
145

    
146
    def test_vm_lookup(self):
147
        # vm id does not exist
148
        r = self.client.get(reverse('helpdesk-details',
149
                            args=["vm-123"]), user_token='0001')
150
        self.assertContains(r, 'User with Virtual Machine')
151

    
152
        # vm exists, 'test' account discovered
153
        r = self.client.get(reverse('helpdesk-details',
154
                            args=["vm1001"]), user_token='0001')
155
        self.assertEqual(r.context['account'], USER1)
156
        r = self.client.get(reverse('helpdesk-details',
157
                            args=["vm1002"]), user_token='0001')
158
        self.assertEqual(r.context['account'], USER2)
159
        # dash also works
160
        r = self.client.get(reverse('helpdesk-details',
161
                            args=["vm-1002"]), user_token='0001')
162
        self.assertEqual(r.context['account'], USER2)
163

    
164
    def test_view_permissions(self):
165
        # anonymous user gets 403
166
        r = self.client.get(reverse('helpdesk-index'), user_token=None)
167
        self.assertEqual(r.status_code, 403)
168
        r = self.client.get(reverse('helpdesk-details',
169
                                    args=['testuser@test.com']),
170
                            user_token=None)
171
        self.assertEqual(r.status_code, 403)
172

    
173
        # user not in helpdesk group gets 403
174
        r = self.client.get(reverse('helpdesk-index'))
175
        self.assertEqual(r.status_code, 403)
176
        r = self.client.get(reverse('helpdesk-details',
177
                                    args=['testuser@test.com']))
178
        self.assertEqual(r.status_code, 403)
179

    
180
        # user with helpdesk group gets 200
181
        r = self.client.get(reverse('helpdesk-index'), user_token="0001")
182
        self.assertEqual(r.status_code, 200)
183
        r = self.client.get(reverse('helpdesk-details',
184
                                    args=['testuser@test.com']),
185
                            user_token="0001")
186
        self.assertEqual(r.status_code, 200)
187

    
188
        r = self.client.post(reverse('helpdesk-suspend-vm', args=(1001,)))
189
        r = self.client.get(reverse('helpdesk-suspend-vm', args=(1001,)),
190
                            user_token="0001", data={'token': '1234'})
191
        self.assertEqual(r.status_code, 403)
192
        r = self.client.get(reverse('helpdesk-suspend-vm', args=(1001,)),
193
                            user_token="0001")
194
        self.assertEqual(r.status_code, 403)
195
        r = self.client.post(reverse('helpdesk-suspend-vm', args=(1001,)),
196
                             user_token="0001", data={'token': '0001'})
197
        self.assertEqual(r.status_code, 302)
198
        r = self.client.post(reverse('helpdesk-suspend-vm', args=(1001,)),
199
                             user_token="0000", data={'token': '0000'})
200
        self.assertEqual(r.status_code, 403)
201

    
202
    def test_suspend_vm(self):
203
        r = self.client.get(reverse('helpdesk-details',
204
                                    args=['testuser@test.com']),
205
                            user_token="0001")
206
        self.assertEqual(r.status_code, 200)
207
        vmid = r.context['vms'][0].pk
208
        r = self.client.post(reverse('helpdesk-suspend-vm', args=(vmid,)),
209
                             data={'token': '0001'}, user_token="0001")
210
        self.assertEqual(r.status_code, 302)
211

    
212
        r = self.client.get(reverse('helpdesk-details',
213
                                    args=['testuser@test.com']),
214
                            user_token="0001")
215
        self.assertTrue(r.context['vms'][0].suspended)
216

    
217
        r = self.client.post(reverse('helpdesk-suspend-vm-release',
218
                                     args=(vmid,)), data={'token': '0001'},
219
                             user_token="0001")
220
        self.assertEqual(r.status_code, 302)
221
        r = self.client.get(reverse('helpdesk-details',
222
                                    args=['testuser@test.com']),
223
                            user_token="0001")
224
        self.assertFalse(r.context['vms'][0].suspended)
225

    
226
    def test_results_get_filtered(self):
227
        """
228
        Test that view context data are filtered based on userid provided.
229
        Check helpdesk_test.json to see the existing database data.
230
        """
231

    
232
        # 'testuser@test.com' details, see
233
        # helpdes/fixtures/helpdesk_test.json for more details
234
        r = self.client.get(reverse('helpdesk-details',
235
                                    args=['testuser@test.com']),
236
                            user_token="0001")
237
        account = r.context['account']
238
        vms = r.context['vms']
239
        nets = r.context['networks']
240
        self.assertEqual(account, USER1)
241
        self.assertEqual(vms[0].name, "user1 vm")
242
        self.assertEqual(vms.count(), 1)
243
        self.assertEqual(len(nets), 2)
244
        self.assertEqual(r.context['account_exists'], True)
245

    
246
        # 'testuser2@test.com' details, see helpdesk
247
        # /fixtures/helpdesk_test.json for more details
248
        r = self.client.get(reverse('helpdesk-details',
249
                                    args=['testuser2@test.com']),
250
                            user_token="0001")
251
        account = r.context['account']
252
        vms = r.context['vms']
253
        nets = r.context['networks']
254
        self.assertEqual(account, USER2)
255
        self.assertEqual(vms.count(), 2)
256
        self.assertEqual(sorted([vms[0].name, vms[1].name]),
257
                         sorted(["user2 vm1", "user2 vm2"]))
258
        self.assertEqual(len(nets), 0)
259
        self.assertEqual(r.context['account_exists'], True)
260

    
261
        # 'testuser5@test.com' does not exist, should be redirected to
262
        # helpdesk home
263
        r = self.client.get(reverse('helpdesk-details',
264
                                    args=['testuser5@test.com']),
265
                            user_token="0001")
266
        vms = r.context['vms']
267
        self.assertEqual(r.context['account_exists'], False)
268
        self.assertEqual(vms.count(), 0)
269

    
270
    def test_start_shutdown(self):
271
        from synnefo.logic import backend
272

    
273
        self.vm1 = mfactory.VirtualMachineFactory(userid=USER1)
274
        pk = self.vm1.pk
275

    
276
        r = self.client.post(reverse('helpdesk-vm-shutdown', args=(pk,)))
277
        self.assertEqual(r.status_code, 403)
278

    
279
        r = self.client.post(reverse('helpdesk-vm-shutdown', args=(pk,)),
280
                             data={'token': '0001'})
281
        self.assertEqual(r.status_code, 403)
282

    
283
        backend.shutdown_instance = shutdown = mock.Mock()
284
        shutdown.return_value = 1
285
        self.vm1.operstate = 'STARTED'
286
        self.vm1.save()
287
        r = self.client.post(reverse('helpdesk-vm-shutdown', args=(pk,)),
288
                             data={'token': '0001'}, user_token='0001')
289
        self.assertEqual(r.status_code, 302)
290
        self.assertTrue(shutdown.called)
291
        self.assertEqual(len(shutdown.mock_calls), 1)
292

    
293
        backend.startup_instance = startup = mock.Mock()
294
        startup.return_value = 2
295
        self.vm1.operstate = 'STOPPED'
296
        self.vm1.save()
297
        r = self.client.post(reverse('helpdesk-vm-start', args=(pk,)),
298
                             data={'token': '0001'}, user_token='0001')
299
        self.assertEqual(r.status_code, 302)
300
        self.assertTrue(startup.called)
301
        self.assertEqual(len(startup.mock_calls), 1)