Upstream/harbor
1
0
Fork 0
mirror of https://github.com/goharbor/harbor.git synced 2024-09-28 13:27:31 +02:00
Code Issues Projects Releases Wiki Activity

Since harbor 1.8 released, API test for replication was not update yet, so in this PR, replicaiton test will be adapt for harbor 1.8.

Browse Source 
Signed-off-by: danfengliu <danfengl@vmware.com>
This commit is contained in:
danfengliu 2019-08-08 16:16:02 +08:00
parent 9cbcc93e8a
commit 1b0931c034
4 changed files with 60 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
tests/apitests/python/library/registry.py
View File

@ -5,15 +5,19 @@ import base
import swagger_client
class Registry(base.Base):
def create_registry(self, endpoint, name=None, username="",
password="", insecure=True, **kwargs):
if name is None:
name = base._random_name("registry")
def create_registry(self, url, registry_type= "harbor", description="", credentialType = "basic",
access_key = "admin", access_secret = "Harbor12345", name=base._random_name("registry"),
insecure=True, expect_status_code = 201, **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
registryCredential = swagger_client.RegistryCredential(type=credentialType, access_key=access_key, access_secret=access_secret)
registry = swagger_client.Registry(name=name, url=url,
description= description, type=registry_type,
insecure=insecure, credential=registryCredential)
_, status_code, header = client.registries_post_with_http_info(registry)
base._assert_status_code(expect_status_code, status_code)
return base._get_id_from_header(header), _
def get_registry_id_by_endpoint(self, endpoint, **kwargs):
client = self._get_client(**kwargs)
@ -21,4 +25,9 @@ class Registry(base.Base):
for registry in registries or []:
if registry.endpoint == endpoint:
return registry.id
raise Exception("registry %s not found" % endpoint)
raise Exception("registry %s not found" % endpoint)
def delete_registry(self, registry_id, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
_, status_code, _ = client.registries_id_delete_with_http_info(registry_id)
base._assert_status_code(expect_status_code, status_code)

32
tests/apitests/python/library/replication.py
View File

@ -6,15 +6,31 @@ import base
import swagger_client
class Replication(base.Base):
def create_replication_policy(self, dest_registry=None, src_registry=None, name=None, description="",
dest_namespace = "", filters=None, trigger=swagger_client.ReplicationTrigger(type="manual",trigger_settings=swagger_client.TriggerSettings(cron="")),
deletion=False, override=True, enabled=True, expect_status_code = 201, **kwargs):
if name is None:
name = base._random_name("rule")
if filters is None:
filters = []
for policy_filter in filters:
policy_filter["value"] = int(policy_filter["value"])
client = self._get_client(**kwargs)
policy = swagger_client.ReplicationPolicy(name=name, description=description,dest_namespace=dest_namespace,
dest_registry=dest_registry, src_registry=src_registry,filters=filters,
trigger=trigger, deletion=deletion, override=override, enabled=enabled)
_, status_code, header = client.replication_policies_post_with_http_info(policy)
base._assert_status_code(expect_status_code, status_code)
return base._get_id_from_header(header), name
def get_replication_rule(self, param = None, rule_id = None, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
if rule_id is None:
if param is None:
param = dict()
data, status_code, _ = client.policies_replication_get_with_http_info(param)
data, status_code, _ = client.replication_policies_id_get_with_http_info(param)
else:
data, status_code, _ = client.policies_replication_id_get_with_http_info(rule_id)
data, status_code, _ = client.replication_policies_id_get_with_http_info(rule_id)
base._assert_status_code(expect_status_code, status_code)
return data
@ -24,11 +40,11 @@ class Replication(base.Base):
raise Exception(r"Check replication rule failed, expect <{}> actual <{}>.".format(expect_rule_name, str(rule_data.name)))
else:
print r"Check Replication rule passed, rule name <{}>.".format(str(rule_data.name))
get_trigger = str(rule_data.trigger.kind)
if expect_trigger is not None and get_trigger == str(expect_trigger):
print r"Check Replication rule trigger passed, trigger name <{}>.".format(get_trigger)
else:
raise Exception(r"Check replication rule trigger failed, expect <{}> actual <{}>.".format(expect_trigger, get_trigger))
#get_trigger = str(rule_data.trigger.kind)
#if expect_trigger is not None and get_trigger == str(expect_trigger):
# print r"Check Replication rule trigger passed, trigger name <{}>.".format(get_trigger)
#else:
# raise Exception(r"Check replication rule trigger failed, expect <{}> actual <{}>.".format(expect_trigger, get_trigger))
def start_replication(self, rule_id, **kwargs):
@ -55,5 +71,5 @@ class Replication(base.Base):
def delete_replication_rule(self, rule_id, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
_, status_code, _ = client.policies_replication_id_delete_with_http_info(rule_id)
_, status_code, _ = client.replication_policies_id_delete_with_http_info(rule_id)
base._assert_status_code(expect_status_code, status_code)

39
tests/apitests/python/library/target.py
View File

@ -1,39 +0,0 @@
# -*- coding: utf-8 -*-
import time
import base
import swagger_client
class Target(base.Base):
def create_target(self,
endpoint_target = None,
username_target = "target_user", password_target = "Aa123456", name_target=base._random_name("target"),
target_type=0, insecure_target=True, expect_status_code = 201,
**kwargs):
if endpoint_target is None:
endpoint_target = r"https://{}.{}.{}.{}".format(int(round(time.time() * 1000)) % 100,
int(round(time.time() * 1000)) % 200,
int(round(time.time() * 1000)) % 100,
int(round(time.time() * 1000)) % 254)
client = self._get_client(**kwargs)
policy = swagger_client.RepTarget(name=name_target, endpoint=endpoint_target,
username=username_target, password=password_target, type=target_type,
insecure=insecure_target)
_, status_code, header = client.targets_post_with_http_info(policy)
base._assert_status_code(expect_status_code, status_code)
return base._get_id_from_header(header), name_target
def get_target(self, expect_status_code = 200, params = None, **kwargs):
client = self._get_client(**kwargs)
data = []
if params is None:
params = {}
data, status_code, _ = client.targets_get_with_http_info(**params)
base._assert_status_code(expect_status_code, status_code)
return data
def delete_target(self, target_id, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
_, status_code, _ = client.targets_id_delete_with_http_info(target_id)
base._assert_status_code(expect_status_code, status_code)

42
tests/apitests/python/test_add_replication_rule.py
View File

@ -2,11 +2,12 @@ from __future__ import absolute_import
import unittest
from testutils import ADMIN_CLIENT
from testutils import harbor_server
from testutils import TEARDOWN
from library.project import Project
from library.user import User
from library.replication import Replication
from library.target import Target
from library.registry import Registry
import swagger_client
class TestProjects(unittest.TestCase):
@ -21,8 +22,8 @@ class TestProjects(unittest.TestCase):
replication = Replication()
self.replication= replication
target = Target()
self.target= target
registry = Registry()
self.registry= registry
@classmethod
def tearDown(self):
@ -31,11 +32,10 @@ class TestProjects(unittest.TestCase):
@unittest.skipIf(TEARDOWN == False, "Test data won't be erased.")
def test_ClearData(self):
#1. Delete rule(RA);
for rule_id in TestProjects.rule_id_list:
self.replication.delete_replication_rule(rule_id, **ADMIN_CLIENT)
self.replication.delete_replication_rule(TestProjects.rule_id, **ADMIN_CLIENT)
#2. Delete target(TA);
self.target.delete_target(TestProjects.target_id, **ADMIN_CLIENT)
#2. Delete registry(TA);
self.registry.delete_registry(TestProjects.registry_id, **ADMIN_CLIENT)
#3. Delete project(PA);
self.project.delete_project(TestProjects.project_add_rule_id, **TestProjects.USER_add_rule_CLIENT)
@ -50,12 +50,12 @@ class TestProjects(unittest.TestCase):
Test step and expected result:
1. Create a new user(UA);
2. Create a new private project(PA) by user(UA);
3. Create a new target(TA)/registry;
4. Create a new rule for project(PA) and target(TA);
5. Check if rule is exist.
3. Create a new registry;
4. Create a new rule for this registry;
5. Check rule should be exist.
Tear down:
1. Delete rule(RA);
2. Delete targe(TA);
2. Delete registry(TA);
3. Delete project(PA);
4. Delete user(UA).
"""
@ -74,21 +74,15 @@ class TestProjects(unittest.TestCase):
self.project.projects_should_exist(dict(public=False), expected_count = 1,
expected_project_id = TestProjects.project_add_rule_id, **TestProjects.USER_add_rule_CLIENT)
#3. Create a new target(TA)/registry
TestProjects.target_id, _ = self.target.create_target(**ADMIN_CLIENT)
print "TestProjects.target_id:", TestProjects.target_id
#3. Create a new registry
TestProjects.registry_id, _ = self.registry.create_registry("https://" + harbor_server,**ADMIN_CLIENT)
print "TestProjects.registry_id:", TestProjects.registry_id
TestProjects.rule_id_list = []
#4. Create a new rule for this registry;
TestProjects.rule_id, rule_name = self.replication.create_replication_policy(dest_registry=swagger_client.Registry(id=int(TestProjects.registry_id)), **ADMIN_CLIENT)
trigger_values_to_set = ["Manual", "Immediate"]
for value in trigger_values_to_set:
#4. Create a new rule for project(PA) and target(TA)
rule_id, rule_name = self.replication.create_replication_rule([TestProjects.project_add_rule_id],
[TestProjects.target_id], trigger=swagger_client.RepTrigger(kind=value), **ADMIN_CLIENT)
TestProjects.rule_id_list.append(rule_id)
#5. Check rule should be exist
self.replication.check_replication_rule_should_exist(rule_id, rule_name, expect_trigger = value, **ADMIN_CLIENT)
#5. Check rule should be exist
self.replication.check_replication_rule_should_exist(TestProjects.rule_id, rule_name, **ADMIN_CLIENT)
if __name__ == '__main__':