user.py 12.3 KB
Newer Older
1
import attr
2
import logging
3
import requests
4
5
import time

6
from gitlab import Gitlab  # python-gitlab
7
8
9
10
11
12
13
from .selenium import (
    webdriver_wait_get,
)
from .session import (
    login_as_root,
    sign_in_page_login,
)
14

15
logger = logging.getLogger(__name__)
16
17
18
19
20
21
22
23
24
25


def default_password(username):
    return username + '.' * (8 - len(username))


@attr.s
class User(object):
    name = attr.ib()
    id = attr.ib()
Georges Racinet's avatar
Georges Racinet committed
26
    heptapod = attr.ib(repr=False)
Georges Racinet's avatar
Georges Racinet committed
27
    email = attr.ib(default=None, repr=False)
Georges Racinet's avatar
Georges Racinet committed
28
    password = attr.ib(default=None, repr=False)
29
    token = attr.ib(default=None, repr=False)
30
    _webdriver = attr.ib(default=None, repr=False)
31
    _python_gitlab_client = attr.ib(default=None, repr=False)
32
33

    @classmethod
Georges Racinet's avatar
Georges Racinet committed
34
35
36
37
    def create(cls, heptapod, name,
               password=None,
               email=None,
               fullname=None, **kw):
38
39
40
        if password is None:
            # requirements on password length are also enforced in the API
            password = default_password(name)
41
42
        if fullname is None:
            fullname = name
Georges Racinet's avatar
Georges Racinet committed
43
44
        if email is None:
            email = name + '@heptapod.example.net',
45
46
47

        resp = requests.post(
            '/'.join((heptapod.api_url, 'users')),
48
            data=dict(name=fullname,
49
50
51
                      username=name,
                      password=password,
                      reset_password=False,
Georges Racinet's avatar
Georges Racinet committed
52
                      email=email,
53
                      skip_confirmation=True,
Georges Racinet's avatar
Georges Racinet committed
54
                      **kw
55
56
57
58
                      ),
            headers=heptapod.root_token_headers,
        )
        assert resp.status_code < 300, "User %r creation failed" % name
59
        user = cls(name=name, heptapod=heptapod, password=password,
Georges Racinet's avatar
Georges Racinet committed
60
                   email=email,
61
                   id=resp.json()['id'])
62
63
        user.store_in_heptapod()
        return user
64
65

    @classmethod
66
67
    def init_root(cls, heptapod, password):
        """Singled out because we have to use hardcoded values for root."""
68
69
70
        user = cls(id=1, heptapod=heptapod, name='root', password=password)
        user.store_in_heptapod()
        return user
71

72
73
74
75
76
77
78
79
80
81
82
    @classmethod
    def ensure(cls, heptapod, name, password=None, fullname=None):
        """Retrieve user or create it.

        :returns: `User` instance
        """
        if password is None:
            password = default_password(password)

        user = heptapod.users.get(name)
        if user is None:
83
            user = cls.search(heptapod, name)
84
            if user is None:
85
86
                user = cls.create(heptapod, name,
                                  password=password, fullname=fullname)
87
88
89
90
91
92
            user.store_in_heptapod()

        user.password = password
        user.ensure_private_token()
        return user

93
    @classmethod
94
    def search(cls, heptapod, name):
95
96
97
98
99
100
        """Lookup user in API and returns instance.

        The returned User instance doesn't know its password. Either it
        has an expected value, or it won't be suitable for webdriver
        authentication (API auth is usually done with access tokens).
        """
101
102
103
104
105
106
107
108
109
110
        resp = requests.get(
            '/'.join((heptapod.api_url, 'users')),
            params=dict(username=name),
            headers=heptapod.root_token_headers,
        )
        found = resp.json()
        if resp.status_code == 404 or not found:
            return

        assert len(found) == 1
Georges Racinet's avatar
Georges Racinet committed
111
112
113
114
115
        details = found[0]
        return cls(name=name,
                   id=details['id'],
                   email=details['email'],
                   heptapod=heptapod)
116
117

    def store_in_heptapod(self):
118
        self.heptapod.users[self.name] = self
119

120
121
122
    @property
    def webdriver(self):
        """Selenium webdriver, authentified as self.
123
        """
124
125
126
127
128
129
130
131
        driver = self._webdriver
        if driver is not None:
            return driver

        logger.info("Initializing a signed-in webdriver for user %r",
                    self.name)

        # guaranteeing driver to be available for teardown as soon as created
132
133
134
135
136
137
138
139
140
141
142
143
144
        driver = self._webdriver = self.heptapod.new_webdriver()

        heptapod = self.heptapod
        if self.name == 'root':
            login_as_root(driver, heptapod, self.password)
        else:
            start = time.time()
            webdriver_wait_get(heptapod, driver, relative_uri='/users/sign_in')
            sign_in_page_login(driver, heptapod, self.name,
                               password=self.password)
            logger.info("Signed in user %s in %.2f seconds",
                        self.name, time.time() - start)
        return driver
145

146
147
148
149
150
151
152
153
154
155
    def close_webdriver(self):
        """Close current web driver if there is one."""
        driver = self._webdriver
        if driver is None:
            return

        logger.info("Closing webdriver for user %r", self.name)
        driver.close()
        self._webdriver = None

Georges Racinet's avatar
Georges Racinet committed
156
157
158
159
160
161
162
163
164
165
166
167
168
    def api_request(self, method, subpath=None, **data):
        """Perform a request on user API as root"""
        hepta = self.heptapod
        segments = [hepta.api_url, 'users', str(self.id)]
        if subpath is not None:
            segments.append(subpath)

        return requests.request(method, '/'.join(segments),
                                headers=hepta.root_token_headers)

    def api_post(self, subpath=None, **data):
        return self.api_request('POST', subpath=subpath, **data)

169
    def delete(self, hard=True):
170
171
        self.close_webdriver()
        # TODO also remove from Heptapod instance
172
173
174
175
176
177
178
179
        hepta = self.heptapod
        resp = requests.delete(
            '/'.join((hepta.api_url, 'users', str(self.id))),
            data=dict(id=self.id, hard_delete=hard),
            headers=hepta.root_token_headers,
        )

        assert resp.status_code == 204, "Failed to delete %r" % self
180

Georges Racinet's avatar
Georges Racinet committed
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
    def edit(self, **data):
        resp = self.api_request('PUT', **data)
        assert resp.status_code == 200, "Failed to edit %r" % self

    def block(self):
        resp = self.api_post(subpath='block')
        assert resp.status_code == 201, "Failed to block %r" % self

    def unblock(self):
        resp = self.api_post(subpath='unblock')
        assert resp.status_code == 201, "Failed to unblock %r" % self

    def deactivate(self):
        resp = self.api_post(subpath='deactivate')
        assert resp.status_code == 201, "Failed to deactivate %r" % self

    def activate(self):
        resp = self.api_post(subpath='activate')
        assert resp.status_code == 201, "Failed to activate %r" % self

201
    def ensure_ssh_pub_key(self, pubkey, title='heptapod-tests'):
202
203
204
205
206
        """Upload the SSH public key if needed.

        Does not require any special privilege if a token for the user
        is already known
        """
207
        hepta = self.heptapod
208
209
210
211
212
213
214
215
216
        if self.token:
            keys_api_url = '/'.join((
                hepta.api_url, 'user', 'keys'))
            headers = self.token_headers()
        else:
            keys_api_url = '/'.join((
                hepta.api_url, 'users', str(self.id), 'keys'))
            headers = hepta.root_token_headers

217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
        ls_resp = requests.get(keys_api_url, headers=headers)
        assert ls_resp.status_code == 200

        # check existence based on title. This is quite lame (we should
        # check fingerprint), but that's good enough for these first tests.
        for key_info in ls_resp.json():
            if key_info['title'] == title:
                return

        create_resp = requests.post(keys_api_url,
                                    data=dict(id=self.id,
                                              title=title,
                                              key=pubkey),
                                    headers=headers,
                                    )
        assert create_resp.status_code == 201
        assert create_resp.json()['title'] == title
234

235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
    def delete_ssh_keys(self):
        """Delete all SSH keys of current user.

        Does not require any special privilege.
        """
        hepta = self.heptapod

        if self.token:
            keys_api_url = '/'.join((
                hepta.api_url, 'user', 'keys'))
            headers = self.token_headers()
        else:
            keys_api_url = '/'.join((
                hepta.api_url, 'users', str(self.id), 'keys'))
            headers = hepta.root_token_headers

        ls_resp = requests.get(keys_api_url, headers=headers)
        assert ls_resp.status_code == 200
        # check existence based on title. This is quite lame (we should
        # check fingerprint), but that's good enough for these first tests.
        for key_info in ls_resp.json():
            resp = requests.delete('%s/%d' % (keys_api_url, key_info['id']),
                                   headers=headers)
            assert resp.status_code == 204

260
261
262
263
264
265
266
267
268
269
270
271
272
    @property
    def ssh_command(self):
        """Return a SSH command to perform operations as the given user.
        """
        # IdentitiesOnly makes sure no other SSH key than the one
        # specified with -i are going to be used. Otherwise, current user
        # SSH keys could be attempted before the correct one, leading
        # to either too much auth failures, or use of the wrong key if
        # one happens to be also known by Heptapod
        return ' '.join(('ssh',
                         '-o', 'IdentitiesOnly=yes',
                         '-o', 'NoHostAuthenticationForLocalhost=yes ',
                         '-o', 'StrictHostKeyChecking=no',
273
                         '-i', self.ssh['priv']))
274

Georges Racinet's avatar
Georges Racinet committed
275
276
277
278
279
    def token_headers(self):
        """Return a *new* headers dict suitable for authenticated API acces.
        """
        return {'Private-Token': self.token}

280
281
282
283
284
285
286
287
288
    @property
    def basic_auth_url(self):
        url = self.heptapod.parsed_url
        return "{scheme}://{auth}@{netloc}".format(
            scheme=url.scheme,
            netloc=url.netloc,
            auth=':'.join((self.name, self.password)),
        )

Georges Racinet's avatar
Georges Racinet committed
289
290
291
292
293
294
295
296
297
298
    def api_get(self):
        """Perform GET on users API for self, using :attr:`token`.

        :return: response or `None` if :attr:`token` is None
        """
        token = self.token
        if token is None:
            return None

        return requests.get(self.heptapod.api_url + '/users/%d' % self.id,
299
300
301
302
303
304
305
306
307
308
309
310
311
312
                            headers=self.token_headers())

    def api_owned_groups(self):
        """List the groups owned by this user.

        TODO it lists actually the first page, which is expected to
        be good enough for the foreseeable future.
        """
        resp = requests.get(self.heptapod.api_url + '/groups',
                            params=dict(owned=True),
                            headers=self.token_headers(),
                            )
        assert resp.status_code < 400
        return resp.json()
Georges Racinet's avatar
Georges Racinet committed
313

314
315
316
317
318
    def ensure_private_token(self):
        """Generate a private token for already logged in user if needed.

        The generation involves obtaining a signed-in webdriver for the user.
        It is therefore costly.
319
        """
320
321
322
323
324
        resp = self.api_get()
        if resp is not None and resp.status_code == 200:
            logger.info("%r: reusing existing access token.", self)
            return

325
326
        heptapod = self.heptapod
        start = time.time()
327
        webdriver = self.webdriver
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
        webdriver.get(heptapod.url + '/profile/personal_access_tokens')
        name_elt = webdriver.find_element_by_xpath(
            '//input[@name="personal_access_token[name]"]')
        name_elt.send_keys("heptapod-tests")

        for scope in webdriver.find_elements_by_xpath(
                '//input[@type="checkbox"]'):
            scope.click()
        submit = webdriver.find_element_by_xpath('//input[@type="submit"]')
        submit.click()

        elem = webdriver.find_element_by_id('created-personal-access-token')
        token = elem.get_attribute('value')
        self.token = token

        logger.info("%r: access token generated in %.2f seconds",
                    self, time.time() - start)
345

346
347
348
349
350
351
352
353
354
355
356
357
    @property
    def python_gitlab_client(self):
        """Return the API client from the python-gitlab library

        See https://python-gitlab.readthedocs.io
        """
        cl = self._python_gitlab_client
        if cl is None:
            cl = self._python_gitlab_client = Gitlab(self.heptapod.url,
                                                     private_token=self.token)
        return cl

358
359
360
361
362
363
    def __enter__(self):
        return self

    def __exit__(self, *exc_args):
        self.delete()
        return False