mirror of
https://github.com/goharbor/harbor.git
synced 2024-12-04 08:03:37 +01:00
acc027afd7
1. Docker v20 is the default version for git action, dns should be set manually. Python v3.8 is the default version for git action, so remove deployment for python v3.6. 2. Some of API tests would be affect by docker v20, like manifest get message will not be triggered if there is one locally, to avoid this impact, API tests will be tested with docker E2E image, in this image, docker v19 is enabled. Signed-off-by: danfengliu <danfengl@vmware.com>
112 lines
4.1 KiB
Python
112 lines
4.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import sys
|
|
import time
|
|
import subprocess
|
|
import swagger_client
|
|
try:
|
|
from urllib import getproxies
|
|
except ImportError:
|
|
from urllib.request import getproxies
|
|
|
|
class Server:
|
|
def __init__(self, endpoint, verify_ssl):
|
|
self.endpoint = endpoint
|
|
self.verify_ssl = verify_ssl
|
|
|
|
class Credential:
|
|
def __init__(self, type, username, password):
|
|
self.type = type
|
|
self.username = username
|
|
self.password = password
|
|
|
|
def _create_client(server, credential, debug):
|
|
cfg = swagger_client.Configuration()
|
|
cfg.host = server.endpoint
|
|
cfg.verify_ssl = server.verify_ssl
|
|
# support basic auth only for now
|
|
cfg.username = credential.username
|
|
cfg.password = credential.password
|
|
cfg.debug = debug
|
|
|
|
proxies = getproxies()
|
|
proxy = proxies.get('http', proxies.get('all', None))
|
|
if proxy:
|
|
cfg.proxy = proxy
|
|
|
|
return swagger_client.ProductsApi(swagger_client.ApiClient(cfg))
|
|
|
|
def _assert_status_code(expect_code, return_code):
|
|
if str(return_code) != str(expect_code):
|
|
raise Exception(r"HTTPS status code 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)))
|
|
|
|
def _get_id_from_header(header):
|
|
try:
|
|
location = header["Location"]
|
|
return int(location.split("/")[-1])
|
|
except Exception:
|
|
return None
|
|
|
|
def _get_string_from_unicode(udata):
|
|
result=''
|
|
for u in udata:
|
|
tmp = u.encode('utf8')
|
|
result = result + tmp.strip('\n\r\t')
|
|
return result
|
|
|
|
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:
|
|
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
|
|
|
|
class Base:
|
|
def __init__(self,
|
|
server = Server(endpoint="http://localhost:8080/api", verify_ssl=False),
|
|
credential = Credential(type="basic_auth", username="admin", password="Harbor12345"),
|
|
debug = True):
|
|
if not isinstance(server.verify_ssl, bool):
|
|
server.verify_ssl = server.verify_ssl == "True"
|
|
self.server = server
|
|
self.credential = credential
|
|
self.debug = debug
|
|
self.client = _create_client(server, credential, debug)
|
|
|
|
def _get_client(self, **kwargs):
|
|
if len(kwargs) == 0:
|
|
return self.client
|
|
server = self.server
|
|
if "endpoint" in kwargs:
|
|
server.endpoint = kwargs.get("endpoint")
|
|
if "verify_ssl" in kwargs:
|
|
if not isinstance(kwargs.get("verify_ssl"), bool):
|
|
server.verify_ssl = kwargs.get("verify_ssl") == "True"
|
|
else:
|
|
server.verify_ssl = kwargs.get("verify_ssl")
|
|
credential = self.credential
|
|
if "type" in kwargs:
|
|
credential.type = kwargs.get("type")
|
|
if "username" in kwargs:
|
|
credential.username = kwargs.get("username")
|
|
if "password" in kwargs:
|
|
credential.password = kwargs.get("password")
|
|
return _create_client(server, credential, self.debug) |