Read about our upcoming Code of Conduct on this issue

group.py 3.92 KB
Newer Older
Georges Racinet's avatar
Georges Racinet committed
1
2
import attr
import requests
3
4
5
6
7
8
9
10
11
12
13
14
15


@attr.s
class UserNameSpace:
    user_name = attr.ib()
    heptapod = attr.ib()

    @property
    def full_path(self):
        return self.user_name

    def __eq__(self, other):
        return other.__class__ == self.__class__ and self.user == other.user
Georges Racinet's avatar
Georges Racinet committed
16
17
18
19
20
21
22
23
24
25
26
27
28
29


@attr.s
class Group:

    id = attr.ib()
    path = attr.ib()
    full_path = attr.ib()
    heptapod = attr.ib()
    owner_name = attr.ib(default=None)

    api_uri = '/api/v4/groups'

    @classmethod
Georges Racinet's avatar
Georges Racinet committed
30
31
32
33
34
35
    def api_create(cls, heptapod, group_path, user_name='root', parent=None):
        data = dict(name=group_path,
                    path=group_path)
        if parent is not None:
            data['parent_id'] = parent.id

36
        headers = {'Private-Token': heptapod.users[user_name].token}
Georges Racinet's avatar
Georges Racinet committed
37
38
        resp = requests.post(heptapod.url + cls.api_uri,
                             headers=headers,
Georges Racinet's avatar
Georges Racinet committed
39
                             data=data)
Georges Racinet's avatar
Georges Racinet committed
40
41
42
43
44
45
46
47
        assert resp.status_code == 201
        res = resp.json()
        return cls.api_retrieve(heptapod, group_id=res['id'])

    def __eq__(self, other):
        return self.id == other.id

    @classmethod
48
49
50
51
52
53
54
    def api_retrieve(cls, heptapod, group_id, owner_name=None):
        """Return a checked Group object for the given id.

        :owner_name: if specified, registered as :attr:`owner_name` on the
           returned object, and used for all API calls, including the check
           performed by this method.
        """
Georges Racinet's avatar
Georges Racinet committed
55
56
57
        grp = Group(heptapod=heptapod,
                    id=group_id,
                    full_path=None,
58
                    owner_name=owner_name,
Georges Racinet's avatar
Georges Racinet committed
59
60
61
62
63
64
65
66
                    path=None)
        resp = grp.api_get()
        assert resp.status_code == 200
        info = resp.json()
        grp.path = info['path']
        grp.full_path = info['full_path']
        return grp

67
68
69
    def private_token(self):
        """Return a token strong enough for all operations.

70
        Namely, using a token for an owner, if there's a known one or
71
72
        an Administrator token
        """
Georges Racinet's avatar
Georges Racinet committed
73
        user_name = self.owner_name if self.owner_name is not None else 'root'
74
75
76
77
        return {'Private-Token': self.heptapod.users[user_name].token}

    def api_get(self):
        return requests.get(self.api_url, headers=self.private_token())
Georges Racinet's avatar
Georges Racinet committed
78

79
    def api_post(self, subpath='', **params):
80
        return requests.post('/'.join((self.api_url, subpath)),
81
                             headers=self.private_token(),
82
83
84
85
86
87
88
                             data=params)

    def api_add_member(self, user, access_level):
        return self.api_post(subpath='members',
                             user_id=user.id,
                             access_level=access_level)

Georges Racinet's avatar
Georges Racinet committed
89
90
    @classmethod
    def api_search(cls, heptapod, group_name, user_name='root'):
91
        headers = {'Private-Token': heptapod.users[user_name].token}
Georges Racinet's avatar
Georges Racinet committed
92
93
94
95
96
97
98
99
100
101
102
103
104
105
        resp = requests.get(heptapod.url + cls.api_uri,
                            headers=headers,
                            params=dict(search=group_name))
        assert resp.status_code == 200
        data = resp.json()
        assert len(data) == 1
        group = data[0]
        return cls.api_retrieve(heptapod, group_id=group['id'])

    @property
    def api_url(self):
        return self.heptapod.url + self.api_uri + '/%d' % self.id

    def api_delete(self):
106
        resp = requests.delete(self.api_url, headers=self.private_token())
107
        assert resp.status_code in (202, 204)
108

109
110
111
    def fs_path(self):
        return '/'.join((self.heptapod.repositories_root, self.full_path))

112
    def put_hgrc(self, lines):
113
114
115
116
        """Replace group's server-side HGRC with given lines.

        The lines have to include LF, same as with `writelines()`.
        """
117
118
119
        fs_path = self.fs_path()
        self.heptapod.run_shell(('mkdir', '-p', fs_path), user='git')
        self.heptapod.put_file_lines(fs_path + '/hgrc', lines)
120
121
122
123
124
125
126

    def __enter__(self):
        return self

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