Commit 894e0591 authored by danfengliu's avatar danfengliu Committed by Yan
Browse files

add new api tests (#6127)

Add API test libraries which were based on swagger doc,
it's including user and project , also add one scenario
test case which is test_add_member_to_private_project to
verify a user can see a project when the uesr is the member of it.
Delete 2 old test case.Delete a quota.
Signed-off-by: default avatardanfengliu <>
parent b4874849
......@@ -25,6 +25,10 @@ def _create_client(server, credential, debug):
cfg.debug = debug
return swagger_client.ProductsApi(swagger_client.ApiClient(cfg))
def _assert_status_code(expect_code, return_code):
if str(return_code) != str(expect_code):
raise Exception(r"HTTPS status code is not {} but {}".format(expect_code, return_code))
def _random_name(prefix):
return "%s-%d" % (prefix, int(round(time.time() * 1000)))
......@@ -32,6 +36,13 @@ def _get_id_from_header(header):
location = header["Location"]
return location.split("/")[-1]
def _get_string_from_unicode(udata):
for u in udata:
tmp = u.encode('utf8')
result = result + tmp.strip('\n\r\t')
return result
class Base:
def __init__(self,
server = Server(endpoint="http://localhost:8080/api", verify_ssl=False),
......@@ -62,4 +73,4 @@ class Base:
credential.username = kwargs.get("username")
if "password" in kwargs:
credential.password = kwargs.get("password")
return _create_client(server, credential, self.debug)
\ No newline at end of file
return _create_client(server, credential, self.debug)
......@@ -5,11 +5,86 @@ import base
import swagger_client
class Project(base.Base):
def create_project(self, name=None,
metadata = {}, **kwargs):
def create_project(self, name=None, metadata=None, **kwargs):
if name is None:
name = base._random_name("project")
if metadata is None:
metadata = {}
client = self._get_client(**kwargs)
_, _, header = client.projects_post_with_http_info(
_, status_code, header = client.projects_post_with_http_info(
swagger_client.ProjectReq(name, metadata))
return base._get_id_from_header(header), name
\ No newline at end of file
base._assert_status_code(201, status_code)
project_id = base._get_id_from_header(header)
return name, project_id
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
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,
def get_project(self, project_id, **kwargs):
client = self._get_client(**kwargs)
data, status_code, _ = client.projects_project_id_get_with_http_info(project_id)
base._assert_status_code(200, status_code)
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)
def delete_project(self, project_id, **kwargs):
client = self._get_client(**kwargs)
_, status_code, _ = client.projects_project_id_delete_with_http_info(project_id)
base._assert_status_code(200, status_code)
def get_project_metadata_by_name(self, project_id, meta_name, **kwargs):
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)
base._assert_status_code(200, status_code)
return {
'public': ProjectMetadata.public,
'enable_content_trust': ProjectMetadata.enable_content_trust,
'prevent_vul': ProjectMetadata.prevent_vul,
'auto_scan': ProjectMetadata.auto_scan,
'severity': ProjectMetadata.severity,
def get_project_members(self, project_id, **kwargs):
client = self._get_client(**kwargs)
data = []
data, status_code, _ = client.projects_project_id_members_get_with_http_info(project_id)
print "****************data for get_project_members:", data
base._assert_status_code(200, status_code)
return data
def add_project_members(self, project_id, user_id, member_role_id = None, **kwargs):
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 = []
data, status_code, _ = client.projects_project_id_members_post_with_http_info(project_id, project_member = projectMember)
base._assert_status_code(201, status_code)
return data
\ No newline at end of file
# -*- coding: utf-8 -*-
import sys
import base
import swagger_client
class Repository(base.Base):
def list_tags(self, repository, **kwargs):
client = self._get_client(**kwargs)
return client.repositories_repo_name_tags_get(repository)
def image_exists(self, repository, tag, **kwargs):
tags = self.list_tags(repository, **kwargs)
exist = False
for t in tags:
if == tag:
exist = True
return exist
def image_should_exist(self, repository, tag, **kwargs):
if not self.image_exists(repository, tag, **kwargs):
raise Exception("image %s:%s not exist" % (repository, tag))
def image_should_not_exist(self, repository, tag, **kwargs):
if self.image_exists(repository, tag, **kwargs):
raise Exception("image %s:%s exists" % (repository, tag))
\ No newline at end of file
# -*- coding: utf-8 -*-
import base
import swagger_client
class User(base.Base):
def create_user(self, name=None,
email = None, user_password=None,realname = None, **kwargs):
if name is None:
name = base._random_name("user")
if realname is None:
realname = base._random_name("realname")
if email is None:
email = '' % (realname,"vmware")
if user_password is None:
user_password = "Harbor12345678"
client = self._get_client(**kwargs)
user = swagger_client.User(None, name, email, user_password, realname, None, None, None, None, None, None, None, None, None)
_, status_code, header = client.users_post_with_http_info(user)
base._assert_status_code(201, status_code)
return base._get_id_from_header(header), name
def get_users(self, username=None, email=None, page=None, page_size=None, **kwargs):
client = self._get_client(**kwargs)
if username is not None:
params["username"] = username
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
data, status_code, _ = client.users_get_with_http_info(**params)
base._assert_status_code(200, status_code)
return data
def get_user(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)
print "data in lib:", data
return data
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, **kwargs):
client = self._get_client(**kwargs)
_, status_code, _ = client.users_user_id_delete_with_http_info(user_id)
base._assert_status_code(200, 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_uesr_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_uesr_role_as_sysadmin(self, user_id, IsAdmin, **kwargs):
client = self._get_client(**kwargs)
has_admin_role = swagger_client.HasAdminRole(IsAdmin)
print "has_admin_role:", has_admin_role
_, status_code, _ = client.users_user_id_sysadmin_put_with_http_info(user_id, has_admin_role)
base._assert_status_code(200, status_code)
return user_id
from __future__ import absolute_import
import unittest
from testutils import CLIENT
from library.project import Project
from library.user import User
class TestProjects(unittest.TestCase):
"""UserGroup unit test stubs"""
def setUp(self):
project = Project()
self.project= project
user = User()
self.user= user
def tearDown(self):
def testAddProjectMember(self):
Test case:
Add a new user to a certain private project as member
Test step and Expectation:
1. Login harbor as admin, then to create a user(UA) with non-admin role;
2. Login harbor as admin, then to create a private project(PA);
3. Login harbor as user(UA), then to get all private projects, projects count must be zero;
4. Login harbor as admin, then to add user(UA) in project(PA);
5. Login harbor as user(UA), then to get all private project, there must be project(PA) only.
url = CLIENT["endpoint"]
user_001_password = "Aa123456"
admin_user = "admin"
admin_pwd = "Harbor12345"
ADMIN_CLIENT=dict(endpoint = url, username = admin_user, password = admin_pwd)
#1. Create user-001
user_001_id, user_001_name = self.user.create_user(user_password = user_001_password, **ADMIN_CLIENT)
self.assertNotEqual(user_001_id, None, msg="Failed to create user, return user is {}".format(user_001_id))
#2. Create private project-001
_metadata = {"public": "false"}
project_001_name, project_001_id = self.project.create_project(metadata = _metadata, **ADMIN_CLIENT)
self.assertNotEqual(project_001_name, None, msg="Project was created failed, return project name is {} and id is {}.".format(project_001_name, project_001_id))
#3.1 Get private projects of user-001
USER_001_CLIENT=dict(endpoint = url, username = user_001_name, password = user_001_password)
params = {}
params["public"] = False
project_001_data = self.project.get_projects(params, **USER_001_CLIENT)
#3.2 Check user-001 has no any private project
self.assertEqual(project_001_data, None, msg="user-001 should has no any private project, but we got {}".format(project_001_data))
#4. Add user-001 as a member of project-001
result = self.project.add_project_members(project_001_id, user_001_id, **ADMIN_CLIENT)
self.assertNotEqual(result, False, msg="Failed to add member user_001 to project_001, result is {}".format(result))
#5 Get private project of uesr-001, uesr-001 can see only one private project which is project-001
params = {}
params["public"] = False
project_data = self.project.get_projects(params, **USER_001_CLIENT)
self.assertEqual(len(project_data), 1, msg="Private project count should be 1.")
self.assertEqual(str(project_data[0].project_id), str(project_001_id), msg="Project-id check failed, please check this test case.")
if __name__ == '__main__':
import time
import os
import sys
from import ApiException
import swagger_client.models
from pprint import pprint
harbor_server = os.environ["HARBOR_HOST"]
def GetProductApi(username, password, harbor_server= os.environ["HARBOR_HOST"]):
cfg = swagger_client.Configuration()
......@@ -16,4 +21,4 @@ def GetProductApi(username, password, harbor_server= os.environ["HARBOR_HOST"]):
cfg.debug = True
api_client = swagger_client.ApiClient(cfg)
api_instance = swagger_client.ProductsApi(api_client)
return api_instance
return api_instance
\ No newline at end of file
......@@ -17,15 +17,5 @@ ${SERVER_API_ENDPOINT} ${SERVER_URL}/api
&{SERVER_CONFIG} endpoint=${SERVER_API_ENDPOINT} verify_ssl=False
*** Test Cases ***
Test Case - Create Project
${project_id} ${project_name} = Create Project
Log To Console ${project_name}
Test Case - Push Image
Sleep 1
Docker Pull hello-world:latest
${project_id} ${project_name} = Create Project
Log To Console ${project_name}
Docker Login ${SERVER} admin Harbor12345
Docker Tag hello-world:latest ${SERVER}/${project_name}/hello-world:1.0
Docker Push ${SERVER}/${project_name}/hello-world:1.0
Test Case - Add Private Project Member and Check User Can See It
Harbor API Test ./tests/apitests/python/
\ No newline at end of file
......@@ -20,7 +20,7 @@ harbor_logs_bucket="harbor-ci-logs"
# GS util
function uploader {
sudo gsutil cp $1 gs://$2/$1
sudo gsutil acl ch -u AllUsers:R gs://$2/$1 &> /dev/null
sudo gsutil acl ch -u AllUsers:R gs://$2/$1
set +e
......@@ -39,7 +39,8 @@ rc=$?
echo $rc
## --------------------------------------------- Upload Harbor CI Logs -------------------------------------------
timestamp=$(date +%s)
sudo tar -zcvf $outfile output.xml log.html /var/log/harbor/*
if [ -f "$outfile" ]; then
uploader $outfile $harbor_logs_bucket
......@@ -51,4 +52,4 @@ else
echo "No log output file to upload"
exit $rc
\ No newline at end of file
exit $rc
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment