project.py 12 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 60 61 62 63 64 65
    def check_project_name_exist(self, name=None, **kwargs):
        client = self._get_client(**kwargs)
        _, status_code, _ = client.projects_head_with_http_info(name)
        return {
            200: True,
            404: False,
        }.get(status_code,'error')

66
    def get_project(self, project_id, expect_status_code = 200, expect_response_body = None, **kwargs):
danfengliu's avatar
danfengliu committed
67
        client = self._get_client(**kwargs)
68 69 70
        try:
            data, status_code, _ = client.projects_project_id_get_with_http_info(project_id)
        except ApiException as e:
71 72 73 74 75
            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

76
        base._assert_status_code(expect_status_code, status_code)
77
        base._assert_status_code(200, status_code)
danfengliu's avatar
danfengliu committed
78 79 80 81 82 83 84 85
        return data

    def update_project(self, project_id, metadata, **kwargs):
        client = self._get_client(**kwargs)
        project = swagger_client.Project(project_id, None, None, None, None, None, None, None, None, None, None, metadata)
        _, status_code, _ = client.projects_project_id_put_with_http_info(project_id, project)
        base._assert_status_code(200, status_code)

86
    def delete_project(self, project_id, expect_status_code = 200, **kwargs):
danfengliu's avatar
danfengliu committed
87 88
        client = self._get_client(**kwargs)
        _, status_code, _ = client.projects_project_id_delete_with_http_info(project_id)
89
        base._assert_status_code(expect_status_code, status_code)
danfengliu's avatar
danfengliu committed
90

91
    def get_project_metadata_by_name(self, project_id, meta_name, expect_status_code = 200, **kwargs):
danfengliu's avatar
danfengliu committed
92 93 94
        client = self._get_client(**kwargs)
        ProjectMetadata = swagger_client.ProjectMetadata()
        ProjectMetadata, status_code, _ = client.projects_project_id_metadatas_meta_name_get_with_http_info(project_id, meta_name)
95
        base._assert_status_code(expect_status_code, status_code)
danfengliu's avatar
danfengliu committed
96 97 98 99 100 101 102 103
        return {
            'public': ProjectMetadata.public,
            'enable_content_trust': ProjectMetadata.enable_content_trust,
            'prevent_vul': ProjectMetadata.prevent_vul,
            'auto_scan': ProjectMetadata.auto_scan,
            'severity': ProjectMetadata.severity,
        }.get(meta_name,'error')

104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120
    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
121
    def get_project_members(self, project_id, **kwargs):
122 123 124
        client = self._get_client(**kwargs)
        return client.projects_project_id_members_get(project_id)

125
    def get_project_member(self, project_id, member_id, expect_status_code = 200, expect_response_body = None, **kwargs):
danfengliu's avatar
danfengliu committed
126 127
        client = self._get_client(**kwargs)
        data = []
128 129 130
        try:
            data, status_code, _ = client.projects_project_id_members_mid_get_with_http_info(project_id, member_id,)
        except ApiException as e:
131 132 133 134 135
            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

136
        base._assert_status_code(expect_status_code, status_code)
137
        base._assert_status_code(200, status_code)
138 139
        return data

140 141 142 143 144 145 146 147
    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

148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165
    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 = []
        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)
166
        base._assert_status_code(200, status_code)
danfengliu's avatar
danfengliu committed
167 168
        return data

169 170 171 172
    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)
173
        base._assert_status_code(200, status_code)
174

175
    def add_project_members(self, project_id, user_id, member_role_id = None, expect_status_code = 201, **kwargs):
danfengliu's avatar
danfengliu committed
176 177 178 179 180 181
        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 = []
182
        data, status_code, header = client.projects_project_id_members_post_with_http_info(project_id, project_member = projectMember)
183
        base._assert_status_code(expect_status_code, status_code)
184 185
        return base._get_id_from_header(header)

186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232
    def add_project_robot_account(self, project_id, project_name, robot_name = None, robot_desc = None, has_pull_right = True,  has_push_right = True,  expect_status_code = 201, **kwargs):
        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"
        resource_by_project_name = "/project/"+project_name+"/repository"
        action_pull = "pull"
        action_push = "push"
        if has_pull_right is True:
            robotAccountAccess = swagger_client.RobotAccountAccess(resource = resource_by_project_id, action = action_pull)
            access_list.append(robotAccountAccess)
            robotAccountAccess = swagger_client.RobotAccountAccess(resource = resource_by_project_name, 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)
            robotAccountAccess = swagger_client.RobotAccountAccess(resource = resource_by_project_name, action = action_push)
            access_list.append(robotAccountAccess)
        robotAccountCreate = swagger_client.RobotAccountCreate(robot_name, robot_desc, access_list)
        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)
        base._assert_status_code(200, status_code)