CM-30564-fix review

cycode/cli/commands/scan/code_scanner.py

@@ -99,7 +99,7 @@ def set_issue_detected_by_scan_results(context: click.Context, scan_results: Lis
 
 
 def _should_use_scan_service(scan_type: str, scan_parameters: Optional[dict] = None) -> bool:
-    return scan_type == consts.SECRET_SCAN_TYPE and scan_parameters['report'] is True
+    return scan_type == consts.SECRET_SCAN_TYPE and scan_parameters is not None and scan_parameters['report'] is True
 
 
 def _enrich_scan_result_with_data_from_detection_rules(
@@ -326,14 +326,13 @@ def scan_commit_range_documents(
     local_scan_result = error_message = None
     scan_completed = False
     scan_id = str(_generate_unique_id())
-    should_use_scan_service = _should_use_scan_service(scan_type, scan_parameters)
     from_commit_zipped_documents = InMemoryZip()
     to_commit_zipped_documents = InMemoryZip()
 
     try:
         progress_bar.set_section_length(ScanProgressBarSection.SCAN, 1)
 
-        scan_result = init_default_scan_result(cycode_client, scan_id, scan_type, should_use_scan_service)
+        scan_result = init_default_scan_result(cycode_client, scan_id, scan_type)
         if should_scan_documents(from_documents_to_scan, to_documents_to_scan):
             logger.debug('Preparing from-commit zip')
             from_commit_zipped_documents = zip_documents(scan_type, from_documents_to_scan)
@@ -348,7 +347,6 @@ def scan_commit_range_documents(
                 scan_type,
                 scan_parameters,
                 timeout,
-                should_use_scan_service,
             )
 
         progress_bar.update(ScanProgressBarSection.SCAN)
@@ -443,7 +441,7 @@ def perform_scan(
     should_use_scan_service: bool = False,
 ) -> ZippedFileScanResult:
     if scan_type in (consts.SCA_SCAN_TYPE, consts.SAST_SCAN_TYPE) or should_use_scan_service:
-        return perform_scan_async(cycode_client, zipped_documents, scan_type, scan_parameters, should_use_scan_service)
+        return perform_scan_async(cycode_client, zipped_documents, scan_type, scan_parameters)
 
     if is_commit_range:
         return cycode_client.commit_range_zipped_file_scan(scan_type, zipped_documents, scan_id)
@@ -456,19 +454,15 @@ def perform_scan_async(
     zipped_documents: 'InMemoryZip',
     scan_type: str,
     scan_parameters: dict,
-    should_use_scan_service: bool = False,
 ) -> ZippedFileScanResult:
-    scan_async_result = cycode_client.zipped_file_scan_async(
-        zipped_documents, scan_type, scan_parameters, should_use_scan_service=should_use_scan_service
-    )
+    scan_async_result = cycode_client.zipped_file_scan_async(zipped_documents, scan_type, scan_parameters)
     logger.debug('scan request has been triggered successfully, scan id: %s', scan_async_result.scan_id)
 
     return poll_scan_results(
         cycode_client,
         scan_async_result.scan_id,
         scan_type,
         scan_parameters.get('report'),
-        should_use_scan_service=should_use_scan_service,
     )
 
 
@@ -479,20 +473,14 @@ def perform_commit_range_scan_async(
     scan_type: str,
     scan_parameters: dict,
     timeout: Optional[int] = None,
-    should_use_scan_service: bool = False,
 ) -> ZippedFileScanResult:
     scan_async_result = cycode_client.multiple_zipped_file_scan_async(
         from_commit_zipped_documents, to_commit_zipped_documents, scan_type, scan_parameters
     )
 
     logger.debug('scan request has been triggered successfully, scan id: %s', scan_async_result.scan_id)
     return poll_scan_results(
-        cycode_client,
-        scan_async_result.scan_id,
-        scan_type,
-        scan_parameters.get('report'),
-        timeout,
-        should_use_scan_service,
+        cycode_client, scan_async_result.scan_id, scan_type, scan_parameters.get('report'), timeout
     )
 
 
@@ -502,7 +490,6 @@ def poll_scan_results(
     scan_type: str,
     should_get_report: bool = False,
     polling_timeout: Optional[int] = None,
-    should_use_scan_service: bool = False,
 ) -> ZippedFileScanResult:
     if polling_timeout is None:
         polling_timeout = configuration_manager.get_scan_polling_timeout_in_seconds()
@@ -511,7 +498,7 @@ def poll_scan_results(
     end_polling_time = time.time() + polling_timeout
 
     while time.time() < end_polling_time:
-        scan_details = cycode_client.get_scan_details(scan_type, scan_id, should_use_scan_service)
+        scan_details = cycode_client.get_scan_details(scan_type, scan_id)
 
         if scan_details.scan_update_at is not None and scan_details.scan_update_at != last_scan_update_at:
             last_scan_update_at = scan_details.scan_update_at
@@ -610,6 +597,7 @@ def get_default_scan_parameters(context: click.Context) -> dict:
         'report': context.obj.get('report'),
         'package_vulnerabilities': context.obj.get('package-vulnerabilities'),
         'license_compliance': context.obj.get('license-compliance'),
+        'command_type': context.info_name,
     }
 
 

cycode/cyclient/models.py

@@ -171,17 +171,13 @@ def __init__(
         self.err = err
 
 
-class ScanReportUrlResponse(Schema):
-    def __init__(
-        self,
-        report_url: str,
-    ) -> None:
-        super().__init__()
-        self.report_url = report_url
+@dataclass
+class ScanReportUrlResponse:
+    report_url: str
 
 
 class ScanReportUrlResponseSchema(Schema):
-    report_url = fields
+    report_url = fields.String()
 
     @post_load
     def build_dto(self, data: Dict[str, Any], **_) -> 'ScanReportUrlResponse':

cycode/cyclient/scan_client.py

@@ -76,10 +76,10 @@ def get_scan_report_url(self, scan_id: str, scan_type: str) -> models.ScanReport
         response = self.scan_cycode_client.get(url_path=self.get_scan_report_url_path(scan_id, scan_type))
         return models.ScanReportUrlResponseSchema().build_dto(response.json())
 
-    def get_zipped_file_scan_async_url_path(self, scan_type: str, should_use_scan_service: bool = False) -> str:
+    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)
-        scan_service_url_path = self.get_scan_service_url_path(scan_type, should_use_scan_service)
+        scan_service_url_path = self.get_scan_service_url_path(scan_type, True)
         return f'{scan_service_url_path}/{async_scan_type}/{async_entity_type}'
 
     def zipped_file_scan_async(
@@ -88,11 +88,10 @@ def zipped_file_scan_async(
         scan_type: str,
         scan_parameters: dict,
         is_git_diff: bool = False,
-        should_use_scan_service: bool = False,
     ) -> models.ScanInitializationResponse:
         files = {'file': ('multiple_files_scan.zip', zip_file.read())}
         response = self.scan_cycode_client.post(
-            url_path=self.get_zipped_file_scan_async_url_path(scan_type, should_use_scan_service),
+            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,
         )
@@ -118,16 +117,14 @@ def multiple_zipped_file_scan_async(
         )
         return models.ScanInitializationResponseSchema().load(response.json())
 
-    def get_scan_details_path(self, scan_type: str, scan_id: str, should_use_scan_service: bool = False) -> str:
-        return f'{self.get_scan_service_url_path(scan_type, should_use_scan_service)}/{scan_id}'
+    def get_scan_details_path(self, scan_type: str, scan_id: str) -> str:
+        return f'{self.get_scan_service_url_path(scan_type, should_use_scan_service=True)}/{scan_id}'
 
     def get_scan_report_url_path(self, scan_id: str, scan_type: str) -> str:
-        return f'{self.get_scan_service_url_path(scan_type, True)}/reportUrl/{scan_id}'
+        return f'{self.get_scan_service_url_path(scan_type, should_use_scan_service=True)}/reportUrl/{scan_id}'
 
-    def get_scan_details(
-        self, scan_type: str, scan_id: str, should_use_scan_service: bool = False
-    ) -> models.ScanDetailsResponse:
-        path = self.get_scan_details_path(scan_type, scan_id, should_use_scan_service)
+    def get_scan_details(self, scan_type: str, scan_id: str) -> models.ScanDetailsResponse:
+        path = self.get_scan_details_path(scan_type, scan_id)
         response = self.scan_cycode_client.get(url_path=path)
         return models.ScanDetailsResponseSchema().load(response.json())
 

cycode/cyclient/scan_config_base.py

@@ -28,15 +28,11 @@ def get_async_entity_type(scan_type: str) -> str:
     def get_detections_prefix(self) -> str:
         ...
 
-    @abstractmethod
-    def get_scan_service_prefix(self) -> str:
-        ...
-
 
 class DevScanConfig(ScanConfigBase):
     def get_service_name(self, scan_type: str, should_use_scan_service: bool = False) -> str:
         if should_use_scan_service:
-            return self.get_scan_service_prefix()
+            return '5004'
         if scan_type == 'secret':
             return '5025'
         if scan_type == 'iac':
@@ -48,14 +44,11 @@ def get_service_name(self, scan_type: str, should_use_scan_service: bool = False
     def get_detections_prefix(self) -> str:
         return '5016'
 
-    def get_scan_service_prefix(self) -> str:
-        return '5004'
-
 
 class DefaultScanConfig(ScanConfigBase):
     def get_service_name(self, scan_type: str, should_use_scan_service: bool = False) -> str:
         if should_use_scan_service:
-            return self.get_scan_service_prefix()
+            return 'scans'
         if scan_type == 'secret':
             return 'secret'
         if scan_type == 'iac':
@@ -66,6 +59,3 @@ def get_service_name(self, scan_type: str, should_use_scan_service: bool = False
 
     def get_detections_prefix(self) -> str:
         return 'detections'
-
-    def get_scan_service_prefix(self) -> str:
-        return 'scans'

tests/cyclient/mocked_responses/scan_client.py

@@ -88,7 +88,7 @@ def get_scan_report_url(scan_id: Optional[UUID], scan_client: ScanClient, scan_t
 def get_scan_report_url_response(url: str, scan_id: Optional[UUID] = None) -> responses.Response:
     if not scan_id:
         scan_id = uuid4()
-    json_response = {'report_url': 'https://app.domain/on-demand-scans/{scan_id}'.format(scan_id=scan_id)}
+    json_response = {'report_url': f'https://app.domain/on-demand-scans/{scan_id}'}
 
     return responses.Response(method=responses.GET, url=url, json=json_response, status=200)
 

tests/cyclient/scan_config/test_default_scan_config.py

@@ -15,9 +15,3 @@ def test_get_detections_prefix() -> None:
     default_scan_config = DefaultScanConfig()
 
     assert default_scan_config.get_detections_prefix() == 'detections'
-
-
-def test_get_scan_service_prefix() -> None:
-    dev_scan_config = DefaultScanConfig()
-
-    assert dev_scan_config.get_scan_service_prefix() == 'scans'

tests/cyclient/scan_config/test_dev_scan_config.py

@@ -8,16 +8,10 @@ def test_get_service_name() -> None:
     assert dev_scan_config.get_service_name('iac') == '5026'
     assert dev_scan_config.get_service_name('sca') == '5004'
     assert dev_scan_config.get_service_name('sast') == '5004'
-    assert dev_scan_config.get_service_name('secret', True) == '5004'
+    assert dev_scan_config.get_service_name('secret', should_use_scan_service=True) == '5004'
 
 
 def test_get_detections_prefix() -> None:
     dev_scan_config = DevScanConfig()
 
     assert dev_scan_config.get_detections_prefix() == '5016'
-
-
-def test_get_scan_service_prefix() -> None:
-    dev_scan_config = DevScanConfig()
-
-    assert dev_scan_config.get_scan_service_prefix() == '5004'