Statistics
| Branch: | Tag: | Revision:

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

History | View | Annotate | Download (11.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
    fixtures = ['helpdesk_test']
102

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

    
108
    def test_enabled_setting(self):
109
        settings.HELPDESK_ENABLED = False
110

    
111
        # helpdesk is disabled
112
        r = self.client.get(reverse('helpdesk-index'), user_token="0001")
113
        self.assertEqual(r.status_code, 404)
114
        r = self.client.get(reverse('helpdesk-details',
115
                                    args=['testuser@test.com']),
116
                            user_token="0001")
117
        self.assertEqual(r.status_code, 404)
118

    
119
    def test_ip_lookup(self):
120
        # ip does not exist, proper message gets displayed
121
        r = self.client.get(reverse('helpdesk-details',
122
                            args=["195.251.221.122"]), user_token='0001')
123
        self.assertContains(r, 'User with IP')
124

    
125
        # ip exists, 'test' account discovered
126
        r = self.client.get(reverse('helpdesk-details',
127
                            args=["195.251.222.211"]), user_token='0001')
128
        self.assertEqual(r.context['account'], USER2)
129

    
130
    def test_vm_lookup(self):
131
        # vm id does not exist
132
        r = self.client.get(reverse('helpdesk-details',
133
                            args=["vm-123"]), user_token='0001')
134
        self.assertContains(r, 'User with Virtual Machine')
135

    
136
        # vm exists, 'test' account discovered
137
        r = self.client.get(reverse('helpdesk-details',
138
                            args=["vm1001"]), user_token='0001')
139
        self.assertEqual(r.context['account'], USER1)
140
        r = self.client.get(reverse('helpdesk-details',
141
                            args=["vm1002"]), user_token='0001')
142
        self.assertEqual(r.context['account'], USER2)
143
        # dash also works
144
        r = self.client.get(reverse('helpdesk-details',
145
                            args=["vm-1002"]), user_token='0001')
146
        self.assertEqual(r.context['account'], USER2)
147

    
148
    def test_view_permissions(self):
149
        # anonymous user gets 403
150
        r = self.client.get(reverse('helpdesk-index'), user_token=None)
151
        self.assertEqual(r.status_code, 403)
152
        r = self.client.get(reverse('helpdesk-details',
153
                                    args=['testuser@test.com']),
154
                            user_token=None)
155
        self.assertEqual(r.status_code, 403)
156

    
157
        # user not in helpdesk group gets 403
158
        r = self.client.get(reverse('helpdesk-index'))
159
        self.assertEqual(r.status_code, 403)
160
        r = self.client.get(reverse('helpdesk-details',
161
                                    args=['testuser@test.com']))
162
        self.assertEqual(r.status_code, 403)
163

    
164
        # user with helpdesk group gets 200
165
        r = self.client.get(reverse('helpdesk-index'), user_token="0001")
166
        self.assertEqual(r.status_code, 200)
167
        r = self.client.get(reverse('helpdesk-details',
168
                                    args=['testuser@test.com']),
169
                            user_token="0001")
170
        self.assertEqual(r.status_code, 200)
171

    
172
        r = self.client.post(reverse('helpdesk-suspend-vm', args=(1001,)))
173
        r = self.client.get(reverse('helpdesk-suspend-vm', args=(1001,)),
174
                            user_token="0001", data={'token': '1234'})
175
        self.assertEqual(r.status_code, 403)
176
        r = self.client.get(reverse('helpdesk-suspend-vm', args=(1001,)),
177
                            user_token="0001")
178
        self.assertEqual(r.status_code, 403)
179
        r = self.client.post(reverse('helpdesk-suspend-vm', args=(1001,)),
180
                             user_token="0001", data={'token': '0001'})
181
        self.assertEqual(r.status_code, 302)
182
        r = self.client.post(reverse('helpdesk-suspend-vm', args=(1001,)),
183
                             user_token="0000", data={'token': '0000'})
184
        self.assertEqual(r.status_code, 403)
185

    
186
    def test_suspend_vm(self):
187
        r = self.client.get(reverse('helpdesk-details',
188
                                    args=['testuser@test.com']),
189
                            user_token="0001")
190
        self.assertEqual(r.status_code, 200)
191
        vmid = r.context['vms'][0].pk
192
        r = self.client.post(reverse('helpdesk-suspend-vm', args=(vmid,)),
193
                             data={'token': '0001'}, user_token="0001")
194
        self.assertEqual(r.status_code, 302)
195

    
196
        r = self.client.get(reverse('helpdesk-details',
197
                                    args=['testuser@test.com']),
198
                            user_token="0001")
199
        self.assertTrue(r.context['vms'][0].suspended)
200

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

    
210
    def test_results_get_filtered(self):
211
        """
212
        Test that view context data are filtered based on userid provided.
213
        Check helpdesk_test.json to see the existing database data.
214
        """
215

    
216
        # 'testuser@test.com' details, see
217
        # helpdes/fixtures/helpdesk_test.json for more details
218
        r = self.client.get(reverse('helpdesk-details',
219
                                    args=['testuser@test.com']),
220
                            user_token="0001")
221
        account = r.context['account']
222
        vms = r.context['vms']
223
        nets = r.context['networks']
224
        self.assertEqual(account, USER1)
225
        self.assertEqual(vms[0].name, "user1 vm")
226
        self.assertEqual(vms.count(), 1)
227
        self.assertEqual(len(nets), 1)
228
        self.assertEqual(r.context['account_exists'], True)
229

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

    
245
        # 'testuser5@test.com' does not exist, should be redirected to
246
        # helpdesk home
247
        r = self.client.get(reverse('helpdesk-details',
248
                                    args=['testuser5@test.com']),
249
                            user_token="0001")
250
        vms = r.context['vms']
251
        self.assertEqual(r.context['account_exists'], False)
252
        self.assertEqual(vms.count(), 0)
253

    
254
    def test_start_shutdown(self):
255
        from synnefo.logic import backend
256

    
257
        self.vm1 = mfactory.VirtualMachineFactory(userid=USER1)
258
        pk = self.vm1.pk
259

    
260
        r = self.client.post(reverse('helpdesk-vm-shutdown', args=(pk,)))
261
        self.assertEqual(r.status_code, 403)
262

    
263
        r = self.client.post(reverse('helpdesk-vm-shutdown', args=(pk,)),
264
                             data={'token': '0001'})
265
        self.assertEqual(r.status_code, 403)
266

    
267
        backend.shutdown_instance = shutdown = mock.Mock()
268
        self.vm1.operstate = 'STARTED'
269
        self.vm1.save()
270
        r = self.client.post(reverse('helpdesk-vm-shutdown', args=(pk,)),
271
                             data={'token': '0001'}, user_token='0001')
272
        self.assertEqual(r.status_code, 302)
273
        self.assertTrue(shutdown.called)
274
        self.assertEqual(len(shutdown.mock_calls), 1)
275

    
276
        backend.startup_instance = startup = mock.Mock()
277
        self.vm1.operstate = 'STOPPED'
278
        self.vm1.save()
279
        r = self.client.post(reverse('helpdesk-vm-start', args=(pk,)),
280
                             data={'token': '0001'}, user_token='0001')
281
        self.assertEqual(r.status_code, 302)
282
        self.assertTrue(startup.called)
283
        self.assertEqual(len(startup.mock_calls), 1)