project.py 11.8 KB
Newer Older
1 2 3 4
# -*- coding: utf-8 -*-

import base
import swagger_client
5
from swagger_client.rest import ApiException
6

7 8 9 10 11 12 13 14 15 16 17
def is_member_exist_in_project(members, member_user_name, expected_member_role_id = None):
    result = False
    for member in members:
        if member.entity_name == member_user_name:
            if expected_member_role_id != None:
                if member.role_id == expected_member_role_id:
                    return True
            else:
                return True
    return result

18 19 20 21 22 23
def get_member_id_by_name(members, member_user_name):
    for member in members:
        if member.entity_name == member_user_name:
            return member.id
    return None

24
class Project(base.Base):
25
    def create_project(self, name=None, metadata=None, expect_status_code = 201, expect_response_body = None, **kwargs):
Yan's avatar
Yan committed
26 27
        if name is None:
            name = base._random_name("project")
danfengliu's avatar
danfengliu committed
28 29
        if metadata is None:
            metadata = {}
30
        client = self._get_client(**kwargs)
31 32 33 34

        try:
            _, status_code, header = client.projects_post_with_http_info(swagger_client.ProjectReq(name, metadata))
        except ApiException as e:
35 36 37 38
            base._assert_status_code(expect_status_code, e.status)
            if expect_response_body is not None:
                base._assert_status_body(expect_response_body, e.body)
            return
39
        base._assert_status_code(expect_status_code, status_code)
40
        base._assert_status_code(201, status_code)
41
        return base._get_id_from_header(header), name
danfengliu's avatar
danfengliu committed
42 43 44 45 46 47 48 49

    def get_projects(self, params, **kwargs):
        client = self._get_client(**kwargs)
        data = []
        data, status_code, _ = client.projects_get_with_http_info(**params)
        base._assert_status_code(200, status_code)
        return data

50 51 52 53 54 55 56 57
    def projects_should_exist(self, params, expected_count = None, expected_project_id = None, **kwargs):
        project_data = self.get_projects(params, **kwargs)
        actual_count = len(project_data)
        if expected_count is not None and actual_count!= expected_count:
            raise Exception(r"Private project count should be {}.".format(expected_count))
        if expected_project_id is not None and actual_count == 1 and str(project_data[0].project_id) != str(expected_project_id):
            raise Exception(r"Project-id check failed, expect {} but got {}, please check this test case.".format(str(expected_project_id), str(project_data[0].project_id)))

danfengliu's avatar
danfengliu committed
58 59
    def check_project_name_exist(self, name=None, **kwargs):
        client = self._get_client(**kwargs)
60 61 62 63
        try:
            _, status_code, _ = client.projects_head_with_http_info(name)
        except ApiException as e:
            status_code = -1
danfengliu's avatar
danfengliu committed
64 65 66
        return {
            200: True,
            404: False,
67
        }.get(status_code,False)
danfengliu's avatar
danfengliu committed
68

69
    def get_project(self, project_id, expect_status_code = 200, expect_response_body = None, **kwargs):
danfengliu's avatar
danfengliu committed
70
        client = self._get_client(**kwargs)
71 72 73
        try:
            data, status_code, _ = client.projects_project_id_get_with_http_info(project_id)
        except ApiException as e:
74 75 76 77 78
            base._assert_status_code(expect_status_code, e.status)
            if expect_response_body is not None:
                base._assert_status_body(expect_response_body, e.body)
            return

79
        base._assert_status_code(expect_status_code, status_code)
80
        base._assert_status_code(200, status_code)
danfengliu's avatar
danfengliu committed
81 82
        return data

83
    def update_project(self, project_id, expect_status_code=200, metadata=None, cve_allowlist=None, **kwargs):
danfengliu's avatar
danfengliu committed
84
        client = self._get_client(**kwargs)
85
        project = swagger_client.ProjectReq(metadata=metadata, cve_allowlist=cve_allowlist)
86 87 88 89 90 91
        try:
            _, sc, _ = client.projects_project_id_put_with_http_info(project_id, project)
        except ApiException as e:
            base._assert_status_code(expect_status_code, e.status)
        else:
            base._assert_status_code(expect_status_code, sc)
danfengliu's avatar
danfengliu committed
92

93
    def delete_project(self, project_id, expect_status_code = 200, **kwargs):
danfengliu's avatar
danfengliu committed
94 95
        client = self._get_client(**kwargs)
        _, status_code, _ = client.projects_project_id_delete_with_http_info(project_id)
96
        base._assert_status_code(expect_status_code, status_code)
danfengliu's avatar
danfengliu committed
97

98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114
    def get_project_log(self, project_id, expect_status_code = 200, **kwargs):
        client = self._get_client(**kwargs)
        body, status_code, _ = client.projects_project_id_logs_get_with_http_info(project_id)
        base._assert_status_code(expect_status_code, status_code)
        return body

    def filter_project_logs(self, project_id, operator, repository, tag, operation_type, **kwargs):
        access_logs = self.get_project_log(project_id, **kwargs)
        count = 0
        for each_access_log in list(access_logs):
            if each_access_log.username == operator and \
               each_access_log.repo_name.strip(r'/') == repository and \
               each_access_log.repo_tag == tag and \
               each_access_log.operation == operation_type:
                count = count + 1
        return count

danfengliu's avatar
danfengliu committed
115
    def get_project_members(self, project_id, **kwargs):
116 117 118
        client = self._get_client(**kwargs)
        return client.projects_project_id_members_get(project_id)

119
    def get_project_member(self, project_id, member_id, expect_status_code = 200, expect_response_body = None, **kwargs):
danfengliu's avatar
danfengliu committed
120 121
        client = self._get_client(**kwargs)
        data = []
122 123 124
        try:
            data, status_code, _ = client.projects_project_id_members_mid_get_with_http_info(project_id, member_id,)
        except ApiException as e:
125 126 127 128 129
            base._assert_status_code(expect_status_code, e.status)
            if expect_response_body is not None:
                base._assert_status_body(expect_response_body, e.body)
            return

130
        base._assert_status_code(expect_status_code, status_code)
131
        base._assert_status_code(200, status_code)
132 133
        return data

134 135 136 137 138 139 140 141
    def get_project_member_id(self, project_id, member_user_name, **kwargs):
        members = self.get_project_members(project_id, **kwargs)
        result = get_member_id_by_name(list(members), member_user_name)
        if result == None:
            raise Exception(r"Failed to get member id of member {} in project {}.".format(member_user_name, project_id))
        else:
            return result

142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158
    def check_project_member_not_exist(self, project_id, member_user_name, **kwargs):
        members = self.get_project_members(project_id, **kwargs)
        result = is_member_exist_in_project(list(members), member_user_name)
        if result == True:
            raise Exception(r"User {} should not be a member of project with ID {}.".format(member_user_name, project_id))

    def check_project_members_exist(self, project_id, member_user_name, expected_member_role_id = None, **kwargs):
        members = self.get_project_members(project_id, **kwargs)
        result = is_member_exist_in_project(members, member_user_name, expected_member_role_id = expected_member_role_id)
        if result == False:
            raise Exception(r"User {} should be a member of project with ID {}.".format(member_user_name, project_id))

    def update_project_member_role(self, project_id, member_id, member_role_id, expect_status_code = 200, **kwargs):
        client = self._get_client(**kwargs)
        role = swagger_client.Role(role_id = member_role_id)
        data, status_code, _ = client.projects_project_id_members_mid_put_with_http_info(project_id, member_id, role = role)
        base._assert_status_code(expect_status_code, status_code)
159
        base._assert_status_code(200, status_code)
danfengliu's avatar
danfengliu committed
160 161
        return data

162 163 164 165
    def delete_project_member(self, project_id, member_id, expect_status_code = 200, **kwargs):
        client = self._get_client(**kwargs)
        _, status_code, _ = client.projects_project_id_members_mid_delete_with_http_info(project_id, member_id)
        base._assert_status_code(expect_status_code, status_code)
166
        base._assert_status_code(200, status_code)
167

168
    def add_project_members(self, project_id, user_id, member_role_id = None, expect_status_code = 201, **kwargs):
danfengliu's avatar
danfengliu committed
169 170 171 172 173 174
        if member_role_id is None:
            member_role_id = 1
        _member_user = {"user_id": int(user_id)}
        projectMember = swagger_client.ProjectMember(member_role_id, member_user = _member_user)
        client = self._get_client(**kwargs)
        data = []
175
        data, status_code, header = client.projects_project_id_members_post_with_http_info(project_id, project_member = projectMember)
176
        base._assert_status_code(expect_status_code, status_code)
177 178
        return base._get_id_from_header(header)

danfengliu's avatar
danfengliu committed
179
    def add_project_robot_account(self, project_id, project_name, expires_at, robot_name = None, robot_desc = None, has_pull_right = True,  has_push_right = True, has_chart_read_right = True,  has_chart_create_right = True, expect_status_code = 201, **kwargs):
180 181 182 183 184 185 186 187
        if robot_name is None:
            robot_name = base._random_name("robot")
        if robot_desc is None:
            robot_desc = base._random_name("robot_desc")
        if has_pull_right is False and has_push_right is False:
            has_pull_right = True
        access_list = []
        resource_by_project_id = "/project/"+str(project_id)+"/repository"
danfengliu's avatar
danfengliu committed
188
        resource_helm_by_project_id = "/project/"+str(project_id)+"/helm-chart"
189
        resource_helm_create_by_project_id = "/project/"+str(project_id)+"/helm-chart-version"
190 191
        action_pull = "pull"
        action_push = "push"
danfengliu's avatar
danfengliu committed
192 193
        action_read = "read"
        action_create = "create"
194 195 196 197 198 199
        if has_pull_right is True:
            robotAccountAccess = swagger_client.RobotAccountAccess(resource = resource_by_project_id, action = action_pull)
            access_list.append(robotAccountAccess)
        if has_push_right is True:
            robotAccountAccess = swagger_client.RobotAccountAccess(resource = resource_by_project_id, action = action_push)
            access_list.append(robotAccountAccess)
danfengliu's avatar
danfengliu committed
200 201 202 203
        if has_chart_read_right is True:
            robotAccountAccess = swagger_client.RobotAccountAccess(resource = resource_helm_by_project_id, action = action_read)
            access_list.append(robotAccountAccess)
        if has_chart_create_right is True:
204
            robotAccountAccess = swagger_client.RobotAccountAccess(resource = resource_helm_create_by_project_id, action = action_create)
danfengliu's avatar
danfengliu committed
205 206
            access_list.append(robotAccountAccess)

207
        robotAccountCreate = swagger_client.RobotAccountCreate(robot_name, robot_desc, expires_at, access_list)
208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230
        client = self._get_client(**kwargs)
        data = []
        data, status_code, header = client.projects_project_id_robots_post_with_http_info(project_id, robotAccountCreate)
        base._assert_status_code(expect_status_code, status_code)
        base._assert_status_code(201, status_code)
        return base._get_id_from_header(header), data

    def get_project_robot_account_by_id(self, project_id, robot_id, **kwargs):
        client = self._get_client(**kwargs)
        data, status_code, _ = client.projects_project_id_robots_robot_id_get_with_http_info(project_id, robot_id)
        return data

    def disable_project_robot_account(self, project_id, robot_id, disable,  expect_status_code = 200, **kwargs):
        client = self._get_client(**kwargs)
        robotAccountUpdate = swagger_client.RobotAccountUpdate(disable)
        _, status_code, _ = client.projects_project_id_robots_robot_id_put_with_http_info(project_id, robot_id, robotAccountUpdate)
        base._assert_status_code(expect_status_code, status_code)
        base._assert_status_code(200, status_code)

    def delete_project_robot_account(self, project_id, robot_id, expect_status_code = 200, **kwargs):
        client = self._get_client(**kwargs)
        _, status_code, _ = client.projects_project_id_robots_robot_id_delete_with_http_info(project_id, robot_id)
        base._assert_status_code(expect_status_code, status_code)
231
        base._assert_status_code(200, status_code)