# -*- coding: utf-8 -*- from __future__ import absolute_import import time import unittest from testutils import harbor_server, suppress_urllib3_warning from testutils import ADMIN_CLIENT from library.scan_data_export import Scan_data_export from library.project import Project from library.user import User from library.artifact import Artifact from library.scan import Scan from library.repository import push_self_build_image_to_project class TestScanDataExport(unittest.TestCase): @suppress_urllib3_warning def setUp(self): self.scan_data_export = Scan_data_export() self.project = Project() self.user = User() self.scan = Scan() self.artifact = Artifact() self.image = "alpine" self.tag = "latest" self.x_scan_data_type = "application/vnd.security.vulnerability.report; version=1.1" def testScanDataExportArtifact(self): """ Test case: Scan Data Export API Test step and expected result: 1. Create a new user(UA); 2. Create a new project(PA) by user(UA); 3. Push a new image(IA) in project(PA) by user(UA); 4. Send scan image command and get tag(TA) information to check scan result, it should be finished; 5. Verify trigger export scan data execution but does not specify Scan-Data-Type status code should be 422; 6. Verify trigger export scan data execution but specifying multiple project status code should be 400; 7. Trigger export scan data execution correctly; 8. Verify that the export scan data execution triggered by the user(UA) cannot be queried by other users; 9. User (UA) should be able to query the triggered export scan data execution; 10. Wait for the export scan data execution to succeed; 11. Verify that the export scan data execution triggered by the user (UA) cannot be download by other users; 12. User (UA) should be able to download the triggered export scan data execution 13. Verify that the downloaded export scan data execution cannot be downloaded again """ url = ADMIN_CLIENT["endpoint"] user_password = "Aa123456" # 1. Create user(UA) user_id, user_name = self.user.create_user(user_password = user_password, **ADMIN_CLIENT) user_client = dict(endpoint = url, username = user_name, password = user_password) # 2.1. Create private project(PA) by user(UA) project_id, project_name = self.project.create_project(metadata = {"public": "false"}, **user_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 = project_id, **user_client) # 3. Push a new image(IA) in project(PA) by user(UA) push_self_build_image_to_project(project_name, harbor_server, user_name, user_password, self.image, self.tag) # 4. Send scan image command and get tag(TA) information to check scan result, it should be finished self.scan.scan_artifact(project_name, self.image, self.tag, **user_client) self.artifact.check_image_scan_result(project_name, self.image, self.tag, with_scan_overview = True, **user_client) # 5. Verify trigger export scan data execution but does not specify Scan-Data-Type status code should be 422 self.scan_data_export.export_scan_data("", projects=[project_id], expect_status_code=422, expect_response_body="X-Scan-Data-Type in header is required") # 6. Verify trigger export scan data execution but specifying multiple project status code should be 400 self.scan_data_export.export_scan_data(self.x_scan_data_type, projects=[1, project_id], expect_status_code=400, expect_response_body="bad request: only support export single project") # 7. Trigger export scan data execution correctly execution_id = self.scan_data_export.export_scan_data(self.x_scan_data_type, projects=[project_id], **user_client).id print("execution_id:", execution_id) # 8.1. Verify that the export scan data execution triggered by the user(UA) cannot be queried by other users by get scan data export execution list API execution_list = self.scan_data_export.get_scan_data_export_execution_list() if not execution_list: self.assertNotEqual(execution_id, execution_list.items[0].id) self.assertEqual(ADMIN_CLIENT["username"], execution_list.items[0].user_name) # 8.2. Verify that the export scan data execution triggered by the user(UA) cannot be queried by other users by get scan_data export execution API self.scan_data_export.get_scan_data_export_execution(execution_id, expect_status_code=403, expect_response_body="FORBIDDEN") # 9. User (UA) should be able to query the triggered export scan data execution execution_list = self.scan_data_export.get_scan_data_export_execution_list(**user_client) self.assertEqual(execution_id, execution_list.items[0].id) self.assertEqual(user_name, execution_list.items[0].user_name) # 10. Wait for the export scan data execution to succeed executio_status = None for i in range(5): print("wait for the job to finish:", i) execution = self.scan_data_export.get_scan_data_export_execution(execution_id, **user_client) executio_status = execution.status if executio_status == "Success": self.assertEqual(user_name, execution.user_name) self.assertEqual(user_id, execution.user_id) break time.sleep(2) self.assertEqual(executio_status, "Success") # 11. Verify that the export scan data execution triggered by the user (UA) cannot be download by other users self.scan_data_export.download_scan_data(execution_id, expect_status_code=403) # 12. User (UA) should be able to download the triggered export scan data execution self.scan_data_export.download_scan_data(execution_id, **user_client) # 13. Verify that the downloaded export scan data execution cannot be downloaded again self.scan_data_export.download_scan_data(execution_id, expect_status_code=404, **user_client) if __name__ == '__main__': unittest.main()