Upstream/harbor
1
0
Fork 0
mirror of https://github.com/goharbor/harbor.git synced 2025-02-16 20:01:35 +01:00
Code Issues Projects Releases Wiki Activity

Merge pull request #13781 from danfengliu/add-py-test-for-system-robot-account

Browse Source 
Add py test for system robot account
This commit is contained in:
danfengliu 2021-01-08 14:03:50 +08:00 committed by GitHub
commit b43b59cc8e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
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