Unverified Commit 04fd4d44 authored by danfengliu's avatar danfengliu Committed by GitHub
Browse files

Add API test case for user view logs. (#6370)



Add API test case for user view logs.
Signed-off-by: default avatardanfengliu <danfengl@vmware.com>
parent 68b1b98f
......@@ -27,7 +27,11 @@ def _create_client(server, credential, debug):
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))
raise Exception(r"HTTPS status code s not as we expected. Expected {}, while actual HTTPS status code is {}.".format(expect_code, return_code))
def _assert_status_body(expect_status_body, returned_status_body):
if expect_status_body.strip() != returned_status_body.strip():
raise Exception(r"HTTPS status body s not as we expected. Expected {}, while actual HTTPS status body is {}.".format(expect_status_body, returned_status_body))
def _random_name(prefix):
return "%s-%d" % (prefix, int(round(time.time() * 1000)))
......
......@@ -2,25 +2,57 @@
import base
import swagger_client
from swagger_client.rest import ApiException
def set_configurations(client, expect_status_code = 200, expect_response_body = None, **config):
conf = swagger_client.Configurations()
if "project_creation_restriction" in config:
conf.project_creation_restriction = config.get("project_creation_restriction")
if "token_expiration" in config:
conf.token_expiration = config.get("token_expiration")
try:
_, status_code, _ = client.configurations_put_with_http_info(conf)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
if expect_response_body is not None:
base._assert_status_body(expect_response_body, e.body)
return
base._assert_status_code(expect_status_code, status_code)
class Configurations(base.Base):
def get_configurations(self, item_name = None, expect_status_code = 200, **kwargs):
def get_configurations(self, item_name = None, expect_status_code = 200, expect_response_body = None, **kwargs):
client = self._get_client(**kwargs)
data, status_code, _ = client.configurations_get_with_http_info()
try:
data, status_code, _ = client.configurations_get_with_http_info()
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
if expect_response_body is not None:
base._assert_status_body(expect_response_body, e.body)
return
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(200, status_code)
if item_name is not None:
return {
'project_creation_restriction': data.project_creation_restriction.value,
}.get(item_name,'error')
'token_expiration': data.token_expiration.value,
}.get(item_name,'Get Configutation Error: Item name {} is not exist'.format(item_name))
return data
def set_configurations_of_project_creation_restriction_success(self, project_creation_restriction, expect_status_code = 200, **kwargs):
def set_configurations_of_project_creation_restriction(self, project_creation_restriction, **kwargs):
client = self._get_client(**kwargs)
config=dict(project_creation_restriction=project_creation_restriction)
set_configurations(client, **config)
def set_configurations_of_token_expiration(self, token_expiration, **kwargs):
client = self._get_client(**kwargs)
conf = swagger_client.Configurations(project_creation_restriction = project_creation_restriction)
_, status_code, _ = client.configurations_put_with_http_info(conf)
base._assert_status_code(200, status_code)
item_value = self.get_configurations(item_name = "project_creation_restriction", **kwargs)
if item_value != project_creation_restriction:
raise Exception("Failed to set system configuration item {} to value {},\
actual value is {}".format("project_creation_restriction", project_creation_restriction, item_value))
config=dict(token_expiration=token_expiration)
set_configurations(client, **config)
......@@ -26,17 +26,15 @@ class Project(base.Base):
try:
_, status_code, header = client.projects_post_with_http_info(swagger_client.ProjectReq(name, metadata))
except ApiException as e:
if e.status == expect_status_code:
if expect_response_body is not None and e.body.strip() != expect_response_body.strip():
raise Exception(r"Create project result body is not as expected {} actual status is {}.".format(expect_response_body.strip(), e.body.strip()))
else:
return e.reason, e.body
else:
raise Exception(r"Create project result is not as expected {} actual status is {}.".format(expect_status_code, e.status))
base._assert_status_code(expect_status_code, e.status)
if expect_response_body is not None:
base._assert_status_body(expect_response_body, e.body)
return
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 get_projects(self, params, **kwargs):
client = self._get_client(**kwargs)
data = []
......@@ -60,16 +58,18 @@ class Project(base.Base):
404: False,
}.get(status_code,'error')
def get_project(self, project_id, expect_status_code = 200, **kwargs):
def get_project(self, project_id, expect_status_code = 200, expect_response_body = None, **kwargs):
client = self._get_client(**kwargs)
try:
data, status_code, _ = client.projects_project_id_get_with_http_info(project_id)
except ApiException as e:
if e.status == expect_status_code:
return e.reason, e.body
else:
raise Exception(r"Get project result is not as expected {} actual status is {}.".format(expect_status_code, e.status))
base._assert_status_code(expect_status_code, e.status)
if expect_response_body is not None:
base._assert_status_body(expect_response_body, e.body)
return
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(200, status_code)
return data
def update_project(self, project_id, metadata, **kwargs):
......@@ -83,11 +83,11 @@ class Project(base.Base):
_, status_code, _ = client.projects_project_id_delete_with_http_info(project_id)
base._assert_status_code(expect_status_code, status_code)
def get_project_metadata_by_name(self, project_id, meta_name, **kwargs):
def get_project_metadata_by_name(self, project_id, meta_name, expect_status_code = 200, **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)
base._assert_status_code(expect_status_code, status_code)
return {
'public': ProjectMetadata.public,
'enable_content_trust': ProjectMetadata.enable_content_trust,
......@@ -96,21 +96,40 @@ class Project(base.Base):
'severity': ProjectMetadata.severity,
}.get(meta_name,'error')
def get_project_log(self, project_id, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
body, status_code, _ = client.projects_project_id_logs_get_with_http_info(project_id)
base._assert_status_code(expect_status_code, status_code)
return body
def filter_project_logs(self, project_id, operator, repository, tag, operation_type, **kwargs):
access_logs = self.get_project_log(project_id, **kwargs)
count = 0
for each_access_log in list(access_logs):
if each_access_log.username == operator and \
each_access_log.repo_name.strip(r'/') == repository and \
each_access_log.repo_tag == tag and \
each_access_log.operation == operation_type:
count = count + 1
return count
def get_project_members(self, project_id, **kwargs):
client = self._get_client(**kwargs)
return client.projects_project_id_members_get(project_id)
def get_project_member(self, project_id, member_id, expect_status_code = 200, **kwargs):
def get_project_member(self, project_id, member_id, expect_status_code = 200, expect_response_body = None, **kwargs):
client = self._get_client(**kwargs)
data = []
try:
data, status_code, _ = client.projects_project_id_members_mid_get_with_http_info(project_id, member_id,)
except ApiException as e:
if e.status == expect_status_code:
return e.reason, e.body
else:
raise Exception(r"Get project member result is not as expected {} actual status is {}.".format(expect_status_code, e.status))
base._assert_status_code(expect_status_code, e.status)
if expect_response_body is not None:
base._assert_status_body(expect_response_body, e.body)
return
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(200, status_code)
return data
def check_project_member_not_exist(self, project_id, member_user_name, **kwargs):
......
......@@ -132,4 +132,10 @@ class Repository(base.Base):
if is_repo_exist_in_project(repositories, repo_name) == False:
raise Exception("Repository {} is not exist.".format(repo_name))
def signature_should_exist(self, repo_name, tag, **kwargs):
signatures = self.get_repo_signatures(repo_name, **kwargs)
for each_sign in signatures:
if each_sign.tag == tag and len(each_sign.hashes["sha256"]) == 44:
print "sha256:", len(each_sign.hashes["sha256"])
return
raise Exception(r"Signature of {}:{} is not exist!".format(repo_name, tag))
\ No newline at end of file
......@@ -26,14 +26,6 @@ class User(base.Base):
return base._get_id_from_header(header), name
def create_user_success(self, name=None,
email = None, user_password=None, realname = None, role_id = None, **kwargs):
user_id, user_name = self.create_user(name, email, user_password, realname, role_id, **kwargs)
if user_id != None:
return user_id, user_name
else:
raise Exception("user id is not exist, please contact developer to solve this problem")
def get_users(self, username=None, email=None, page=None, page_size=None, **kwargs):
client = self._get_client(**kwargs)
params={}
......
......@@ -63,7 +63,7 @@ class TestProjects(unittest.TestCase):
user_add_rule_password = "Aa123456"
#1. Create user(UA)
TestProjects.user_add_rule_id, user_add_rule_name = self.user.create_user_success(user_password = user_add_rule_password, **ADMIN_CLIENT)
TestProjects.user_add_rule_id, user_add_rule_name = self.user.create_user(user_password = user_add_rule_password, **ADMIN_CLIENT)
TestProjects.USER_add_rule_CLIENT=dict(endpoint = url, username = user_add_rule_name, password = user_add_rule_password)
......
......@@ -66,7 +66,7 @@ class TestProjects(unittest.TestCase):
user_001_password = "Aa123456"
#1. Create user-001
TestProjects.user_add_g_lbl_id, user_add_g_lbl_name = self.user.create_user_success(user_password = user_001_password, **ADMIN_CLIENT)
TestProjects.user_add_g_lbl_id, user_add_g_lbl_name = self.user.create_user(user_password = user_001_password, **ADMIN_CLIENT)
TestProjects.USER_add_g_lbl_CLIENT=dict(endpoint = url, username = user_add_g_lbl_name, password = user_001_password)
......
......@@ -57,7 +57,7 @@ class TestProjects(unittest.TestCase):
user_del_repo_password = "Aa123456"
#1. Create a new user(UA);
TestProjects.user_del_repo_id, user_del_repo_name = self.user.create_user_success(user_password = user_del_repo_password, **ADMIN_CLIENT)
TestProjects.user_del_repo_id, user_del_repo_name = self.user.create_user(user_password = user_del_repo_password, **ADMIN_CLIENT)
TestProjects.USER_del_repo_CLIENT=dict(endpoint = url, username = user_del_repo_name, password = user_del_repo_password)
......
......@@ -50,19 +50,19 @@ class TestProjects(unittest.TestCase):
user_edit_project_creation_password = "Aa123456"
#1. Create a new user(UA);
TestProjects.user_edit_project_creation_id, user_edit_project_creation_name = self.user.create_user_success(user_password = user_edit_project_creation_password, **ADMIN_CLIENT)
TestProjects.user_edit_project_creation_id, user_edit_project_creation_name = self.user.create_user(user_password = user_edit_project_creation_password, **ADMIN_CLIENT)
TestProjects.USER_edit_project_creation_CLIENT=dict(endpoint = url, username = user_edit_project_creation_name, password = user_edit_project_creation_password)
#2. Set project creation to "admin only";
self.conf.set_configurations_of_project_creation_restriction_success("adminonly", **ADMIN_CLIENT)
self.conf.set_configurations_of_project_creation_restriction("adminonly", **ADMIN_CLIENT)
#3. Create a new project(PA) by user(UA), and fail to create a new project;
self.project.create_project(metadata = {"public": "false"}, expect_status_code = 403,
expect_response_body = "Only system admin can create project", **TestProjects.USER_edit_project_creation_CLIENT)
#4. Set project creation to "everyone";
self.conf.set_configurations_of_project_creation_restriction_success("everyone", **ADMIN_CLIENT)
self.conf.set_configurations_of_project_creation_restriction("everyone", **ADMIN_CLIENT)
#5. Create a new project(PA) by user(UA), success to create a project.
TestProjects.project_edit_project_creation_id, _ = self.project.create_project(metadata = {"public": "false"}, **TestProjects.USER_edit_project_creation_CLIENT)
......
......@@ -62,7 +62,7 @@ class TestProjects(unittest.TestCase):
user_gc_password = "Aa123456"
#1. Create a new user(UA);
TestProjects.user_gc_id, user_gc_name = self.user.create_user_success(user_password = user_gc_password, **ADMIN_CLIENT)
TestProjects.user_gc_id, user_gc_name = self.user.create_user(user_password = user_gc_password, **ADMIN_CLIENT)
TestProjects.USER_GC_CLIENT=dict(endpoint = url, username = user_gc_name, password = user_gc_password)
......
......@@ -63,7 +63,7 @@ class TestProjects(unittest.TestCase):
user_content_trust_password = "Aa123456"
#1. Create a new user(UA);
TestProjects.user_content_trust_id, user_content_trust_name = self.user.create_user_success(user_password = user_content_trust_password, **ADMIN_CLIENT)
TestProjects.user_content_trust_id, user_content_trust_name = self.user.create_user(user_password = user_content_trust_password, **ADMIN_CLIENT)
TestProjects.USER_CONTENT_TRUST_CLIENT=dict(endpoint = url, username = user_content_trust_name, password = user_content_trust_password)
......
......@@ -56,7 +56,7 @@ class TestProjects(unittest.TestCase):
user_001_password = "Aa123456"
#1. Create user-001
TestProjects.user_scan_image_id, user_scan_image_name = self.user.create_user_success(user_password = user_001_password, **ADMIN_CLIENT)
TestProjects.user_scan_image_id, user_scan_image_name = self.user.create_user(user_password = user_001_password, **ADMIN_CLIENT)
TestProjects.USER_scan_image_CLIENT=dict(endpoint = url, username = user_scan_image_name, password = user_001_password)
......
from __future__ import absolute_import
import unittest
from testutils import ADMIN_CLIENT
from testutils import TEARDOWN
from testutils import TestResult
from library.user import User
from library.project import Project
from library.repository import Repository
from library.repository import push_image_to_project
from testutils import harbor_server
class TestProjects(unittest.TestCase):
@classmethod
def setUp(self):
test_result = TestResult()
self.test_result= test_result
project = Project()
self.project= project
user = User()
self.user= user
repo = Repository()
self.repo= repo
@classmethod
def tearDown(self):
self.test_result.get_final_result()
print "Case completed"
@unittest.skipIf(TEARDOWN == False, "Test data won't be erased.")
def test_ClearData(self):
#1. Delete project(PA);
self.project.delete_project(TestProjects.project_user_view_logs_id, **TestProjects.USER_USER_VIEW_LOGS_CLIENT)
#2. Delete user(UA);
self.user.delete_user(TestProjects.user_user_view_logs_id, **ADMIN_CLIENT)
def testUserViewLogs(self):
"""
Test case:
User View Logs
Test step and expected result:
1. Create a new user(UA);
2. Create a new project(PA) by user(UA), in project(PA), there should be 1 'create' log record;;
3. Push a new image(IA) in project(PA) by admin, in project(PA), there should be 1 'push' log record;;
4. Delete repository(RA) by user(UA), in project(PA), there should be 1 'delete' log record;;
Tear down:
1. Delete project(PA);
2. Delete user(UA).
"""
url = ADMIN_CLIENT["endpoint"]
admin_name = ADMIN_CLIENT["username"]
admin_password = ADMIN_CLIENT["password"]
user_content_trust_password = "Aa123456"
#1. Create a new user(UA);
TestProjects.user_user_view_logs_id, user_user_view_logs_name = self.user.create_user(user_password = user_content_trust_password, **ADMIN_CLIENT)
TestProjects.USER_USER_VIEW_LOGS_CLIENT=dict(endpoint = url, username = user_user_view_logs_name, password = user_content_trust_password)
#2.1 Create a new project(PA) by user(UA);
TestProjects.project_user_view_logs_id, project_user_view_logs_name = self.project.create_project(metadata = {"public": "false"}, **TestProjects.USER_USER_VIEW_LOGS_CLIENT)
#2.2 In project(PA), there should be 1 'create' log record;
tag = "N/A"
operation = "create"
log_count = self.project.filter_project_logs(TestProjects.project_user_view_logs_id, user_user_view_logs_name, project_user_view_logs_name, tag, operation, **TestProjects.USER_USER_VIEW_LOGS_CLIENT)
if log_count != 1:
self.test_result.add_test_result("Failed to get log with user:{}, repository:{}, tag:{} and operation:{}".format(user_user_view_logs_name, project_user_view_logs_name, tag, operation))
#3.1 Push a new image(IA) in project(PA) by admin;
repo_name, tag = push_image_to_project(project_user_view_logs_name, harbor_server, admin_name, admin_password, "tomcat", "latest")
#3.2 In project(PA), there should be 1 'push' log record;
operation = "push"
log_count = self.project.filter_project_logs(TestProjects.project_user_view_logs_id, admin_name, repo_name, tag, "push", **TestProjects.USER_USER_VIEW_LOGS_CLIENT)
if log_count != 1:
self.test_result.add_test_result("Failed to get log with user:{}, repository:{}, tag:{} and operation:{}".format(user_user_view_logs_name, project_user_view_logs_name, tag, operation))
#4.1 Delete repository(RA) by user(UA);
self.repo.delete_repoitory(repo_name, **TestProjects.USER_USER_VIEW_LOGS_CLIENT)
#4.2 In project(PA), there should be 1 'delete' log record;
operation = "delete"
log_count = self.project.filter_project_logs(TestProjects.project_user_view_logs_id, user_user_view_logs_name, repo_name, tag, "delete", **TestProjects.USER_USER_VIEW_LOGS_CLIENT)
if log_count != 1:
self.test_result.add_test_result("Failed to get log with user:{}, repository:{}, tag:{} and operation:{}".format(user_user_view_logs_name, project_user_view_logs_name, tag, operation))
if __name__ == '__main__':
unittest.main()
\ No newline at end of file
......@@ -26,4 +26,16 @@ 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
\ No newline at end of file
return api_instance
class TestResult(object):
def __init__(self):
self.num_errors = 0
self.error_message = []
def add_test_result(self, error_message):
self.num_errors = self.num_errors + 1
self.error_message.append(error_message)
def get_final_result(self):
if self.num_errors > 0:
for each_err_msg in self.error_message:
print "Error message:", each_err_msg
raise Exception(r"Test case failed with {} errors.".format(self.num_errors))
......@@ -34,4 +34,6 @@ Test Case - Manage Project Member
Test Case - Project Level Policy Content Trust
Harbor API Test ./tests/apitests/python/test_project_level_policy_content_trust.py
Test Case - Garbage Collection
Harbor API Test ./tests/apitests/python/test_garbage_collection.py
\ No newline at end of file
Harbor API Test ./tests/apitests/python/test_garbage_collection.py
Test Case - User View Logs
Harbor API Test ./tests/apitests/python/test_user_view_logs.py
\ No newline at end of file
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