diff --git a/tests/apitests/python/library/registry.py b/tests/apitests/python/library/registry.py index a0466ce58..fe25fa7a6 100644 --- a/tests/apitests/python/library/registry.py +++ b/tests/apitests/python/library/registry.py @@ -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) \ No newline at end of file + 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) \ No newline at end of file diff --git a/tests/apitests/python/library/replication.py b/tests/apitests/python/library/replication.py index 9406339b1..7e1cdd646 100644 --- a/tests/apitests/python/library/replication.py +++ b/tests/apitests/python/library/replication.py @@ -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) diff --git a/tests/apitests/python/library/target.py b/tests/apitests/python/library/target.py deleted file mode 100644 index ef350ea92..000000000 --- a/tests/apitests/python/library/target.py +++ /dev/null @@ -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) \ No newline at end of file diff --git a/tests/apitests/python/test_add_replication_rule.py b/tests/apitests/python/test_add_replication_rule.py index 86035ee81..82e500c5c 100644 --- a/tests/apitests/python/test_add_replication_rule.py +++ b/tests/apitests/python/test_add_replication_rule.py @@ -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__':