Commit 95c0c259 authored by Wenkai Yin's avatar Wenkai Yin Committed by Yan
Browse files

Add E2E API test cases for replication based on labels (#5639)

This commit creates a new E2E API test case to verify the replication based on label
Signed-off-by: default avatarWenkai Yin <>
parent 7521330d
......@@ -471,7 +471,7 @@ swagger_client:
rm -rf harborclient
mkdir harborclient
java -jar swagger-codegen-cli.jar generate -i docs/swagger.yaml -l python -o harborclient
python ./harborclient/ -q install --user --prefix= || true
cd harborclient; python ./ install
pip install docker -q
pip freeze
# -*- coding: utf-8 -*-
import project
import label
import registry
import replication
import repository
import swagger_client
class Harbor(project.Project, label.Label,
registry.Registry, replication.Replication,
\ No newline at end of file
# -*- coding: utf-8 -*-
import sys
import time
import swagger_client
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() = 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
return swagger_client.ProductsApi(swagger_client.ApiClient(cfg))
def _random_name(prefix):
return "%s-%d" % (prefix, time.time())
def _get_id_from_header(header):
location = header["Location"]
return location.split("/")[-1]
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"
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)
\ No newline at end of file
# -*- coding: utf-8 -*-
import sys
import base
import swagger_client
class Label(base.Base):
def create_label(self, name=base._random_name("label"), desc="",
color="", scope="g", project_id=0, **kwargs):
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)))
\ No newline at end of file
# -*- coding: utf-8 -*-
import sys
import base
import swagger_client
class Project(base.Base):
def create_project(self, name=base._random_name("project"),
metadata = {}, **kwargs):
client = self._get_client(**kwargs)
_, _, header = client.projects_post_with_http_info(
swagger_client.ProjectReq(name, metadata))
return base._get_id_from_header(header), name
\ No newline at end of file
# -*- coding: utf-8 -*-
import sys
import base
import swagger_client
class Registry(base.Base):
def create_registry(self, endpoint, name = base._random_name("registry"), username="",
password="", insecure=True, **kwargs):
client = self._get_client(**kwargs)
registry = swagger_client.RepTargetPost(name=name, endpoint=endpoint,
username=username, password=password, insecure=insecure)
_, _, header = client.targets_post_with_http_info(registry)
return base._get_id_from_header(header), name
def get_registry_id_by_endpoint(self, endpoint, **kwargs):
client = self._get_client(**kwargs)
registries = client.targets_get()
for registry in registries or []:
if registry.endpoint == endpoint:
raise Exception("registry %s not found" % endpoint)
\ No newline at end of file
# -*- coding: utf-8 -*-
import sys
import time
import base
import swagger_client
class Replication(base.Base):
def create_replication_rule(self,
projectIDList, targetIDList, name=base._random_name("rule"), desc="",
filters=[], trigger=swagger_client.RepTrigger(kind="Manual"),
projects = []
for projectID in projectIDList:
targets = []
for targetID in targetIDList:
for filter in filters:
filter["value"] = int(filter["value"])
client = self._get_client(**kwargs)
policy = swagger_client.RepPolicy(name=name, description=desc,
projects=projects, targets=targets, filters=filters,
trigger=trigger, replicate_deletion=replicate_deletion,
_, _, header = client.policies_replication_post_with_http_info(policy)
return base._get_id_from_header(header), name
def start_replication(self, rule_id, **kwargs):
client = self._get_client(**kwargs)
return client.replications_post(swagger_client.Replication(int(rule_id)))
def list_replication_jobs(self, rule_id, **kwargs):
client = self._get_client(**kwargs)
return client.jobs_replication_get(int(rule_id))
def wait_until_jobs_finish(self, rule_id, retry=10, interval=5, **kwargs):
finished = True
for i in range(retry):
finished = True
jobs = self.list_replication_jobs(rule_id, **kwargs)
for job in jobs:
if job.status != "finished":
finished = False
if not finished:
if not finished:
raise Exception("The jobs not finished")
# -*- coding: utf-8 -*-
import sys
import base
import swagger_client
class Repository(base.Base):
def list_tags(self, repository, **kwargs):
client = self._get_client(**kwargs)
return client.repositories_repo_name_tags_get(repository)
def image_exists(self, repository, tag, **kwargs):
tags = self.list_tags(repository, **kwargs)
exist = False
for t in tags:
if == tag:
exist = True
return exist
def image_should_exist(self, repository, tag, **kwargs):
if not self.image_exists(repository, tag, **kwargs):
raise Exception("image %s:%s not exist" % (repository, tag))
def image_should_not_exist(self, repository, tag, **kwargs):
if self.image_exists(repository, tag, **kwargs):
raise Exception("image %s:%s exists" % (repository, tag))
\ No newline at end of file
......@@ -152,3 +152,23 @@ Docker Login Fail
Should Not Be Equal As Integers ${rc} 0
Should Contain ${output} unauthorized: authentication required
Should Not Contain ${output} 500 Internal Server Error
Docker Login
[Arguments] ${server} ${username} ${password}
${rc} ${output}= Run And Return Rc And Output docker login -u ${username} -p ${password} ${server}
Should Be Equal As Integers ${rc} 0
Docker Pull
[Arguments] ${image}
${rc} ${output}= Run And Return Rc And Output docker pull ${image}
Should Be Equal As Integers ${rc} 0
Docker Tag
[Arguments] ${src_image} ${dst_image}
${rc} ${output}= Run And Return Rc And Output docker tag ${src_image} ${dst_image}
Should Be Equal As Integers ${rc} 0
Docker Push
[Arguments] ${image}
${rc} ${output}= Run And Return Rc And Output docker push ${image}
Should Be Equal As Integers ${rc} 0
\ No newline at end of file
......@@ -14,6 +14,7 @@
*** Settings ***
Documentation Harbor BATs
Library ../../apitests/python/library/ ${SERVER_CONFIG}
Resource ../../resources/Util.robot
Default Tags Replication
......@@ -21,6 +22,13 @@ Default Tags Replication
${HARBOR_URL} https://${ip}
${SSH_USER} root
${SERVER} ${ip}
${SERVER_URL} https://${SERVER}
&{SERVER_CONFIG} endpoint=${SERVER_API_ENDPOINT} verify_ssl=False
*** Test Cases ***
Test Case - Get Harbor Version
......@@ -186,3 +194,31 @@ Test Case - Project LeveL Replication Operation
Go Into Repo hello-world
Page Should Contain latest
Close Browser
Test Case - Replicate based on label
${project_id} ${project_name} = Create Project
Docker Pull hello-world:latest
Docker Login ${SERVER} admin Harbor12345
Docker Tag hello-world:latest ${SERVER}/${project_name}/hello-world:1.0
Docker Push ${SERVER}/${project_name}/hello-world:1.0
Docker Tag hello-world:latest ${SERVER}/${project_name}/hello-world:2.0
Docker Push ${SERVER}/${project_name}/hello-world:2.0
${label_id} ${label_name} = Create Label
Add Label To Image ${label_id} ${project_name}/hello-world 1.0
${registry_id} = Get Registry Id By Endpoint ${REMOTE_SERVER_URL}
${projects} = Create List ${project_id}
${registries} = Create List ${registry_id}
${label_filter} = Create Dictionary kind=label value=${label_id}
${filters} = Create List ${label_filter}
${rule_id} ${rule_name} = Create Replication Rule ${projects} ${registries} filters=${filters}
Start Replication ${rule_id}
Wait Until Jobs Finish ${rule_id}
Image Should Exist ${project_name}/hello-world 1.0 endpoint=${REMOTE_SERVER_API_ENDPOINT} verify_ssl=False
Image Should Not Exist ${project_name}/hello-world 2.0 endpoint=${REMOTE_SERVER_API_ENDPOINT} verify_ssl=False
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