mirror of
https://github.com/goharbor/harbor.git
synced 2024-11-17 07:45:24 +01:00
100 lines
4.1 KiB
Python
100 lines
4.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import base
|
|
import swagger_client
|
|
from swagger_client.rest import ApiException
|
|
|
|
class User(base.Base):
|
|
|
|
def create_user(self, name=None,
|
|
email = None, user_password=None, realname = None, role_id = None, expect_status_code=201, **kwargs):
|
|
if name is None:
|
|
name = base._random_name("user")
|
|
if realname is None:
|
|
realname = base._random_name("realname")
|
|
if email is None:
|
|
email = '%s@%s.com' % (realname,"vmware")
|
|
if user_password is None:
|
|
user_password = "Harbor12345678"
|
|
if role_id is None:
|
|
role_id = 0
|
|
|
|
client = self._get_client(**kwargs)
|
|
user = swagger_client.User(username = name, email = email, password = user_password, realname = realname, role_id = role_id)
|
|
|
|
try:
|
|
_, status_code, header = client.users_post_with_http_info(user)
|
|
except ApiException as e:
|
|
base._assert_status_code(expect_status_code, e.status)
|
|
else:
|
|
base._assert_status_code(expect_status_code, status_code)
|
|
return base._get_id_from_header(header), name
|
|
|
|
def get_users(self, user_name=None, email=None, page=None, page_size=None, expect_status_code=200, **kwargs):
|
|
client = self._get_client(**kwargs)
|
|
params={}
|
|
if user_name is not None:
|
|
params["username"] = user_name
|
|
if email is not None:
|
|
params["email"] = email
|
|
if page is not None:
|
|
params["page"] = page
|
|
if page_size is not None:
|
|
params["page_size"] = page_size
|
|
try:
|
|
data, status_code, _ = client.users_get_with_http_info(**params)
|
|
except ApiException as e:
|
|
base._assert_status_code(expect_status_code, e.status)
|
|
else:
|
|
base._assert_status_code(expect_status_code, status_code)
|
|
return data
|
|
|
|
def get_user_by_id(self, user_id, **kwargs):
|
|
client = self._get_client(**kwargs)
|
|
data, status_code, _ = client.users_user_id_get_with_http_info(user_id)
|
|
base._assert_status_code(200, status_code)
|
|
return data
|
|
|
|
def get_user_by_name(self, name, expect_status_code=200, **kwargs):
|
|
users = self.get_users(user_name=name, expect_status_code=expect_status_code , **kwargs)
|
|
for user in users:
|
|
if user.username == name:
|
|
return user
|
|
return None
|
|
|
|
|
|
def get_user_current(self, **kwargs):
|
|
client = self._get_client(**kwargs)
|
|
data, status_code, _ = client.users_current_get_with_http_info()
|
|
base._assert_status_code(200, status_code)
|
|
return data
|
|
|
|
def delete_user(self, user_id, expect_status_code = 200, **kwargs):
|
|
client = self._get_client(**kwargs)
|
|
_, status_code, _ = client.users_user_id_delete_with_http_info(user_id)
|
|
base._assert_status_code(expect_status_code, status_code)
|
|
return user_id
|
|
|
|
def update_user_pwd(self, user_id, new_password=None, old_password=None, **kwargs):
|
|
if old_password is None:
|
|
old_password = ""
|
|
password = swagger_client.Password(old_password, new_password)
|
|
client = self._get_client(**kwargs)
|
|
_, status_code, _ = client.users_user_id_password_put_with_http_info(user_id, password)
|
|
base._assert_status_code(200, status_code)
|
|
return user_id
|
|
|
|
def update_user_profile(self, user_id, email=None, realname=None, comment=None, **kwargs):
|
|
client = self._get_client(**kwargs)
|
|
user_rofile = swagger_client.UserProfile(email, realname, comment)
|
|
_, status_code, _ = client.users_user_id_put_with_http_info(user_id, user_rofile)
|
|
base._assert_status_code(200, status_code)
|
|
return user_id
|
|
|
|
def update_user_role_as_sysadmin(self, user_id, IsAdmin, **kwargs):
|
|
client = self._get_client(**kwargs)
|
|
sysadmin_flag = swagger_client.SysAdminFlag(IsAdmin)
|
|
_, status_code, _ = client.users_user_id_sysadmin_put_with_http_info(user_id, sysadmin_flag)
|
|
base._assert_status_code(200, status_code)
|
|
return user_id
|