Upstream/harbor
1
0
Fork 0
mirror of https://github.com/goharbor/harbor.git synced 2024-09-27 21:12:42 +02:00
Code Issues Projects Releases Wiki Activity

Add py-test for system robot account

Browse Source 
In this py-test, add a new system robot account, and verify all project privileges.

Signed-off-by: danfengliu <danfengl@vmware.com>
This commit is contained in:
danfengliu 2020-12-16 18:06:41 +08:00 committed by Danfeng Liu (c)
parent 9402077695
commit f57dc13081
11 changed files with 438 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
tests/apitests/python/library/artifact.py
View File

@ -56,10 +56,17 @@ class Artifact(base.Base, object):
client = self._get_client(**kwargs)
return client.get_addition_with_http_info(project_name, repo_name, reference, addition)
def add_label_to_reference(self, project_name, repo_name, reference, label_id, **kwargs):
def add_label_to_reference(self, project_name, repo_name, reference, label_id, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
label = v2_swagger_client.Label(id = label_id)
return client.add_label_with_http_info(project_name, repo_name, reference, label)
try:
body, status_code, _ = client.add_label_with_http_info(project_name, repo_name, reference, label)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
else:
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(200, status_code)
return body
def copy_artifact(self, project_name, repo_name, _from, expect_status_code = 201, expect_response_body = None, **kwargs):
client = self._get_client(**kwargs)
@ -84,10 +91,11 @@ class Artifact(base.Base, object):
except ApiException as e:
if e.status == 409 and ignore_conflict == True:
return
else:
raise Exception("Create tag error, {}.".format(e.body))
base._assert_status_code(expect_status_code, e.status)
else:
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(201, status_code)
def delete_tag(self, project_name, repo_name, reference, tag_name, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
@ -97,6 +105,7 @@ class Artifact(base.Base, object):
base._assert_status_code(expect_status_code, e.status)
else:
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(200, status_code)
def check_image_scan_result(self, project_name, repo_name, reference, expected_scan_status = "Success", **kwargs):
timeout_count = 30

18
tests/apitests/python/library/base.py
View File

@ -65,9 +65,9 @@ def _create_client(server, credential, debug, api_type="products"):
"gc": v2_swagger_client.GcApi(v2_swagger_client.ApiClient(cfg)),
}.get(api_type,'Error: Wrong API type')
def _assert_status_code(expect_code, return_code):
def _assert_status_code(expect_code, return_code, err_msg = r"HTTPS status code s not as we expected. Expected {}, while actual HTTPS status code is {}."):
if str(return_code) != str(expect_code):
raise Exception(r"HTTPS status code s not as we expected. Expected {}, while actual HTTPS status code is {}.".format(expect_code, return_code))
raise Exception(err_msg.format(expect_code, return_code))
def _assert_status_body(expect_status_body, returned_status_body):
if str(returned_status_body.strip()).lower().find(expect_status_body.lower()) < 0:
@ -90,15 +90,23 @@ def _get_string_from_unicode(udata):
result = result + tmp.strip('\n\r\t')
return result
def run_command(command):
def run_command(command, expected_error_message = None):
print("Command: ", subprocess.list2cmdline(command))
try:
output = subprocess.check_output(command,
stderr=subprocess.STDOUT,
universal_newlines=True)
except subprocess.CalledProcessError as e:
raise Exception('Error: Exited with error code: %s. Output:%s'% (e.returncode, e.output))
return output
print("Run command error:", str(e))
print("expected_error_message:", expected_error_message)
if expected_error_message is not None:
if str(e.output).lower().find(expected_error_message.lower()) < 0:
raise Exception(r"Error message {} is not as expected {}".format(str(e.output), expected_error_message))
else:
raise Exception('Error: Exited with error code: %s. Output:%s'% (e.returncode, e.output))
else:
print("output:", output)
return output
class Base(object):
def __init__(self, server=None, credential=None, debug=True, api_type="products"):

38
tests/apitests/python/library/chart.py
View File

@ -1,4 +1,5 @@
import base
from client.rest import ApiException
class Chart(base.Base, object):
def __init__(self):
@ -6,23 +7,40 @@ class Chart(base.Base, object):
def upload_chart(self, repository, chart, prov = None, expect_status_code = 201, **kwargs):
client = self._get_client(**kwargs)
_, status_code, _ = client.chartrepo_repo_charts_post_with_http_info(repository, chart)
base._assert_status_code(expect_status_code, status_code)
try:
_, status_code, _ = client.chartrepo_repo_charts_post_with_http_info(repository, chart)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
else:
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(201, status_code)
def get_charts(self, repository, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
body, status_code, _ = client.chartrepo_repo_charts_get_with_http_info(repository)
base._assert_status_code(expect_status_code, status_code)
return body
try:
body, status_code, _ = client.chartrepo_repo_charts_get_with_http_info(repository)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
return []
else:
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(200, status_code)
return body
def chart_should_exist(self, repository, chart_name, **kwargs):
charts_data = self.get_charts(repository, **kwargs)
def chart_should_exist(self, repository, chart_name, expect_status_code = 200, **kwargs):
charts_data = self.get_charts(repository, expect_status_code = expect_status_code, **kwargs)
for chart in charts_data:
if chart.name == chart_name:
return True
raise Exception(r"Chart {} does not exist in project {}.".format(chart_name, repository))
if expect_status_code == 200:
raise Exception(r"Chart {} does not exist in project {}.".format(chart_name, repository))
def delete_chart_with_version(self, repository, chart_name, version, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
_, status_code, _ = client.chartrepo_repo_charts_name_version_delete_with_http_info(repository, chart_name, version)
base._assert_status_code(expect_status_code, status_code)
try:
_, status_code, _ = client.chartrepo_repo_charts_name_version_delete_with_http_info(repository, chart_name, version)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
else:
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(200, status_code)

27
tests/apitests/python/library/helm.py
View File

@ -14,18 +14,15 @@ def get_chart_file(file_name):
def helm_login(harbor_server, user, password):
os.putenv("HELM_EXPERIMENTAL_OCI", "1")
command = ["helm3", "registry", "login", harbor_server, "-u", user, "-p", password]
print("Command: ", command)
ret = base.run_command(command)
print("Command return: ", ret)
def helm_save(chart_archive, harbor_server, project, repo_name):
command = ["helm3", "chart","save", chart_archive, harbor_server+"/"+project+"/"+repo_name]
print("Command: ", command)
base.run_command(command)
def helm_push(harbor_server, project, repo_name, version):
command = ["helm3", "chart","push", harbor_server+"/"+project+"/"+repo_name+":"+version]
print("Command: ", command)
ret = base.run_command(command)
return ret
@ -35,13 +32,27 @@ def helm_chart_push_to_harbor(chart_file, archive, harbor_server, project, repo_
helm_save(archive, harbor_server, project, repo_name)
return helm_push(harbor_server, project, repo_name, version)
def helm2_add_repo(helm_repo_name, harbor_url, project, username, password):
def helm2_add_repo(helm_repo_name, harbor_url, project, username, password, expected_error_message = None):
command = ["helm2", "repo", "add", "--username=" + username, "--password=" + password, helm_repo_name, harbor_url + "/chartrepo/" + project]
print("Command: ", command)
base.run_command(command)
ret = base.run_command(command, expected_error_message = expected_error_message)
def helm2_push(helm_repo_name, chart_file, project, username, password):
get_chart_file(chart_file)
command = ["helm2", "push", "--username=" + username, "--password=" + password, chart_file.split('/')[-1], helm_repo_name]
print("Command: ", command)
base.run_command(command)
base.run_command(command)
def helm2_repo_update():
command = ["helm2", "repo", "update"]
base.run_command(command)
def helm2_fetch_chart_file(helm_repo_name, harbor_url, project, username, password, chart_file, expected_add_repo_error_message = None):
helm2_add_repo(helm_repo_name, harbor_url, project, username, password, expected_error_message = expected_add_repo_error_message)
if expected_add_repo_error_message is not None:
return
helm2_repo_update()
command_ls = ["ls"]
base.run_command(command_ls)
command = ["helm2", "fetch", "{}/{}".format(helm_repo_name, chart_file)]
base.run_command(command)
base.run_command(command_ls)

27
tests/apitests/python/library/label.py
View File

@ -3,24 +3,27 @@
import sys
import base
import swagger_client
from swagger_client.rest import ApiException
class Label(base.Base):
def create_label(self, name=None, desc="",
color="", scope="g", project_id=0, **kwargs):
def create_label(self, name=None, desc="", color="", scope="g",
project_id=0, expect_status_code = 201, **kwargs):
if name is None:
name = base._random_name("label")
label = swagger_client.Label(name=name,
description=desc, color=color,
label = swagger_client.Label(name=name,
description=desc, color=color,
scope=scope, project_id=project_id)
client = self._get_client(**kwargs)
_, _, header = client.labels_post_with_http_info(label)
return base._get_id_from_header(header), name
def add_label_to_image(self, label_id, repository, tag, **kwargs):
client = self._get_client(**kwargs)
return client.repositories_repo_name_tags_tag_labels_post(repository,
tag, swagger_client.Label(id=int(label_id)))
try:
_, status_code, header = client.labels_post_with_http_info(label)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
else:
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(201, status_code)
return base._get_id_from_header(header), name
def delete_label(self, label_id, **kwargs):
client = self._get_client(**kwargs)
return client.labels_id_delete_with_http_info(int(label_id))

4
tests/apitests/python/library/repository.py
View File

@ -41,6 +41,10 @@ def push_self_build_image_to_project(project_name, registry, username, password,
_docker_api = DockerAPI()
_docker_api.docker_login(registry, username, password, expected_error_message = expected_login_error_message)
time.sleep(2)
if expected_login_error_message != None:
return
push_special_image_to_project(project_name, registry, username, password, image, tags=[tag], size=size, expected_login_error_message = expected_login_error_message, expected_error_message = expected_error_message)
return r'{}/{}'.format(project_name, image), tag

77
tests/apitests/python/library/robot.py
View File

@ -4,11 +4,52 @@ import time
import base
import v2_swagger_client
from v2_swagger_client.rest import ApiException
from base import _assert_status_code
class Robot(base.Base, object):
def __init__(self):
super(Robot,self).__init__(api_type = "robot")
def list_robot(self, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
try:
body, status_code, _ = client.list_robot_with_http_info()
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
return []
else:
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(200, status_code)
return body
def create_access_list(self, right_map = [True] * 10):
_assert_status_code(10, len(right_map), r"Please input full access list for system robot account. Expected {}, while actual input count is {}.")
action_pull = "pull"
action_push = "push"
action_read = "read"
action_create = "create"
action_del = "delete"
access_def_list = [
("repository", action_pull),
("repository", action_push),
("artifact", action_del),
("helm-chart", action_read),
("helm-chart-version", action_create),
("helm-chart-version", action_del),
("tag", action_create),
("tag", action_del),
("artifact-label", action_create),
("scan", action_create)
]
access_list = []
for i in range(len(access_def_list)):
if right_map[i] is True:
robotAccountAccess = v2_swagger_client.Access(resource = access_def_list[i][0], action = access_def_list[i][1])
access_list.append(robotAccountAccess)
return access_list
def create_project_robot(self, project_name, duration, 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):
if robot_name is None:
robot_name = base._random_name("robot")
@ -65,3 +106,39 @@ class Robot(base.Base, object):
_, status_code, _ = client.delete_robot_with_http_info(robot_id)
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(200, status_code)
def create_system_robot(self, permission_list, duration, robot_name = None, robot_desc = None, 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")
robotAccountCreate = v2_swagger_client.RobotCreate(name=robot_name, description=robot_desc, duration=duration, level="system", disable = False, permissions = permission_list)
print("robotAccountCreate:", robotAccountCreate)
client = self._get_client(**kwargs)
data = []
data, status_code, header = client.create_robot_with_http_info(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 update_robot_account(self, robot_id, robot, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
_, status_code, _ = client.update_robot_with_http_info(robot_id, robot)
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(200, status_code)
def update_system_robot_account(self, robot_id, robot_name, robot_account_Permissions_list, disable = None, expect_status_code = 200, **kwargs):
robot = v2_swagger_client.Robot(id = robot_id, name = robot_name, level = "system", permissions = robot_account_Permissions_list)
if disable in (True, False):
robot.disable = disable
self.update_robot_account(robot_id, robot, expect_status_code = expect_status_code, **kwargs)
def refresh_robot_account_secret(self, robot_id, robot_new_sec, expect_status_code = 200, **kwargs):
robot_sec = v2_swagger_client.RobotSec(secret = robot_new_sec)
client = self._get_client(**kwargs)
data, status_code, _ = client.refresh_sec_with_http_info(robot_id, robot_sec)
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(200, status_code)
print("Refresh new secret:", data)
return data

293
tests/apitests/python/test_robot_account.py
View File

@ -1,43 +1,58 @@
from __future__ import absolute_import
import sys
import unittest
from testutils import ADMIN_CLIENT, suppress_urllib3_warning
from testutils import TEARDOWN
from testutils import harbor_server
from testutils import ADMIN_CLIENT, CHART_API_CLIENT, TEARDOWN, harbor_server, harbor_url, suppress_urllib3_warning
from testutils import created_user, created_project
from library.user import User
from library.project import Project
from library.robot import Robot
from library.repository import Repository
from library.artifact import Artifact
from library.repository import pull_harbor_image
from library.repository import push_self_build_image_to_project
from library.base import _assert_status_code
from library.scan import Scan
from library.label import Label
from library.chart import Chart
import library.helm
import base
import v2_swagger_client
class TestProjects(unittest.TestCase):
class TestRobotAccount(unittest.TestCase):
@suppress_urllib3_warning
def setUp(self):
self.project = Project()
self.user = User()
self.repo = Repository()
self.artifact = Artifact()
self.robot = Robot()
self.scan = Scan()
self.label = Label()
self.chart= Chart()
@unittest.skipIf(TEARDOWN == False, "Test data won't be erased.")
def tearDown(self):
TestRobotAccount.url = ADMIN_CLIENT["endpoint"]
TestRobotAccount.user_ra_password = "Aa123456"
print("setup")
@unittest.skipIf(TEARDOWN == True, "Test data won't be erased.")
def do_01_tearDown(self):
#1. Delete repository(RA) by user(UA);
self.repo.delete_repoitory(TestProjects.project_ra_name_a, TestProjects.repo_name_in_project_a.split('/')[1], **TestProjects.USER_RA_CLIENT)
self.repo.delete_repoitory(TestProjects.project_ra_name_b, TestProjects.repo_name_in_project_b.split('/')[1], **TestProjects.USER_RA_CLIENT)
self.repo.delete_repoitory(TestProjects.project_ra_name_c, TestProjects.repo_name_in_project_c.split('/')[1], **TestProjects.USER_RA_CLIENT)
self.repo.delete_repoitory(TestProjects.project_ra_name_a, TestProjects.repo_name_pa.split('/')[1], **TestProjects.USER_RA_CLIENT)
self.repo.delete_repoitory(self.project_ra_name_a, self.repo_name_in_project_a.split('/')[1], **self.USER_RA_CLIENT)
self.repo.delete_repoitory(self.project_ra_name_b, self.repo_name_in_project_b.split('/')[1], **self.USER_RA_CLIENT)
self.repo.delete_repoitory(self.project_ra_name_c, self.repo_name_in_project_c.split('/')[1], **self.USER_RA_CLIENT)
self.repo.delete_repoitory(self.project_ra_name_a, self.repo_name_pa.split('/')[1], **self.USER_RA_CLIENT)
#2. Delete project(PA);
self.project.delete_project(TestProjects.project_ra_id_a, **TestProjects.USER_RA_CLIENT)
self.project.delete_project(TestProjects.project_ra_id_b, **TestProjects.USER_RA_CLIENT)
self.project.delete_project(TestProjects.project_ra_id_c, **TestProjects.USER_RA_CLIENT)
self.project.delete_project(self.project_ra_id_a, **self.USER_RA_CLIENT)
self.project.delete_project(self.project_ra_id_b, **self.USER_RA_CLIENT)
self.project.delete_project(self.project_ra_id_c, **self.USER_RA_CLIENT)
#3. Delete user(UA).
self.user.delete_user(TestProjects.user_ra_id, **ADMIN_CLIENT)
self.user.delete_user(self.user_ra_id, **ADMIN_CLIENT)
def testRobotAccount(self):
def test_01_ProjectlevelRobotAccount(self):
"""
Test case:
Robot Account
@ -62,10 +77,6 @@ class TestProjects(unittest.TestCase):
2. Delete project(PA);
3. Delete user(UA).
"""
url = ADMIN_CLIENT["endpoint"]
admin_name = ADMIN_CLIENT["username"]
admin_password = ADMIN_CLIENT["password"]
user_ra_password = "Aa123456"
image_project_a = "haproxy"
image_project_b = "hello-world"
image_project_c = "httpd"
@ -73,56 +84,256 @@ class TestProjects(unittest.TestCase):
tag = "latest"
#1. Create user(UA);"
TestProjects.user_ra_id, user_ra_name = self.user.create_user(user_password = user_ra_password, **ADMIN_CLIENT)
TestProjects.USER_RA_CLIENT=dict(endpoint = url, username = user_ra_name, password = user_ra_password)
self.user_ra_id, user_ra_name = self.user.create_user(user_password = TestRobotAccount.user_ra_password, **ADMIN_CLIENT)
self.USER_RA_CLIENT=dict(endpoint = TestRobotAccount.url, username = user_ra_name, password = TestRobotAccount.user_ra_password)
#2. Create private project(PA), private project(PB) and public project(PC) by user(UA);
TestProjects.project_ra_id_a, TestProjects.project_ra_name_a = self.project.create_project(metadata = {"public": "false"}, **TestProjects.USER_RA_CLIENT)
TestProjects.project_ra_id_b, TestProjects.project_ra_name_b = self.project.create_project(metadata = {"public": "false"}, **TestProjects.USER_RA_CLIENT)
TestProjects.project_ra_id_c, TestProjects.project_ra_name_c = self.project.create_project(metadata = {"public": "true"}, **TestProjects.USER_RA_CLIENT)
self.project_ra_id_a, self.project_ra_name_a = self.project.create_project(metadata = {"public": "false"}, **self.USER_RA_CLIENT)
self.project_ra_id_b, self.project_ra_name_b = self.project.create_project(metadata = {"public": "false"}, **self.USER_RA_CLIENT)
self.project_ra_id_c, self.project_ra_name_c = self.project.create_project(metadata = {"public": "true"}, **self.USER_RA_CLIENT)
#3. Push image(ImagePA) to project(PA), image(ImagePB) to project(PB) and image(ImagePC) to project(PC) by user(UA);
TestProjects.repo_name_in_project_a, tag_a = push_self_build_image_to_project(TestProjects.project_ra_name_a, harbor_server, user_ra_name, user_ra_password, image_project_a, tag)
TestProjects.repo_name_in_project_b, tag_b = push_self_build_image_to_project(TestProjects.project_ra_name_b, harbor_server, user_ra_name, user_ra_password, image_project_b, tag)
TestProjects.repo_name_in_project_c, tag_c = push_self_build_image_to_project(TestProjects.project_ra_name_c, harbor_server, user_ra_name, user_ra_password, image_project_c, tag)
self.repo_name_in_project_a, tag_a = push_self_build_image_to_project(self.project_ra_name_a, harbor_server, user_ra_name, TestRobotAccount.user_ra_password, image_project_a, tag)
self.repo_name_in_project_b, tag_b = push_self_build_image_to_project(self.project_ra_name_b, harbor_server, user_ra_name, TestRobotAccount.user_ra_password, image_project_b, tag)
self.repo_name_in_project_c, tag_c = push_self_build_image_to_project(self.project_ra_name_c, harbor_server, user_ra_name, TestRobotAccount.user_ra_password, image_project_c, tag)
#4. Create a new robot account(RA) with pull and push privilege in project(PA) by user(UA);
robot_id, robot_account = self.robot.create_project_robot(TestProjects.project_ra_name_a,
30 ,**TestProjects.USER_RA_CLIENT)
robot_id, robot_account = self.robot.create_project_robot(self.project_ra_name_a,
30 ,**self.USER_RA_CLIENT)
#5. Check robot account info, it should has both pull and push privilege;
data = self.robot.get_robot_account_by_id(robot_id, **TestProjects.USER_RA_CLIENT)
data = self.robot.get_robot_account_by_id(robot_id, **self.USER_RA_CLIENT)
_assert_status_code(robot_account.name, data.name)
#6. Pull image(ImagePA) from project(PA) by robot account(RA), it must be successful;
pull_harbor_image(harbor_server, robot_account.name, robot_account.secret, TestProjects.repo_name_in_project_a, tag_a)
pull_harbor_image(harbor_server, robot_account.name, robot_account.secret, self.repo_name_in_project_a, tag_a)
#7. Push image(ImageRA) to project(PA) by robot account(RA), it must be successful;
TestProjects.repo_name_pa, _ = push_self_build_image_to_project(TestProjects.project_ra_name_a, harbor_server, robot_account.name, robot_account.secret, image_robot_account, tag)
self.repo_name_pa, _ = push_self_build_image_to_project(self.project_ra_name_a, harbor_server, robot_account.name, robot_account.secret, image_robot_account, tag)
#8. Push image(ImageRA) to project(PB) by robot account(RA), it must be not successful;
push_self_build_image_to_project(TestProjects.project_ra_name_b, harbor_server, robot_account.name, robot_account.secret, image_robot_account, tag, expected_error_message = "unauthorized to access repository")
push_self_build_image_to_project(self.project_ra_name_b, harbor_server, robot_account.name, robot_account.secret, image_robot_account, tag, expected_error_message = "unauthorized to access repository")
#9. Pull image(ImagePB) from project(PB) by robot account(RA), it must be not successful;
pull_harbor_image(harbor_server, robot_account.name, robot_account.secret, TestProjects.repo_name_in_project_b, tag_b, expected_error_message = "unauthorized to access repository")
pull_harbor_image(harbor_server, robot_account.name, robot_account.secret, self.repo_name_in_project_b, tag_b, expected_error_message = "unauthorized to access repository")
#10. Pull image from project(PC), it must be successful;
pull_harbor_image(harbor_server, robot_account.name, robot_account.secret, TestProjects.repo_name_in_project_c, tag_c)
pull_harbor_image(harbor_server, robot_account.name, robot_account.secret, self.repo_name_in_project_c, tag_c)
#11. Push image(ImageRA) to project(PC) by robot account(RA), it must be not successful;
push_self_build_image_to_project(TestProjects.project_ra_name_c, harbor_server, robot_account.name, robot_account.secret, image_robot_account, tag, expected_error_message = "unauthorized to access repository")
push_self_build_image_to_project(self.project_ra_name_c, harbor_server, robot_account.name, robot_account.secret, image_robot_account, tag, expected_error_message = "unauthorized to access repository")
#12. Update action property of robot account(RA);"
self.robot.disable_robot_account(robot_id, True, **TestProjects.USER_RA_CLIENT)
self.robot.disable_robot_account(robot_id, True, **self.USER_RA_CLIENT)
#13. Pull image(ImagePA) from project(PA) by robot account(RA), it must be not successful;
pull_harbor_image(harbor_server, robot_account.name, robot_account.secret, TestProjects.repo_name_in_project_a, tag_a, expected_login_error_message = "unauthorized: authentication required")
pull_harbor_image(harbor_server, robot_account.name, robot_account.secret, self.repo_name_in_project_a, tag_a, expected_login_error_message = "unauthorized: authentication required")
#14. Push image(ImageRA) to project(PA) by robot account(RA), it must be not successful;
push_self_build_image_to_project(TestProjects.project_ra_name_a, harbor_server, robot_account.name, robot_account.secret, image_robot_account, tag, expected_login_error_message = "unauthorized: authentication required")
push_self_build_image_to_project(self.project_ra_name_a, harbor_server, robot_account.name, robot_account.secret, image_robot_account, tag, expected_login_error_message = "unauthorized: authentication required")
#15. Delete robot account(RA), it must be not successful.
self.robot.delete_robot_account(robot_id, **TestProjects.USER_RA_CLIENT)
self.robot.delete_robot_account(robot_id, **self.USER_RA_CLIENT)
self.do_01_tearDown()
def verify_repository_pushable(self, project_access_list, system_ra_client):
for project_access in project_access_list:
print(r"project_access:", project_access)
if project_access["check_list"][1]: #---repository:push---
repo = push_self_build_image_to_project(project_access["project_name"], harbor_server, system_ra_client["username"], system_ra_client["password"], "test_pushable"+base._random_name("repo"), "v6.8.1"+base._random_name("tag"))
else:
push_self_build_image_to_project(project_access["project_name"], harbor_server, system_ra_client["username"], system_ra_client["password"], "test_unpushable"+base._random_name("repo"), "v6.8.1"+base._random_name("tag"), expected_error_message = "unauthorized to access repository")
def verify_repository_unpushable(self, project_access_list, system_ra_client, expected_login_error_message = "unauthorized: authentication required"):
for project_access in project_access_list: #---repository:push---
push_self_build_image_to_project(
project_access["project_name"],
harbor_server, system_ra_client["username"], system_ra_client["password"],
"test_unpushable"+base._random_name("repo"), "v6.8.1"+base._random_name("tag"),
expected_login_error_message = expected_login_error_message
)
def test_02_SystemlevelRobotAccount(self):
"""
Test case:
Robot Account
Test step and expected result:
1. Define a number of access lists;
2. Create the same number of private projects;
3. Create a system robot account has permission for those projects;
4. Verify the system robot account has the corresponding rights;
5. Disable the system robot account;
6. Verify the system robot account has no the corresponding rights;
7. Enable the system robot account;
8. Verify the system robot account has the corresponding rights;
9. Refresh secret for the system robot account;
10. Verify the system robot account has no the corresponding right with the old secret already;
11. Verify the system robot account still has the corresponding right with the new secret;
12. List system robot account, then add a new project to the system robot account project permission list;
13. Delete this project;
14. List system robot account successfully;
15. Delete the system robot account;
16. Verify the system robot account has no the corresponding right;
17. Add a system robot account with all project coverd;
18. Verify the system robot account has no the corresponding right;
"""
#1. Define a number of access lists;
CHART_FILE_LIST = [dict(name = 'prometheus', version='7.0.2'), dict(name = 'harbor', version='0.2.0')]
for i in range(2):
base.run_command( ["curl", r"-o", "./tests/apitests/python/{}-{}.tgz".format(CHART_FILE_LIST[i]["name"], CHART_FILE_LIST[i]["version"]), "https://storage.googleapis.com/harbor-builds/helm-chart-test-files/{}-{}.tgz".format(CHART_FILE_LIST[i]["name"], CHART_FILE_LIST[i]["version"])])
#Make sure that whether 'True' or 'False' must be included in each line or row.
check_list = [
[True, True, True, True, True, True, False, True, False, True],
[False, False, False, False, True, True, False, True, True, False],
[True, False, True, False, True, False, True, False, True, True],
[False, False, False, True, False, True, False, True, True, False]
]
access_list_list = []
for i in range(len(check_list)):
access_list_list.append(self.robot.create_access_list(check_list[i]))
#2. Create the same number of private projects;
robot_account_Permissions_list = []
project_access_list = []
for i in range(len(check_list)):
with created_user(TestRobotAccount.user_ra_password, _teardown = False) as (user_id, username):
with created_project(metadata={"public": "false"}, user_id=user_id, _teardown = False) as (project_id, project_name):
project_access_list.append(dict(project_name = project_name, project_id = project_id, check_list = check_list[i]))
robot_account_Permissions = v2_swagger_client.Permission(kind = "project", namespace = project_name, access = access_list_list[i])
robot_account_Permissions_list.append(robot_account_Permissions)
#3. Create a system robot account has permission for those projects;
system_robot_account_id, system_robot_account = self.robot.create_system_robot(robot_account_Permissions_list, 300)
print("system_robot_account:", system_robot_account)
SYSTEM_RA_CLIENT = dict(endpoint = TestRobotAccount.url, username = system_robot_account.name, password = system_robot_account.secret)
SYSTEM_RA_CHART_CLIENT = dict(endpoint = CHART_API_CLIENT["endpoint"], username = SYSTEM_RA_CLIENT["username"], password = SYSTEM_RA_CLIENT["password"])
#4. Verify the system robot account has the corresponding rights;
for project_access in project_access_list:
print(r"project_access:", project_access)
if project_access["check_list"][1]: #---repository:push---
repo = push_self_build_image_to_project(project_access["project_name"], harbor_server, SYSTEM_RA_CLIENT["username"], SYSTEM_RA_CLIENT["password"], "test_pushable", "v6.8.1")
else:
push_self_build_image_to_project(project_access["project_name"], harbor_server, SYSTEM_RA_CLIENT["username"], SYSTEM_RA_CLIENT["password"], "test_unpushable", "v6.8.1", expected_error_message = "unauthorized to access repository")
tag_for_del = "v1.0.0"
repo_name, tag = push_self_build_image_to_project(project_access["project_name"], harbor_server, ADMIN_CLIENT["username"], ADMIN_CLIENT["password"], "test_del_artifact", tag_for_del)
if project_access["check_list"][0]: #---repository:pull---
pull_harbor_image(harbor_server, SYSTEM_RA_CLIENT["username"], SYSTEM_RA_CLIENT["password"], repo_name, tag_for_del)
else:
pull_harbor_image(harbor_server, SYSTEM_RA_CLIENT["username"], SYSTEM_RA_CLIENT["password"], repo_name, tag_for_del, expected_error_message = "action: pull: unauthorized to access repository")
if project_access["check_list"][2]: #---artifact:delete---
self.artifact.delete_artifact(project_access["project_name"], repo_name.split('/')[1], tag_for_del, **SYSTEM_RA_CLIENT)
else:
self.artifact.delete_artifact(project_access["project_name"], repo_name.split('/')[1], tag_for_del, expect_status_code = 403, **SYSTEM_RA_CLIENT)
#Prepare for chart read and delete
self.chart.upload_chart(project_access["project_name"], r'./tests/apitests/python/{}-{}.tgz'.format(CHART_FILE_LIST[1]["name"], CHART_FILE_LIST[1]["version"]), **CHART_API_CLIENT)
if project_access["check_list"][3]: #---helm-chart:read---
library.helm.helm2_fetch_chart_file("chart_repo_" + base._random_name("repo"), harbor_url, project_access["project_name"], SYSTEM_RA_CLIENT["username"], SYSTEM_RA_CLIENT["password"], CHART_FILE_LIST[1]["name"])
else:
library.helm.helm2_fetch_chart_file("chart_repo_" + base._random_name("repo"), harbor_url, project_access["project_name"], SYSTEM_RA_CLIENT["username"], SYSTEM_RA_CLIENT["password"], CHART_FILE_LIST[1]["name"], expected_add_repo_error_message = "403 Forbidden")
if project_access["check_list"][4]: #---helm-chart-version:create---
self.chart.upload_chart(project_access["project_name"], r'./tests/apitests/python/{}-{}.tgz'.format(CHART_FILE_LIST[0]["name"], CHART_FILE_LIST[0]["version"]), **SYSTEM_RA_CHART_CLIENT)
else:
self.chart.upload_chart(project_access["project_name"], r'./tests/apitests/python/{}-{}.tgz'.format(CHART_FILE_LIST[0]["name"], CHART_FILE_LIST[0]["version"]), expect_status_code = 403, **SYSTEM_RA_CHART_CLIENT)
if project_access["check_list"][5]: #---helm-chart-version:delete---
self.chart.delete_chart_with_version(project_access["project_name"], CHART_FILE_LIST[1]["name"], CHART_FILE_LIST[1]["version"], **SYSTEM_RA_CHART_CLIENT)
else:
self.chart.delete_chart_with_version(project_access["project_name"], CHART_FILE_LIST[1]["name"], CHART_FILE_LIST[1]["version"], expect_status_code = 403, **SYSTEM_RA_CHART_CLIENT)
repo_name, tag = push_self_build_image_to_project(project_access["project_name"], harbor_server, ADMIN_CLIENT["username"], ADMIN_CLIENT["password"], "test_create_tag", "latest_1")
self.artifact.create_tag(project_access["project_name"], repo_name.split('/')[1], tag, "for_delete", **ADMIN_CLIENT)
if project_access["check_list"][6]: #---tag:create---
self.artifact.create_tag(project_access["project_name"], repo_name.split('/')[1], tag, "1.0", **SYSTEM_RA_CLIENT)
else:
self.artifact.create_tag(project_access["project_name"], repo_name.split('/')[1], tag, "1.0", expect_status_code = 403, **SYSTEM_RA_CLIENT)
if project_access["check_list"][7]: #---tag:delete---
self.artifact.delete_tag(project_access["project_name"], repo_name.split('/')[1], tag, "for_delete", **SYSTEM_RA_CLIENT)
else:
self.artifact.delete_tag(project_access["project_name"], repo_name.split('/')[1], tag, "for_delete", expect_status_code = 403, **SYSTEM_RA_CLIENT)
repo_name, tag = push_self_build_image_to_project(project_access["project_name"], harbor_server, ADMIN_CLIENT["username"], ADMIN_CLIENT["password"], "test_create_artifact_label", "latest_1")
#Add project level label to artifact
label_id, _ = self.label.create_label(project_id = project_access["project_id"], scope = "p", **ADMIN_CLIENT)
if project_access["check_list"][8]: #---artifact-label:create---
self.artifact.add_label_to_reference(project_access["project_name"], repo_name.split('/')[1], tag, int(label_id), **SYSTEM_RA_CLIENT)
else:
self.artifact.add_label_to_reference(project_access["project_name"], repo_name.split('/')[1], tag, int(label_id), expect_status_code = 403, **SYSTEM_RA_CLIENT)
if project_access["check_list"][9]: #---scan:create---
self.scan.scan_artifact(project_access["project_name"], repo_name.split('/')[1], tag, **SYSTEM_RA_CLIENT)
else:
self.scan.scan_artifact(project_access["project_name"], repo_name.split('/')[1], tag, expect_status_code = 403, **SYSTEM_RA_CLIENT)
#5. Disable the system robot account;
self.robot.update_system_robot_account(system_robot_account_id, system_robot_account.name, robot_account_Permissions_list, disable = True, **ADMIN_CLIENT)
#6. Verify the system robot account has no the corresponding rights;
self.verify_repository_unpushable(project_access_list, SYSTEM_RA_CLIENT)
#7. Enable the system robot account;
self.robot.update_system_robot_account(system_robot_account_id, system_robot_account.name, robot_account_Permissions_list, disable = False, **ADMIN_CLIENT)
#8. Verify the system robot account has the corresponding rights;
self.verify_repository_pushable(project_access_list, SYSTEM_RA_CLIENT)
#9. Refresh secret for the system robot account;
new_secret = "new_secret_At_321"
self.robot.refresh_robot_account_secret(system_robot_account_id, new_secret, **ADMIN_CLIENT)
#10. Verify the system robot account has no the corresponding right with the old secret already;
self.verify_repository_unpushable(project_access_list, SYSTEM_RA_CLIENT)
#11. Verify the system robot account still has the corresponding right with the new secret;
SYSTEM_RA_CLIENT["password"] = new_secret
self.verify_repository_pushable(project_access_list, SYSTEM_RA_CLIENT)
#12. List system robot account, then add a new project to the system robot account project permission list;
self.robot.list_robot(**ADMIN_CLIENT)
project_for_del_id, project_for_del_name = self.project.create_project(metadata = {"public": "true"}, **ADMIN_CLIENT)
robot_account_Permissions = v2_swagger_client.Permission(kind = "project", namespace = project_for_del_name, access = access_list_list[0])
robot_account_Permissions_list.append(robot_account_Permissions)
self.robot.update_system_robot_account(system_robot_account_id, system_robot_account.name, robot_account_Permissions_list, **ADMIN_CLIENT)
self.robot.list_robot(**ADMIN_CLIENT)
#13. Delete this project;
self.project.delete_project(project_for_del_id, **ADMIN_CLIENT)
#14. List system robot account successfully;
self.robot.list_robot(**ADMIN_CLIENT)
#15. Delete the system robot account;
self.robot.delete_robot_account(system_robot_account_id, **ADMIN_CLIENT)
#16. Verify the system robot account has no the corresponding right;
self.verify_repository_unpushable(project_access_list, SYSTEM_RA_CLIENT)
#17. Add a system robot account with all project coverd;
all_true_access_list= self.robot.create_access_list( [True] * 10 )
robot_account_Permissions_list = []
robot_account_Permissions = v2_swagger_client.Permission(kind = "project", namespace = "*", access = all_true_access_list)
robot_account_Permissions_list.append(robot_account_Permissions)
_, system_robot_account_cover_all = self.robot.create_system_robot(robot_account_Permissions_list, 300)
#18. Verify the system robot account has no the corresponding right;
print("system_robot_account_cover_all:", system_robot_account_cover_all)
SYSTEM_RA_CLIENT_COVER_ALL = dict(endpoint = TestRobotAccount.url, username = system_robot_account_cover_all.name, password = system_robot_account_cover_all.secret)
projects = self.project.get_projects(dict(), **ADMIN_CLIENT)
print("All projects:", projects)
project_access_list = []
for i in range(len(projects)):
project_access_list.append(dict(project_name = projects[i].name, project_id = projects[i].project_id, check_list = all_true_access_list))
self.verify_repository_pushable(project_access_list, SYSTEM_RA_CLIENT_COVER_ALL)
if __name__ == '__main__':
unittest.main()
suite = unittest.TestSuite(unittest.makeSuite(TestRobotAccount))
result = unittest.TextTestRunner(sys.stdout, verbosity=2, failfast=True).run(suite)
if not result.wasSuccessful():
raise Exception(r"Robot account test failed: {}".format(result))

16
tests/apitests/python/testutils.py
View File

@ -23,6 +23,7 @@ admin_user = "admin"
admin_pwd = "Harbor12345"
harbor_server = os.environ.get("HARBOR_HOST", '')
harbor_url = "https://{}".format(harbor_server)
#CLIENT=dict(endpoint="https://"+harbor_server+"/api")
ADMIN_CLIENT=dict(endpoint = os.environ.get("HARBOR_HOST_SCHEMA", "https")+ "://"+harbor_server+"/api/v2.0", username = admin_user, password = admin_pwd)
CHART_API_CLIENT=dict(endpoint = os.environ.get("HARBOR_HOST_SCHEMA", "https")+ "://"+harbor_server+"/api", username = admin_user, password = admin_pwd)
@ -82,24 +83,31 @@ def suppress_urllib3_warning(func):
from contextlib import contextmanager
@contextmanager
def created_user(password):
def created_user(password, _teardown=None):
from library.user import User
api = User()
is_teardown = TEARDOWN
if _teardown in (True, False):
is_teardown = _teardown
user_id, user_name = api.create_user(user_password=password, **ADMIN_CLIENT)
try:
yield (user_id, user_name)
finally:
if TEARDOWN:
if is_teardown:
api.delete_user(user_id, **ADMIN_CLIENT)
@contextmanager
def created_project(name=None, metadata=None, user_id=None, member_role_id=None):
def created_project(name=None, metadata=None, user_id=None, member_role_id=None, _teardown=None):
from library.project import Project
api = Project()
is_teardown = TEARDOWN
if _teardown in (True, False):
is_teardown = _teardown
project_id, project_name = api.create_project(name=name, metadata=metadata, **ADMIN_CLIENT)
if user_id:
api.add_project_members(project_id, user_id=user_id, member_role_id=member_role_id, **ADMIN_CLIENT)
@ -107,5 +115,5 @@ def created_project(name=None, metadata=None, user_id=None, member_role_id=None)
try:
yield (project_id, project_name)
finally:
if TEARDOWN:
if is_teardown:
api.delete_project(project_id, **ADMIN_CLIENT)

7
tests/resources/Harbor-Pages/Robot_Account.robot
View File

@ -91,8 +91,8 @@ Create A New System Robot Account
... Retry Element Click xpath=${sys_robot_account_expiration_type_select}//option[@value='${expiration_type}']
Run Keyword If '${description}' != '${null}' Retry Text Input ${sys_robot_account_description_textarea} ${description}
Run Keyword If '${is_cover_all}' == '${true}' Retry Double Keywords When Error Retry Element Click ${sys_robot_account_coverall_chb} Retry Checkbox Should Be Selected ${sys_robot_account_coverall_chb_input}
#Clear Global Permissions
Clear Global Permissions By JaveScript
... ELSE Clear Global Permissions By JaveScript
# Select project
FOR ${project} IN @{project_permission_list}
Log To Console project: ${project}
@ -118,5 +118,6 @@ System Robot Account Exist
[Arguments] ${name} ${project_count}
Retry Double Keywords When Error Retry Element Click ${filter_dist_btn} Wait Until Element Is Visible And Enabled ${filter_dist_input}
Retry Text Input ${filter_dist_input} ${name}
Retry Wait Until Page Contains Element //clr-dg-row[contains(.,'${name}') and contains(.,'${project_count} PROJECT')]
${projects}= Set Variable If '${project_count}' == 'all' All projects with ${project_count} PROJECT
Retry Wait Until Page Contains Element //clr-dg-row[contains(.,'${name}') and contains(.,'${projects}')]

2
tests/robot-cases/Group1-Nightly/Common.robot
View File

@ -766,7 +766,7 @@ Test Case - System Robot Account Cover All Projects
${name}= Create A New System Robot Account is_cover_all=${true}
Navigate To Projects
Switch To Robot Account
System Robot Account Exist ${name} All
System Robot Account Exist ${name} all
Close Browser
Test Case - System Robot Account