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

Merge pull request #8608 from danfengliu/replicaiton-ng-api-adapt-1

Browse Source 
API test for replication should be back to travis
This commit is contained in:
Wang Yan 2019-08-09 11:10:17 +08:00 committed by GitHub
commit 20cb8e514f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
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 import swagger_client
class Registry(base.Base): class Registry(base.Base):
def create_registry(self, endpoint, name=None, username="", def create_registry(self, url, registry_type= "harbor", description="", credentialType = "basic",
password="", insecure=True, **kwargs): access_key = "admin", access_secret = "Harbor12345", name=base._random_name("registry"),
if name is None: insecure=True, expect_status_code = 201, **kwargs):
name = base._random_name("registry")
client = self._get_client(**kwargs) client = self._get_client(**kwargs)
registry = swagger_client.RepTargetPost(name=name, endpoint=endpoint, registryCredential = swagger_client.RegistryCredential(type=credentialType, access_key=access_key, access_secret=access_secret)
username=username, password=password, insecure=insecure) registry = swagger_client.Registry(name=name, url=url,
_, _, header = client.targets_post_with_http_info(registry) description= description, type=registry_type,
return base._get_id_from_header(header), name 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): def get_registry_id_by_endpoint(self, endpoint, **kwargs):
client = self._get_client(**kwargs) client = self._get_client(**kwargs)
@ -21,4 +25,9 @@ class Registry(base.Base):
for registry in registries or []: for registry in registries or []:
if registry.endpoint == endpoint: if registry.endpoint == endpoint:
return registry.id 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 import swagger_client
class Replication(base.Base): 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): def get_replication_rule(self, param = None, rule_id = None, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs) client = self._get_client(**kwargs)
if rule_id is None: if rule_id is None:
if param is None: if param is None:
param = dict() 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: 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) base._assert_status_code(expect_status_code, status_code)
return data 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))) raise Exception(r"Check replication rule failed, expect <{}> actual <{}>.".format(expect_rule_name, str(rule_data.name)))
else: else:
print r"Check Replication rule passed, rule name <{}>.".format(str(rule_data.name)) print r"Check Replication rule passed, rule name <{}>.".format(str(rule_data.name))
get_trigger = str(rule_data.trigger.kind) #get_trigger = str(rule_data.trigger.kind)
if expect_trigger is not None and get_trigger == str(expect_trigger): #if expect_trigger is not None and get_trigger == str(expect_trigger):
print r"Check Replication rule trigger passed, trigger name <{}>.".format(get_trigger) # print r"Check Replication rule trigger passed, trigger name <{}>.".format(get_trigger)
else: #else:
raise Exception(r"Check replication rule trigger failed, expect <{}> actual <{}>.".format(expect_trigger, get_trigger)) # raise Exception(r"Check replication rule trigger failed, expect <{}> actual <{}>.".format(expect_trigger, get_trigger))
def start_replication(self, rule_id, **kwargs): 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): def delete_replication_rule(self, rule_id, expect_status_code = 200, **kwargs):
client = self._get_client(**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) 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 import unittest
from testutils import ADMIN_CLIENT from testutils import ADMIN_CLIENT
from testutils import harbor_server
from testutils import TEARDOWN from testutils import TEARDOWN
from library.project import Project from library.project import Project
from library.user import User from library.user import User
from library.replication import Replication from library.replication import Replication
from library.target import Target from library.registry import Registry
import swagger_client import swagger_client
class TestProjects(unittest.TestCase): class TestProjects(unittest.TestCase):
@ -21,8 +22,8 @@ class TestProjects(unittest.TestCase):
replication = Replication() replication = Replication()
self.replication= replication self.replication= replication
target = Target() registry = Registry()
self.target= target self.registry= registry
@classmethod @classmethod
def tearDown(self): def tearDown(self):
@ -31,11 +32,10 @@ class TestProjects(unittest.TestCase):
@unittest.skipIf(TEARDOWN == False, "Test data won't be erased.") @unittest.skipIf(TEARDOWN == False, "Test data won't be erased.")
def test_ClearData(self): def test_ClearData(self):
#1. Delete rule(RA); #1. Delete rule(RA);
for rule_id in TestProjects.rule_id_list: self.replication.delete_replication_rule(TestProjects.rule_id, **ADMIN_CLIENT)
self.replication.delete_replication_rule(rule_id, **ADMIN_CLIENT)
#2. Delete target(TA); #2. Delete registry(TA);
self.target.delete_target(TestProjects.target_id, **ADMIN_CLIENT) self.registry.delete_registry(TestProjects.registry_id, **ADMIN_CLIENT)
#3. Delete project(PA); #3. Delete project(PA);
self.project.delete_project(TestProjects.project_add_rule_id, **TestProjects.USER_add_rule_CLIENT) 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: Test step and expected result:
1. Create a new user(UA); 1. Create a new user(UA);
2. Create a new private project(PA) by user(UA); 2. Create a new private project(PA) by user(UA);
3. Create a new target(TA)/registry; 3. Create a new registry;
4. Create a new rule for project(PA) and target(TA); 4. Create a new rule for this registry;
5. Check if rule is exist. 5. Check rule should be exist.
Tear down: Tear down:
1. Delete rule(RA); 1. Delete rule(RA);
2. Delete targe(TA); 2. Delete registry(TA);
3. Delete project(PA); 3. Delete project(PA);
4. Delete user(UA). 4. Delete user(UA).
""" """
@ -74,21 +74,15 @@ class TestProjects(unittest.TestCase):
self.project.projects_should_exist(dict(public=False), expected_count = 1, self.project.projects_should_exist(dict(public=False), expected_count = 1,
expected_project_id = TestProjects.project_add_rule_id, **TestProjects.USER_add_rule_CLIENT) expected_project_id = TestProjects.project_add_rule_id, **TestProjects.USER_add_rule_CLIENT)
#3. Create a new target(TA)/registry #3. Create a new registry
TestProjects.target_id, _ = self.target.create_target(**ADMIN_CLIENT) TestProjects.registry_id, _ = self.registry.create_registry("https://" + harbor_server,**ADMIN_CLIENT)
print "TestProjects.target_id:", TestProjects.target_id 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"] #5. Check rule should be exist
for value in trigger_values_to_set: self.replication.check_replication_rule_should_exist(TestProjects.rule_id, rule_name, **ADMIN_CLIENT)
#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)
if __name__ == '__main__': if __name__ == '__main__':