return refactoring

Author: MarshalX

cycode/cyclient/scan_client.py

@@ -31,28 +31,36 @@ def content_scan(self, scan_type: str, file_name: str, content: str, is_git_diff
         )
         return self.parse_scan_response(response)
 
+    def get_zipped_file_scan_url_path(self, scan_type: str) -> str:
+        return f'{self.scan_config.get_service_name(scan_type)}/{self.SCAN_CONTROLLER_PATH}/zipped-file'
+
     def zipped_file_scan(
         self, scan_type: str, zip_file: InMemoryZip, scan_id: str, scan_parameters: dict, is_git_diff: bool = False
     ) -> models.ZippedFileScanResult:
-        url_path = f'{self.scan_config.get_service_name(scan_type)}/{self.SCAN_CONTROLLER_PATH}/zipped-file'
         files = {'file': ('multiple_files_scan.zip', zip_file.read())}
 
         response = self.scan_cycode_client.post(
-            url_path=url_path,
+            url_path=self.get_zipped_file_scan_url_path(scan_type),
             data={'scan_id': scan_id, 'is_git_diff': is_git_diff, 'scan_parameters': json.dumps(scan_parameters)},
             files=files,
             hide_response_content_log=self._hide_response_log,
         )
 
         return self.parse_zipped_file_scan_response(response)
 
+    def get_zipped_file_scan_async_url_path(self, scan_type: str) -> str:
+        async_scan_type = self.scan_config.get_async_scan_type(scan_type)
+        async_entity_type = self.scan_config.get_async_entity_type(scan_type)
+
+        url_prefix = self.scan_config.get_scans_prefix()
+        return f'{url_prefix}/{self.SCAN_CONTROLLER_PATH}/{async_scan_type}/{async_entity_type}'
+
     def zipped_file_scan_async(
         self, zip_file: InMemoryZip, scan_type: str, scan_parameters: dict, is_git_diff: bool = False
     ) -> models.ScanInitializationResponse:
-        url_path = f'{self.scan_config.get_scans_prefix()}/{self.SCAN_CONTROLLER_PATH}/{scan_type}/repository'
         files = {'file': ('multiple_files_scan.zip', zip_file.read())}
         response = self.scan_cycode_client.post(
-            url_path=url_path,
+            url_path=self.get_zipped_file_scan_async_url_path(scan_type),
             data={'is_git_diff': is_git_diff, 'scan_parameters': json.dumps(scan_parameters)},
             files=files,
         )
@@ -80,13 +88,17 @@ def multiple_zipped_file_scan_async(
         )
         return models.ScanInitializationResponseSchema().load(response.json())
 
+    def get_scan_details_path(self, scan_id: str) -> str:
+        return f'{self.scan_config.get_scans_prefix()}/{self.SCAN_CONTROLLER_PATH}/{scan_id}'
+
     def get_scan_details(self, scan_id: str) -> models.ScanDetailsResponse:
-        url_path = f'{self.scan_config.get_scans_prefix()}/{self.SCAN_CONTROLLER_PATH}/{scan_id}'
-        response = self.scan_cycode_client.get(url_path=url_path)
+        response = self.scan_cycode_client.get(url_path=self.get_scan_details_path(scan_id))
         return models.ScanDetailsResponseSchema().load(response.json())
 
+    def get_scan_detections_path(self) -> str:
+        return f'{self.scan_config.get_detections_prefix()}/{self.DETECTIONS_SERVICE_CONTROLLER_PATH}'
+
     def get_scan_detections(self, scan_id: str) -> List[dict]:
-        url_path = f'{self.scan_config.get_detections_prefix()}/{self.DETECTIONS_SERVICE_CONTROLLER_PATH}'
         params = {'scan_id': scan_id}
 
         page_size = 200
@@ -100,7 +112,9 @@ def get_scan_detections(self, scan_id: str) -> List[dict]:
             params['page_number'] = page_number
 
             response = self.scan_cycode_client.get(
-                url_path=url_path, params=params, hide_response_content_log=self._hide_response_log
+                url_path=self.get_scan_detections_path(),
+                params=params,
+                hide_response_content_log=self._hide_response_log,
             ).json()
             detections.extend(response)
 
@@ -109,9 +123,13 @@ def get_scan_detections(self, scan_id: str) -> List[dict]:
 
         return detections
 
+    def get_get_scan_detections_count_path(self) -> str:
+        return f'{self.scan_config.get_detections_prefix()}/{self.DETECTIONS_SERVICE_CONTROLLER_PATH}/count'
+
     def get_scan_detections_count(self, scan_id: str) -> int:
-        url_path = f'{self.scan_config.get_detections_prefix()}/{self.DETECTIONS_SERVICE_CONTROLLER_PATH}/count'
-        response = self.scan_cycode_client.get(url_path=url_path, params={'scan_id': scan_id})
+        response = self.scan_cycode_client.get(
+            url_path=self.get_get_scan_detections_count_path(), params={'scan_id': scan_id}
+        )
         return response.json().get('count', 0)
 
     def commit_range_zipped_file_scan(
@@ -126,9 +144,11 @@ def commit_range_zipped_file_scan(
         )
         return self.parse_zipped_file_scan_response(response)
 
+    def get_report_scan_status_path(self, scan_type: str, scan_id: str) -> str:
+        return f'{self.scan_config.get_service_name(scan_type)}/{self.SCAN_CONTROLLER_PATH}/{scan_id}/status'
+
     def report_scan_status(self, scan_type: str, scan_id: str, scan_status: dict) -> None:
-        url_path = f'{self.scan_config.get_service_name(scan_type)}/{self.SCAN_CONTROLLER_PATH}/{scan_id}/status'
-        self.scan_cycode_client.post(url_path=url_path, body=scan_status)
+        self.scan_cycode_client.post(url_path=self.get_report_scan_status_path(scan_type, scan_id), body=scan_status)
 
     @staticmethod
     def parse_scan_response(response: Response) -> models.ScanResult:
@@ -140,6 +160,7 @@ def parse_zipped_file_scan_response(response: Response) -> models.ZippedFileScan
 
     @staticmethod
     def get_service_name(scan_type: str) -> Optional[str]:
+        # TODO(MarshalX): get_service_name should be removed from ScanClient? Because it exists in ScanConfig
         if scan_type == 'secret':
             return 'secret'
         if scan_type == 'iac':

cycode/cyclient/scan_config_base.py

@@ -6,6 +6,20 @@ class ScanConfigBase(ABC):
     def get_service_name(self, scan_type: str) -> str:
         ...
 
+    @staticmethod
+    def get_async_scan_type(scan_type: str) -> str:
+        if scan_type == 'secret':
+            return 'Secrets'
+        if scan_type == 'iac':
+            return 'InfraConfiguration'
+
+        return scan_type.upper()
+
+    @staticmethod
+    def get_async_entity_type(_: str) -> str:
+        # we are migrating to "zippedfile" entity type. will be used later
+        return 'repository'
+
     @abstractmethod
     def get_scans_prefix(self) -> str:
         ...

tests/cyclient/test_scan_client.py

@@ -28,9 +28,8 @@ def zip_scan_resources(scan_type: str, scan_client: ScanClient) -> Tuple[str, In
 
 def get_zipped_file_scan_url(scan_type: str, scan_client: ScanClient) -> str:
     api_url = scan_client.scan_cycode_client.api_url
-    # TODO(MarshalX): create method in the scan client to build this url
-    service_url = f'{api_url}/{scan_client.scan_config.get_service_name(scan_type)}'
-    return f'{service_url}/{scan_client.SCAN_CONTROLLER_PATH}/zipped-file'
+    service_url = scan_client.get_zipped_file_scan_url_path(scan_type)
+    return f'{api_url}/{service_url}'
 
 
 def get_test_zip_file(scan_type: str) -> InMemoryZip: