Merge pull request #12358 from danfengliu/delete-signed-image

Verify error message when deleting signed image

tests/apitests/python/library/replication.py

@@ -13,8 +13,7 @@ class Replication(base.Base):
             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,
@@ -56,20 +55,31 @@ class Replication(base.Base):
         return client.jobs_replication_get(int(rule_id))
 
     def wait_until_jobs_finish(self, rule_id, retry=10, interval=5, **kwargs):
-        finished = True
+        Succeed = False
         for i in range(retry):
-            finished = True
-            jobs = self.list_replication_jobs(rule_id, **kwargs)
+            Succeed = False
+            jobs = self.get_replication_executions(rule_id, **kwargs)
             for job in jobs:
-                if job.status != "finished":
-                    finished = False
-                    break
-            if not finished:
+                if job.status == "Succeed":
+                    return
+            if not Succeed:
                 time.sleep(interval)
-        if not finished:
-            raise Exception("The jobs not finished")
+        if not Succeed:
+            raise Exception("The jobs not Succeed")
 
     def delete_replication_rule(self, rule_id, expect_status_code = 200, **kwargs):
         client = self._get_client(**kwargs)
         _, status_code, _ = client.replication_policies_id_delete_with_http_info(rule_id)
         base._assert_status_code(expect_status_code, status_code)
+
+    def trigger_replication_executions(self, rule_id, expect_status_code = 201, **kwargs):
+        client = self._get_client(**kwargs)
+        _, status_code, _ = client.replication_executions_post_with_http_info({"policy_id":rule_id})
+        base._assert_status_code(expect_status_code, status_code)
+
+
+    def get_replication_executions(self, rule_id, expect_status_code = 200, **kwargs):
+        client = self._get_client(**kwargs)
+        data, status_code, _ = client.replication_executions_get_with_http_info(policy_id=rule_id)
+        base._assert_status_code(expect_status_code, status_code)
+        return data

tests/apitests/python/library/repository.py

@@ -78,11 +78,17 @@ class Repository(base.Base, object):
         if self.image_exists(repository, tag, **kwargs):
             raise Exception("image %s:%s exists" % (repository, tag))
 
-    def delete_repoitory(self, project_name, repo_name, **kwargs):
+    def delete_repoitory(self, project_name, repo_name, expect_status_code = 200, expect_response_body = None, **kwargs):
         client = self._get_client(**kwargs)
+        try:
             _, status_code, _ = client.delete_repository_with_http_info(project_name, repo_name)
+        except Exception as e:
+            base._assert_status_code(expect_status_code, e.status)
+            return e.body
+        base._assert_status_code(expect_status_code, status_code)
         base._assert_status_code(200, status_code)
 
+
     def list_repositories(self, project_name, **kwargs):
         client = self._get_client(**kwargs)
         data, status_code, _ = client.list_repositories_with_http_info(project_name)

tests/apitests/python/library/tag_immutability.py

@@ -0,0 +1,6 @@
+# -*- coding: utf-8 -*-
+
+import base
+import swagger_client
+
+class Tag_Immutability(base.Base):

tests/apitests/python/test_replication_from_dockerhub.py

@@ -0,0 +1,106 @@
+from __future__ import absolute_import
+import unittest
+
+from testutils import ADMIN_CLIENT
+from testutils import TEARDOWN
+from library.project import Project
+from library.user import User
+from library.replication import Replication
+from library.registry import Registry
+from library.artifact import Artifact
+from library.repository import Repository
+import swagger_client
+
+class TestProjects(unittest.TestCase):
+    @classmethod
+    def setUp(self):
+        self.project = Project()
+        self.user = User()
+        self.replication = Replication()
+        self.registry = Registry()
+        self.artifact = Artifact()
+        self.repo = Repository()
+        self.image = "alpine"
+        self.tag = "latest"
+
+    @classmethod
+    def tearDown(self):
+        print "Case completed"
+
+    @unittest.skipIf(TEARDOWN == False, "Test data won't be erased.")
+    def test_ClearData(self):
+        #1. Delete rule(RA);
+        self.replication.delete_replication_rule(TestProjects.rule_id, **ADMIN_CLIENT)
+
+        #2. Delete registry(TA);
+        self.registry.delete_registry(TestProjects.registry_id, **ADMIN_CLIENT)
+
+        #1. Delete repository(RA);
+        self.repo.delete_repoitory(TestProjects.project_name, self.image, **TestProjects.USER_add_rule_CLIENT)
+
+        #3. Delete project(PA);
+        self.project.delete_project(TestProjects.project_add_rule_id, **TestProjects.USER_add_rule_CLIENT)
+
+        #4. Delete user(UA);
+        self.user.delete_user(TestProjects.user_add_rule_id, **ADMIN_CLIENT)
+
+    def testReplicationFromDockerhub(self):
+        """
+        Test case:
+            Replication From Dockerhub
+        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 registry;
+            4. Create a new rule for this registry;
+            5. Check rule should be exist;
+            6. Trigger the rule;
+            7. Wait for completion of this replication job;
+            8. Check image is replicated into target project successfully.
+        Tear down:
+            1. Delete rule(RA);
+            2. Delete registry(TA);
+            3. Delete project(PA);
+            4. Delete user(UA).
+        """
+        url = ADMIN_CLIENT["endpoint"]
+        user_add_rule_password = "Aa123456"
+
+        #1. Create user(UA)
+        TestProjects.user_add_rule_id, user_add_rule_name = self.user.create_user(user_password = user_add_rule_password, **ADMIN_CLIENT)
+
+        TestProjects.USER_add_rule_CLIENT=dict(endpoint = url, username = user_add_rule_name, password = user_add_rule_password)
+
+        #2.1. Create private project(PA) by user(UA)
+        TestProjects.project_add_rule_id, TestProjects.project_name = self.project.create_project(metadata = {"public": "false"}, **TestProjects.USER_add_rule_CLIENT)
+
+        #2.2. Get private project of uesr-001, uesr-001 can see only one private project which is project-001
+        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 registry;
+        TestProjects.registry_id, _ = self.registry.create_registry("https://hub.docker.com", registry_type="docker-hub", access_key = "", access_secret = "", insecure=False, **ADMIN_CLIENT)
+        print "TestProjects.registry_id:", TestProjects.registry_id
+
+        #4. Create a pull-based rule for this registry;
+        TestProjects.rule_id, rule_name = self.replication.create_replication_policy(src_registry=swagger_client.Registry(id=int(TestProjects.registry_id)),
+                                            dest_namespace=TestProjects.project_name,
+                                            filters=[swagger_client.ReplicationFilter(type="name",value="library/"+self.image),swagger_client.ReplicationFilter(type="tag",value=self.tag)],
+                                            **ADMIN_CLIENT)
+
+        #5. Check rule should be exist;
+        self.replication.check_replication_rule_should_exist(TestProjects.rule_id, rule_name, **ADMIN_CLIENT)
+
+        #6. Trigger the rule;
+        self.replication.trigger_replication_executions(TestProjects.rule_id, **ADMIN_CLIENT)
+
+        #7. Wait for completion of this replication job;
+        self.replication.wait_until_jobs_finish(TestProjects.rule_id, **ADMIN_CLIENT)
+
+        #8. Check image is replicated into target project successfully.
+        artifact = self.artifact.get_reference_info(TestProjects.project_name, self.image, self.tag, **ADMIN_CLIENT)
+        print artifact
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/apitests/python/test_sign_image.py

@@ -10,6 +10,7 @@ from library.project import Project
 from library.user import User
 from library.repository import Repository
 from library.repository import push_image_to_project
+from library.repository import push_special_image_to_project
 
 class TestProjects(unittest.TestCase):
     @classmethod
@@ -18,6 +19,7 @@ class TestProjects(unittest.TestCase):
         self.user = User()
         self.artifact = Artifact()
         self.repo = Repository()
+        self.repo_name_1 = "test1_sign"
 
     @classmethod
     def tearDown(self):
@@ -80,5 +82,12 @@ class TestProjects(unittest.TestCase):
         artifact = self.artifact.get_reference_info(TestProjects.project_sign_image_name, image, tag, **TestProjects.USER_sign_image_CLIENT)
         self.assertEqual(artifact[0].tags[0].signed, True)
 
+        push_special_image_to_project(TestProjects.project_sign_image_name, harbor_server, user_sign_image_name, user_001_password, self.repo_name_1, ['1.0'])
+        self.repo.delete_repoitory(TestProjects.project_sign_image_name, self.repo_name_1, **TestProjects.USER_sign_image_CLIENT)
+
+        ret = self.repo.delete_repoitory(TestProjects.project_sign_image_name, TestProjects.repo_name.split('/')[1], expect_status_code=412, **TestProjects.USER_sign_image_CLIENT)
+        self.assertIn("with signature cannot be deleted", ret)
+
+
 if __name__ == '__main__':
     unittest.main()

tests/robot-cases/Group0-BAT/API_DB.robot

@@ -137,3 +137,7 @@ Test Case - Push Singularity file With Singularity CLI
 Test Case - Push Chart File To Chart Repository By Helm V2 With Robot Account
     [Tags]  helm2
     Harbor API Test  ./tests/apitests/python/test_push_chart_by_helm2_helm3_with_robot_Account.py
+
+Test Case - Replication From Dockerhub
+    [Tags]  replic_dockerhub
+    Harbor API Test  ./tests/apitests/python/test_replication_from_dockerhub.py