99
1010from cycode .cli .cli_types import ScanTypeOption
1111from cycode .cli .exceptions .custom_exceptions import (
12+ CycodeError ,
1213 HttpUnauthorizedError ,
1314 RequestConnectionError ,
14- RequestHttpError ,
1515 RequestTimeout ,
1616)
1717from cycode .cli .files_collector .models .in_memory_zip import InMemoryZip
2828)
2929
3030
31- def zip_scan_resources (scan_type : str , scan_client : ScanClient ) -> Tuple [str , InMemoryZip ]:
31+ def zip_scan_resources (scan_type : ScanTypeOption , scan_client : ScanClient ) -> Tuple [str , InMemoryZip ]:
3232 url = get_zipped_file_scan_async_url (scan_type , scan_client )
3333 zip_file = get_test_zip_file (scan_type )
3434
3535 return url , zip_file
3636
3737
38- def get_test_zip_file (scan_type : str ) -> InMemoryZip :
38+ def get_test_zip_file (scan_type : ScanTypeOption ) -> InMemoryZip :
3939 # TODO(MarshalX): refactor scan_disk_files in code_scanner.py to reuse method here instead of this
4040 test_documents : List [Document ] = []
4141 for root , _ , files in os .walk (ZIP_CONTENT_PATH ):
@@ -52,7 +52,7 @@ def get_test_zip_file(scan_type: str) -> InMemoryZip:
5252@pytest .mark .parametrize ('scan_type' , list (ScanTypeOption ))
5353@responses .activate
5454def test_zipped_file_scan_async (
55- scan_type : str , scan_client : ScanClient , api_token_response : responses .Response
55+ scan_type : ScanTypeOption , scan_client : ScanClient , api_token_response : responses .Response
5656) -> None :
5757 """Test the zipped_file_scan_async method for the async flow."""
5858 url , zip_file = zip_scan_resources (scan_type , scan_client )
@@ -61,16 +61,15 @@ def test_zipped_file_scan_async(
6161 responses .add (api_token_response ) # mock token based client
6262 responses .add (get_zipped_file_scan_async_response (url , expected_scan_id ))
6363
64- # Call the method with the correct parameter order
65- scan_initialization_response = scan_client .zipped_file_scan_async (
66- zip_file = zip_file , scan_type = scan_type , scan_parameters = {}
67- )
64+ scan_initialization_response = scan_client .zipped_file_scan_async (zip_file , scan_type , scan_parameters = {})
6865 assert scan_initialization_response .scan_id == str (expected_scan_id )
6966
7067
7168@pytest .mark .parametrize ('scan_type' , list (ScanTypeOption ))
7269@responses .activate
73- def test_get_scan_report_url (scan_type : str , scan_client : ScanClient , api_token_response : responses .Response ) -> None :
70+ def test_get_scan_report_url (
71+ scan_type : ScanTypeOption , scan_client : ScanClient , api_token_response : responses .Response
72+ ) -> None :
7473 """Test getting the scan report URL for the async flow."""
7574 scan_id = uuid4 ()
7675 url = get_scan_aggregation_report_url (scan_id , scan_client , scan_type )
@@ -85,7 +84,7 @@ def test_get_scan_report_url(scan_type: str, scan_client: ScanClient, api_token_
8584@pytest .mark .parametrize ('scan_type' , list (ScanTypeOption ))
8685@responses .activate
8786def test_zipped_file_scan_async_unauthorized_error (
88- scan_type : str , scan_client : ScanClient , api_token_response : responses .Response
87+ scan_type : ScanTypeOption , scan_client : ScanClient , api_token_response : responses .Response
8988) -> None :
9089 """Test handling of unauthorized errors in the async flow."""
9190 url , zip_file = zip_scan_resources (scan_type , scan_client )
@@ -102,7 +101,7 @@ def test_zipped_file_scan_async_unauthorized_error(
102101@pytest .mark .parametrize ('scan_type' , list (ScanTypeOption ))
103102@responses .activate
104103def test_zipped_file_scan_async_bad_request_error (
105- scan_type : str , scan_client : ScanClient , api_token_response : responses .Response
104+ scan_type : ScanTypeOption , scan_client : ScanClient , api_token_response : responses .Response
106105) -> None :
107106 """Test handling of bad request errors in the async flow."""
108107 url , zip_file = zip_scan_resources (scan_type , scan_client )
@@ -113,7 +112,7 @@ def test_zipped_file_scan_async_bad_request_error(
113112 responses .add (api_token_response ) # mock token based client
114113 responses .add (method = responses .POST , url = url , status = expected_status_code , body = expected_response_text )
115114
116- with pytest .raises (RequestHttpError ) as e_info :
115+ with pytest .raises (CycodeError ) as e_info :
117116 scan_client .zipped_file_scan_async (zip_file = zip_file , scan_type = scan_type , scan_parameters = {})
118117
119118 assert e_info .value .status_code == expected_status_code
@@ -123,12 +122,11 @@ def test_zipped_file_scan_async_bad_request_error(
123122@pytest .mark .parametrize ('scan_type' , list (ScanTypeOption ))
124123@responses .activate
125124def test_zipped_file_scan_async_timeout_error (
126- scan_type : str , scan_client : ScanClient , api_token_response : responses .Response
125+ scan_type : ScanTypeOption , scan_client : ScanClient , api_token_response : responses .Response
127126) -> None :
128127 """Test handling of timeout errors in the async flow."""
129128 url , zip_file = zip_scan_resources (scan_type , scan_client )
130129
131- # Create a timeout response
132130 timeout_error = requests .exceptions .Timeout ('Connection timed out' )
133131
134132 responses .add (api_token_response ) # mock token based client
@@ -141,7 +139,7 @@ def test_zipped_file_scan_async_timeout_error(
141139@pytest .mark .parametrize ('scan_type' , list (ScanTypeOption ))
142140@responses .activate
143141def test_zipped_file_scan_async_connection_error (
144- scan_type : str , scan_client : ScanClient , api_token_response : responses .Response
142+ scan_type : ScanTypeOption , scan_client : ScanClient , api_token_response : responses .Response
145143) -> None :
146144 """Test handling of connection errors in the async flow."""
147145 url , zip_file = zip_scan_resources (scan_type , scan_client )
@@ -158,7 +156,9 @@ def test_zipped_file_scan_async_connection_error(
158156
159157@pytest .mark .parametrize ('scan_type' , list (ScanTypeOption ))
160158@responses .activate
161- def test_get_scan_details (scan_type : str , scan_client : ScanClient , api_token_response : responses .Response ) -> None :
159+ def test_get_scan_details (
160+ scan_type : ScanTypeOption , scan_client : ScanClient , api_token_response : responses .Response
161+ ) -> None :
162162 """Test getting scan details in the async flow."""
163163 scan_id = uuid4 ()
164164 url = get_scan_details_url (scan_type , scan_id , scan_client )
0 commit comments