feat: add python testing script for stop scan job (#15779)

Signed-off-by: Shengwen Yu <yshengwen@vmware.com>

Co-authored-by: Shengwen Yu <yshengwen@vmware.com>

tests/apitests/python/library/scan_all_stop.py

@@ -0,0 +1,23 @@
+# -*- coding: utf-8 -*-
+
+import time
+import base
+import v2_swagger_client
+from v2_swagger_client.rest import ApiException
+
+class StopScanAll(base.Base):
+    def __init__(self):
+        super(StopScanAll,self).__init__(api_type="scanall")
+
+    def stop_scan_all(self, expect_status_code=202, expect_response_body=None, **kwargs):
+        try:
+            _, status_code, _ = self._get_client(**kwargs).stop_scan_all_with_http_info()
+        except ApiException as e:
+            if e.status == expect_status_code:
+                if expect_response_body is not None and e.body.strip() != expect_response_body.strip():
+                    raise Exception(r"Stop scan all response body is not as expected {} actual status is {}.".format(expect_response_body.strip(), e.body.strip()))
+                else:
+                    return e.reason, e.body
+            else:
+                raise Exception(r"Stop scan all result is not as expected {} actual status is {}.".format(expect_status_code, e.status))
+        base._assert_status_code(expect_status_code, status_code)

tests/apitests/python/library/scan_stop.py

@@ -0,0 +1,23 @@
+# -*- coding: utf-8 -*-
+
+import time
+import base
+import v2_swagger_client
+from v2_swagger_client.rest import ApiException
+
+class StopScan(base.Base, object):
+    def __init__(self):
+        super(StopScan,self).__init__(api_type = "scan")
+
+    def stop_scan_artifact(self, project_name, repo_name, reference, expect_status_code = 202, expect_response_body = None, **kwargs):
+        try:
+            data, status_code, _ = self._get_client(**kwargs).stop_scan_artifact_with_http_info(project_name, repo_name, reference)
+        except ApiException as e:
+            base._assert_status_code(expect_status_code, e.status)
+            if expect_response_body is not None:
+                base._assert_status_body(expect_response_body, e.body)
+            return
+
+        base._assert_status_code(expect_status_code, status_code)
+
+        return data

tests/apitests/python/test_stop_scan_image_artifact.py

@@ -0,0 +1,91 @@
+from __future__ import absolute_import
+import unittest
+import sys
+
+from testutils import harbor_server, suppress_urllib3_warning
+from testutils import TEARDOWN
+from testutils import ADMIN_CLIENT, BASE_IMAGE, BASE_IMAGE_ABS_PATH_NAME
+from library.project import Project
+from library.user import User
+from library.repository import Repository
+from library.repository import push_self_build_image_to_project
+from library.artifact import Artifact
+from library.scan import Scan
+from library.scan_stop import StopScan
+
+class TestStopScan(unittest.TestCase):
+    @suppress_urllib3_warning
+    def setUp(self):
+        self.project= Project()
+        self.user= User()
+        self.artifact = Artifact()
+        self.repo = Repository()
+        self.scan = Scan()
+        self.stop_scan = StopScan()
+
+        self.url = ADMIN_CLIENT["endpoint"]
+        self.user_password = "Aa123456"
+        self.project_id, self.project_name, self.user_id, self.user_name, self.repo_name1 = [None] * 5
+        self.user_id, self.user_name = self.user.create_user(user_password = self.user_password, **ADMIN_CLIENT)
+        self.USER_CLIENT = dict(with_signature = True, with_immutable_status = True, endpoint = self.url, username = self.user_name, password = self.user_password, with_scan_overview = True)
+
+
+        #2. Create a new private project(PA) by user(UA);
+        self.project_id, self.project_name = self.project.create_project(metadata = {"public": "false"}, **ADMIN_CLIENT)
+
+        #3. Add user(UA) as a member of project(PA) with project-admin role;
+        self.project.add_project_members(self.project_id, user_id = self.user_id, **ADMIN_CLIENT)
+
+    @unittest.skipIf(TEARDOWN == False, "Test data won't be erased.")
+    def do_tearDown(self):
+        #1. Delete repository(RA) by user(UA);
+        self.repo.delete_repository(self.project_name, self.repo_name1.split('/')[1], **self.USER_CLIENT)
+
+        #2. Delete project(PA);
+        self.project.delete_project(self.project_id, **self.USER_CLIENT)
+
+        #3. Delete user(UA);
+        self.user.delete_user(self.user_id, **ADMIN_CLIENT)
+
+    def testStopScanImageArtifact(self):
+        """
+        Test case:
+            Stop Scan An Image Artifact
+        Test step and expected result:
+            1. Create a new user(UA);
+            2. Create a new private project(PA) by user(UA);
+            3. Add user(UA) as a member of project(PA) with project-admin role;
+            4. Get private project of user(UA), user(UA) can see only one private project which is project(PA);
+            5. Create a new repository(RA) and tag(TA) in project(PA) by user(UA);
+            6. Send scan image command;
+            7. Send stop scan image command.
+        Tear down:
+            1. Delete repository(RA) by user(UA);
+            2. Delete project(PA);
+            3. Delete user(UA);
+        """
+
+        #4. Get private project of user(UA), user(UA) can see only one private project which is project(PA);
+        self.project.projects_should_exist(dict(public=False), expected_count = 1,
+                                           expected_project_id = self.project_id, **self.USER_CLIENT)
+
+        #Note: Please make sure that this Image has never been pulled before by any other cases,
+        #      so it is a not-scanned image right after repository creation.
+        image = "docker"
+        src_tag = "1.13"
+        #5. Create a new repository(RA) and tag(TA) in project(PA) by user(UA);
+        self.repo_name1, tag = push_self_build_image_to_project(self.project_name, harbor_server, self.user_name, self.user_password, image, src_tag)
+
+        #6. Send scan image command;
+        self.scan.scan_artifact(self.project_name, self.repo_name1.split('/')[1], tag, **self.USER_CLIENT)
+
+        #7. Send stop scan image command.
+        self.stop_scan.stop_scan_artifact(self.project_name, self.repo_name1.split('/')[1], tag, **self.USER_CLIENT)
+
+        self.do_tearDown()
+
+if __name__ == '__main__':
+    suite = unittest.TestSuite(unittest.makeSuite(TestStopScan))
+    result = unittest.TextTestRunner(sys.stdout, verbosity=2, failfast=True).run(suite)
+    if not result.wasSuccessful():
+        raise Exception(r"Stop Scan test failed: {}".format(result))

tests/apitests/python/test_system_level_stop_scan_all.py

@@ -0,0 +1,94 @@
+from __future__ import absolute_import
+import unittest
+
+from testutils import harbor_server, suppress_urllib3_warning
+from testutils import TEARDOWN
+from testutils import ADMIN_CLIENT
+from library.project import Project
+from library.user import User
+from library.repository import Repository
+from library.repository import push_self_build_image_to_project
+from library.artifact import Artifact
+from library.scan_all import ScanAll
+from library.scan_all_stop import StopScanAll
+
+class TestStopScanAll(unittest.TestCase):
+    @suppress_urllib3_warning
+    def setUp(self):
+        self.project= Project()
+        self.user= User()
+        self.artifact = Artifact()
+        self.repo = Repository()
+        self.scan_all = ScanAll()
+        self.stop_scan_all = StopScanAll()
+
+    @unittest.skipIf(TEARDOWN == False, "Test data won't be erased.")
+    def tearDown(self):
+        #1. Delete Alice's repository and Luca's repository;
+        self.repo.delete_repository(TestStopScanAll.project_Alice_name, TestStopScanAll.repo_Alice_name.split('/')[1], **ADMIN_CLIENT)
+        self.repo.delete_repository(TestStopScanAll.project_Luca_name, TestStopScanAll.repo_Luca_name.split('/')[1], **ADMIN_CLIENT)
+
+        #2. Delete Alice's project and Luca's project;
+        self.project.delete_project(TestStopScanAll.project_Alice_id, **ADMIN_CLIENT)
+        self.project.delete_project(TestStopScanAll.project_Luca_id, **ADMIN_CLIENT)
+
+        #3. Delete user Alice and Luca.
+        self.user.delete_user(TestStopScanAll.user_Alice_id, **ADMIN_CLIENT)
+        self.user.delete_user(TestStopScanAll.user_Luca_id, **ADMIN_CLIENT)
+        print("Case completed")
+
+    def testSystemLevelScanALL(self):
+        """
+        Test case:
+            System level Stop Scan All
+        Test step and expected result:
+            1. Create user Alice and Luca;
+            2. Create 2 new private projects project_Alice and project_Luca;
+            3. Push a image to project_Alice and push another image to project_Luca;
+            4. Trigger scan all event;
+            5. Send stop scan all request.
+        Tear down:
+            1. Delete Alice's repository and Luca's repository;
+            2. Delete Alice's project and Luca's project;
+            3. Delete user Alice and Luca.
+        """
+        url = ADMIN_CLIENT["endpoint"]
+        user_common_password = "Aa123456"
+
+        #1. Create user Alice and Luca;
+        TestStopScanAll.user_Alice_id, user_Alice_name = self.user.create_user(user_password = user_common_password, **ADMIN_CLIENT)
+        TestStopScanAll.user_Luca_id, user_Luca_name = self.user.create_user(user_password = user_common_password, **ADMIN_CLIENT)
+
+        USER_ALICE_CLIENT=dict(endpoint = url, username = user_Alice_name, password = user_common_password, with_scan_overview = True)
+        USER_LUCA_CLIENT=dict(endpoint = url, username = user_Luca_name, password = user_common_password, with_scan_overview = True)
+
+        #2. Create 2 new private projects project_Alice and project_Luca;
+        TestStopScanAll.project_Alice_id, TestStopScanAll.project_Alice_name = self.project.create_project(metadata = {"public": "false"}, **USER_ALICE_CLIENT)
+        TestStopScanAll.project_Luca_id, TestStopScanAll.project_Luca_name = self.project.create_project(metadata = {"public": "false"}, **USER_LUCA_CLIENT)
+
+        #3. Push a image to project_Alice and push another image to project_Luca;
+
+        #Note: Please make sure that this Image has never been pulled before by any other cases,
+        #          so it is a not-scanned image rigth after repository creation.
+        #image = "tomcat"
+        image_a = "mariadb"
+        src_tag = "latest"
+        #3.1 Push a image to project_Alice;
+        TestStopScanAll.repo_Alice_name, tag_Alice = push_self_build_image_to_project(TestStopScanAll.project_Alice_name, harbor_server, user_Alice_name, user_common_password, image_a, src_tag)
+
+        #Note: Please make sure that this Image has never been pulled before by any other cases,
+        #          so it is a not-scanned image rigth after repository creation.
+        image_b = "httpd"
+        src_tag = "latest"
+        #3.2 push another image to project_Luca;
+        TestStopScanAll.repo_Luca_name, tag_Luca = push_self_build_image_to_project(TestStopScanAll.project_Luca_name, harbor_server, user_Luca_name, user_common_password, image_b, src_tag)
+
+        #4. Trigger scan all event;
+        self.scan_all.scan_all_now(**ADMIN_CLIENT)
+        # self.scan_all.wait_until_scans_all_finish(**ADMIN_CLIENT)
+
+        #5. Send stop scan all request.
+        self.stop_scan_all.stop_scan_all()
+
+if __name__ == '__main__':
+    unittest.main()

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

@@ -117,6 +117,14 @@ Test Case - Scan All Images
     [Tags]  scan_all
     Harbor API Test  ./tests/apitests/python/test_system_level_scan_all.py
 
+Test Case - Stop Scan Image
+    [Tags]  stop_scan
+    Harbor API Test  ./tests/apitests/python/test_stop_scan_image_artifact.py
+
+Test Case - Stop Scan All Images
+    [Tags]  stop_scan_all
+    Harbor API Test  ./tests/apitests/python/test_system_level_stop_scan_all.py
+
 Test Case - Registry API
     [Tags]  reg_api
     Harbor API Test  ./tests/apitests/python/test_registry_api.py