harbor/tests/apitests/python/testutils.py

import os
import sys
import warnings
from functools import wraps

sys.path.insert(0, os.environ.get("SWAGGER_CLIENT_PATH", ''))
path=os.getcwd() + "/library"
sys.path.insert(0, path)

path=os.getcwd() + "/tests/apitests/python/library"
sys.path.insert(0, path)
path=os.getcwd() + "/tests/apitests/python/"
sys.path.insert(0, path)
print(sys.path)

files_directory = os.getcwd() + "/tests/files/"

import v2_swagger_client
import swagger_client.models

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)
USER_ROLE=dict(admin=0,normal=1)
TEARDOWN = os.environ.get('TEARDOWN', 'true').lower() in ('true', 'yes')
notary_url = os.environ.get('NOTARY_URL', 'https://'+harbor_server+':4443')
DOCKER_USER = os.environ.get('DOCKER_USER', '')
DOCKER_PWD = os.environ.get('DOCKER_PWD', '')
METRIC_URL = os.environ.get('METRIC_URL', 'http://'+harbor_server+':9090')
SYSLOG_ENDPOINT = os.environ.get('SYSLOG_ENDPOINT', '')
ES_ENDPOINT = os.environ.get('ES_ENDPOINT', '')
BASE_IMAGE = dict(name='busybox', tag='latest')
BASE_IMAGE_ABS_PATH_NAME = '/' + BASE_IMAGE['name'] + '.tar'

def GetProductApi(username, password, harbor_server= os.environ.get("HARBOR_HOST", '')):

    cfg = swagger_client.Configuration()
    cfg.host = "https://"+harbor_server+"/api/v2.0"
    cfg.username = username
    cfg.password = password
    cfg.verify_ssl = False
    cfg.debug = True
    api_client = swagger_client.ApiClient(cfg)
    api_instance = swagger_client.ProductsApi(api_client)
    return api_instance

def GetRepositoryApi(username, password, harbor_server= os.environ.get("HARBOR_HOST", '')):

    cfg = v2_swagger_client.Configuration()
    cfg.host = "https://"+harbor_server+"/api/v2.0"
    cfg.username = username
    cfg.password = password
    cfg.verify_ssl = False
    cfg.debug = True
    api_client = v2_swagger_client.ApiClient(cfg)
    api_instance = v2_swagger_client.RepositoryApi(api_client)
    return api_instance

def GetUserGroupApi(username, password, harbor_server= os.environ.get("HARBOR_HOST", '')):
    cfg = v2_swagger_client.Configuration()
    cfg.host = "https://"+harbor_server+"/api/v2.0"
    cfg.username = username
    cfg.password = password
    cfg.verify_ssl = False
    cfg.debug = True
    api_client = v2_swagger_client.ApiClient(cfg)
    api_instance = v2_swagger_client.UsergroupApi(api_client)
    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))

def suppress_urllib3_warning(func):
    @wraps(func)
    def inner_func(*args):
        warnings.filterwarnings(action="ignore",
                                message="unclosed",
                                category=ResourceWarning)
        warnings.filterwarnings(action='ignore', message='Unverified HTTPS request')
        func(*args)
    return inner_func

from contextlib import contextmanager

@contextmanager
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 is_teardown:
            api.delete_user(user_id, **ADMIN_CLIENT)

@contextmanager
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)

    try:
        yield (project_id, project_name)
    finally:
        if is_teardown:
            api.delete_project(project_id, **ADMIN_CLIENT)