Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_update_contents_does_preflight_check_if_specified(
test_file,
mock_file_path,
preflight_check,
file_size,
preflight_fails,
mock_box_session,
):
with patch.object(File, 'preflight_check', return_value=None):
kwargs = {'file_path': mock_file_path}
if preflight_check:
kwargs['preflight_check'] = preflight_check
kwargs['preflight_expected_size'] = file_size
if preflight_fails:
test_file.preflight_check.side_effect = BoxAPIException(400)
with pytest.raises(BoxAPIException):
test_file.update_contents(**kwargs)
else:
test_file.update_contents(**kwargs)
if preflight_check:
assert test_file.preflight_check.called_once_with(size=file_size)
if preflight_fails:
assert not mock_box_session.post.called
else:
assert mock_box_session.post.called
else:
assert not test_file.preflight_check.called
mock_new_upload_accelerator_url,
upload_using_accelerator_fails,
is_stream,
etag,
sha1,
if_match_sha1_header,
):
# pylint:disable=too-many-locals
file_description = 'Test File Description'
content_created_at = '1970-01-01T00:00:00+00:00'
content_modified_at = '1970-01-01T11:11:11+11:11'
additional_attributes = {'attr': 123}
expected_url = '{0}/files/content'.format(API.UPLOAD_URL)
if upload_using_accelerator:
if upload_using_accelerator_fails:
mock_box_session.options.side_effect = BoxAPIException(400)
else:
mock_box_session.options.return_value = mock_accelerator_response_for_new_uploads
expected_url = mock_new_upload_accelerator_url
mock_box_session.post.return_value = mock_upload_response
if is_stream:
mock_file_stream = BytesIO(mock_content_response.content)
new_file = test_folder.upload_stream(
mock_file_stream,
basename(mock_file_path),
file_description,
upload_using_accelerator=upload_using_accelerator,
content_created_at=content_created_at,
content_modified_at=content_modified_at,
additional_attributes=additional_attributes,
'sha1': third_sha1,
}
part_four = {
'part_id': '4DBC872D',
'offset': 6,
'size': 1,
'sha1': fourth_sha1,
}
third_response_mock.json.return_value = {
'part': part_three
}
fourth_response_mock.json.return_value = {
'part': part_four
}
chunked_uploader = ChunkedUploader(first_upload_session_mock_object, stream, file_size)
first_upload_session_mock_object.upload_part_bytes.side_effect = [uploaded_part_mock_object, BoxAPIException(502)]
second_chunked_uploader.upload_session = second_upload_session_mock_object
second_upload_session_mock_object.upload_part_bytes.side_effect = [third_response_mock, fourth_response_mock]
second_upload_session_mock_object.get_parts.return_value = {
'entries': [
{
'part_id': 'CFEB4BA9',
'offset': 0,
'size': 2,
'sha1': None,
},
{
'part_id': '4DBB872D',
'offset': 2,
'size': 2,
'sha1': None,
},
def test_get_item_info_for_missing_file(box_client, constructor):
with pytest.raises(BoxAPIException) as exc_info:
constructor(box_client, '1').delete()
assert exc_info.value.status == 404
def oauth_dance_retry(oauth_instance, cache_client, oauth_meta_info, conf_obj, bottle_app, file_event_handler, oauth_lock_instance, bottle_thread=None):
with oauth_lock_instance:
temp_client = Client(oauth_instance)
while True:
try:
temp_client.auth._access_token = cache_client.get('diy_crate.auth.access_token')
temp_client.auth._refresh_token = cache_client.get('diy_crate.auth.refresh_token')
if temp_client.auth._access_token:
temp_client.auth._access_token = temp_client.auth._access_token.decode(encoding='utf-8') if isinstance(temp_client.auth._access_token, bytes) else temp_client.auth._access_token
if temp_client.auth._refresh_token:
temp_client.auth._refresh_token = temp_client.auth._refresh_token.decode(encoding='utf-8') if isinstance(temp_client.auth._refresh_token, bytes) else temp_client.auth._refresh_token
temp_client.folder(folder_id='0').get()
break # sweet, we should have valid oauth access, now
except (exception.BoxAPIException, AttributeError):
crate_logger.warning("dance api exc or attr error", exc_info=True)
try:
oauth_meta_info['diy_crate.auth.oauth_dance_retry'] = True
try:
get_access_token(cache_client.get(u'diy_crate.auth.access_token'))
crate_logger.info(u'got new access and refresh tokens')
# might as well get a new set of tokens
except Exception:
cache_client.delete('diy_crate.auth.access_token', 'diy_crate.auth.refresh_token',)
oauth_dance(cache_client, conf_obj, bottle_app, file_event_handler, bottle_thread=bottle_thread)
# wait until the oauth dance has completed
while not (cache_client.get('diy_crate.auth.access_token') and cache_client.get('diy_crate.auth.refresh_token')):
time.sleep(15)
crate_logger.info('Just slept, waiting on the oauth dance')
finally:
:param network_response:
The network response which is being tested for success.
:type network_response:
:class:`NetworkResponse`
:param request:
The API request that could be unsuccessful.
:type request:
:class:`BoxRequest`
"""
if not network_response.ok:
response_json = {}
try:
response_json = network_response.json()
except ValueError:
pass
raise BoxAPIException(
status=network_response.status_code,
headers=network_response.headers,
code=response_json.get('code', None) or response_json.get('error', None),
message=response_json.get('message', None) or response_json.get('error_description', None),
request_id=response_json.get('request_id', None),
url=request.url,
method=request.method,
context_info=response_json.get('context_info', None),
network_response=network_response
)
if request.expect_json_response and not is_json_response(network_response):
raise BoxAPIException(
status=network_response.status_code,
headers=network_response.headers,
message='Non-json response received, while expecting json response.',
url=request.url,
if e.status == 404:
crate_logger.debug('Box says: {obj_id}, '
'{obj_name}, is a 404 status.'.format(obj_id=box_item['id'],
obj_name=box_item[
'name']))
crate_logger.debug(
'But, this is a folder, we do not handle recursive folder deletes correctly yet.')
break
except (ConnectionError, ConnectionResetError, BrokenPipeError):
crate_logger.debug('Attempt {idx}/{limit}; {the_trace}'.format(the_trace=traceback.format_exc(),
idx=i+1, limit=retry_limit))
else:
try:
file_obj = box_item
download_queue.put((file_obj, os.path.join(path, box_item['name']), oauth_obj))
except BoxAPIException as e:
crate_logger.debug(traceback.format_exc())
if e.status == 404:
crate_logger.debug('Box says: {obj_id}, {obj_name}, '
'is a 404 status.'.format(obj_id=box_item['id'], obj_name=box_item['name']))
if r_c.exists(redis_key(box_item['id'])):
crate_logger.debug('Deleting {obj_id}, '
'{obj_name}'.format(obj_id=box_item['id'], obj_name=box_item['name']))
r_c.delete(redis_key(box_item['id']))
redis_set(cache_client=r_c, cloud_item=b_folder, last_modified_time=os.path.getmtime(path),
box_dir_path=BOX_DIR, fresh_download=not r_c.exists(redis_key(box_folder['id'])),
folder=os.path.dirname(path),
sub_ids=ids_in_folder, parent_id=p_id)
response_json = network_response.json()
except ValueError:
pass
raise BoxAPIException(
status=network_response.status_code,
headers=network_response.headers,
code=response_json.get('code', None) or response_json.get('error', None),
message=response_json.get('message', None) or response_json.get('error_description', None),
request_id=response_json.get('request_id', None),
url=request.url,
method=request.method,
context_info=response_json.get('context_info', None),
network_response=network_response
)
if request.expect_json_response and not is_json_response(network_response):
raise BoxAPIException(
status=network_response.status_code,
headers=network_response.headers,
message='Non-json response received, while expecting json response.',
url=request.url,
method=request.method,
network_response=network_response,
)
def get_box_folder(client, cur_box_folder, folder_id, retry_limit):
"""
:param client:
:param cur_box_folder:
:param folder_id:
:param retry_limit:
:return:
"""
for i in range(retry_limit):
try:
box_folder = client.folder(folder_id=folder_id).get()
cur_box_folder = box_folder
break
except (ConnectionError, BrokenPipeError, ProtocolError, ConnectionResetError, BoxAPIException):
if i + 1 >= retry_limit:
crate_logger.warn('Attempt ({retry_count}) out of ({max_count}); Going to give '
'up on the write event because: {trace}'.format(retry_count=i,
max_count=retry_limit,
trace=traceback.format_exc()))
else:
crate_logger.warn('Attempt ({retry_count}) '
'out of ({max_count}): {trace}'.format(retry_count=i,
max_count=retry_limit,
trace=traceback.format_exc()))
return cur_box_folder
def credentials_are_valid(self, user_settings, client):
from addons.box.models import Provider as Box # Avoid circular import
if self.node_settings.has_auth:
if Box(self.node_settings.external_account).refresh_oauth_key():
return True
if user_settings:
oauth = OAuth2(client_id=settings.BOX_KEY, client_secret=settings.BOX_SECRET, access_token=user_settings.external_accounts[0].oauth_key)
client = client or Client(oauth)
try:
client.user()
except (BoxAPIException, IndexError):
return False
return True