7ecf2541-e446-4544-a68f-78cc8fdb3252

Committed 04 Oct 2023 08:10AM CUT coverage: 95.298%. Remained the same

Build # 7ecf2541-e446-4544-a68f-78cc8fdb3252

Build Type

push

circleci

Committed by

github-actions[bot]

Commit Message

docs: Added README."fr".md translation via https://github.com/dephraiim/translate-readme

Run Details

527 of 553 relevant lines covered (95.3%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.77

/src/common/python/utility.py

import math
import constants
import custom_exceptions
import boto3

def eval_exception(error, identifier, taskname):
    result = {}
    error_message = constants.IDENTIFIER + identifier + ' \n' + str(error)
    if taskname == constants.DELETE and \
         (str(error) == constants.CLUSTER_UNAVAILABLE or
          constants.CLUSTER_NOT_FOUND in str(error) or constants.NOT_FOUND in str(error)):
        result['task'] = constants.TASK_COMPLETE
        result['identifier'] = identifier
        result['taskname'] = taskname
        return result
    elif str(error) == constants.INSTANCE_UNAVAILABLE or constants.INSTANCE_NOT_FOUND in str(error) \
            or str(error) == constants.CLUSTER_UNAVAILABLE or constants.CLUSTER_NOT_FOUND in str(error) \
            or constants.NOT_FOUND in str(error):
        raise custom_exceptions.InstanceUnavailableException(error_message)
    elif constants.RATE_EXCEEDED in str(error) or constants.WAITER_MAX in str(error):
        raise custom_exceptions.RateExceededException(error_message)
    elif constants.WAITER_FAILURE in str(error):
        result['task'] = constants.TASK_FAILED
        result['status'] = error_message
        result['identifier'] = identifier
        result['taskname'] = taskname
        return result
    elif taskname in constants.TASK_ERROR_MAP:
        raise constants.TASK_ERROR_MAP[taskname](error_message)
    else:
        raise Exception(error_message)

def get_identifier_from_error(event):
    response = {}
    start_index = event.get('Cause').find(constants.IDENTIFIER) + len(constants.IDENTIFIER)
    end_index = start_index + (event.get('Cause')[start_index:]).find(" ")
    response["modified_identifier"] = event.get('Cause')[(start_index):(end_index)]
    response["original_identifier"] = response["modified_identifier"] + constants.TEMP_POSTFIX
    return response

def get_modified_identifier(identifier):
    response = {}
    temp_substr = identifier.rfind(constants.TEMP_POSTFIX)
    db_instance_id = identifier[0:temp_substr] \
        if temp_substr > 0 \
        else identifier
    db_snapshot_id = db_instance_id + constants.SNAPSHOT_POSTFIX
    response["instance_id"] = db_instance_id
    response["snapshot_id"] = db_snapshot_id
    return response

def eval_snapshot_exception(error, identifier, rds_client):
    error_message = constants.IDENTIFIER + identifier + ' \n' + str(error)
    snapshot_id = identifier + constants.SNAPSHOT_POSTFIX
    if constants.RATE_EXCEEDED in str(error):
        raise custom_exceptions.RateExceededException(error_message)
    elif constants.CLUSTER_SNAPSHOT_EXISTS in str(error):
        waiter = rds_client.get_waiter('db_cluster_snapshot_deleted')
        rds_client.delete_db_cluster_snapshot(
            DBClusterSnapshotIdentifier = snapshot_id
        )
        waiter.wait(DBClusterSnapshotIdentifier = snapshot_id)
        raise custom_exceptions.RetryClusterSnapshotException(error_message)
    elif constants.DB_SNAPSHOT_EXISTS in str(error):
        waiter = rds_client.get_waiter('db_snapshot_deleted')
        rds_client.delete_db_snapshot(
            DBSnapshotIdentifier = snapshot_id
        )
        waiter.wait(DBSnapshotIdentifier = snapshot_id)
        raise custom_exceptions.RetryDBSnapshotException(error_message)
    else:
        raise custom_exceptions.SnapshotCreationException(error_message)

def get_waiter_max_attempts(context):
    # converting remaining time from milliseconds to seconds
    remaining_time = context.get_remaining_time_in_millis() / constants.DIVISOR
    # calculating max attempts for waiter task
    max_attempts = math.floor(abs(remaining_time - constants.DELAY_INTERVAL) / constants.DELAY_INTERVAL)
    return max_attempts

def get_error_message(identifier, error):
    error_message = constants.IDENTIFIER + identifier + ' \n' + str(error)
    return error_message

def get_aws_account_id():
    return boto3.client('sts').get_caller_identity().get('Account')

1	import math	1✔
2	import constants	1✔
3	import custom_exceptions	1✔
4	import boto3	1✔
5
6	def eval_exception(error, identifier, taskname):	1✔
7	result = {}	1✔
8	error_message = constants.IDENTIFIER + identifier + ' \n' + str(error)	1✔
9	if taskname == constants.DELETE and \	1✔
10	(str(error) == constants.CLUSTER_UNAVAILABLE or
11	constants.CLUSTER_NOT_FOUND in str(error) or constants.NOT_FOUND in str(error)):
12	result['task'] = constants.TASK_COMPLETE	×
13	result['identifier'] = identifier	×
14	result['taskname'] = taskname	×
15	return result	×
16	elif str(error) == constants.INSTANCE_UNAVAILABLE or constants.INSTANCE_NOT_FOUND in str(error) \	1✔
17	or str(error) == constants.CLUSTER_UNAVAILABLE or constants.CLUSTER_NOT_FOUND in str(error) \
18	or constants.NOT_FOUND in str(error):
19	raise custom_exceptions.InstanceUnavailableException(error_message)	1✔
20	elif constants.RATE_EXCEEDED in str(error) or constants.WAITER_MAX in str(error):	1✔
21	raise custom_exceptions.RateExceededException(error_message)	1✔
22	elif constants.WAITER_FAILURE in str(error):	1✔
23	result['task'] = constants.TASK_FAILED	1✔
24	result['status'] = error_message	1✔
25	result['identifier'] = identifier	1✔
26	result['taskname'] = taskname	1✔
27	return result	1✔
28	elif taskname in constants.TASK_ERROR_MAP:	1✔
29	raise constants.TASK_ERROR_MAP[taskname](error_message)	1✔
30	else:
31	raise Exception(error_message)	×
32
33	def get_identifier_from_error(event):	1✔
34	response = {}	1✔
35	start_index = event.get('Cause').find(constants.IDENTIFIER) + len(constants.IDENTIFIER)	1✔
36	end_index = start_index + (event.get('Cause')[start_index:]).find(" ")	1✔
37	response["modified_identifier"] = event.get('Cause')[(start_index):(end_index)]	1✔
38	response["original_identifier"] = response["modified_identifier"] + constants.TEMP_POSTFIX	1✔
39	return response	1✔
40
41	def get_modified_identifier(identifier):	1✔
42	response = {}	1✔
43	temp_substr = identifier.rfind(constants.TEMP_POSTFIX)	1✔
44	db_instance_id = identifier[0:temp_substr] \	1✔
45	if temp_substr > 0 \
46	else identifier
47	db_snapshot_id = db_instance_id + constants.SNAPSHOT_POSTFIX	1✔
48	response["instance_id"] = db_instance_id	1✔
49	response["snapshot_id"] = db_snapshot_id	1✔
50	return response	1✔
51
52	def eval_snapshot_exception(error, identifier, rds_client):	1✔
53	error_message = constants.IDENTIFIER + identifier + ' \n' + str(error)	1✔
54	snapshot_id = identifier + constants.SNAPSHOT_POSTFIX	1✔
55	if constants.RATE_EXCEEDED in str(error):	1✔
56	raise custom_exceptions.RateExceededException(error_message)	1✔
57	elif constants.CLUSTER_SNAPSHOT_EXISTS in str(error):	1✔
58	waiter = rds_client.get_waiter('db_cluster_snapshot_deleted')	1✔
59	rds_client.delete_db_cluster_snapshot(	1✔
60	DBClusterSnapshotIdentifier = snapshot_id
61	)
62	waiter.wait(DBClusterSnapshotIdentifier = snapshot_id)	1✔
63	raise custom_exceptions.RetryClusterSnapshotException(error_message)	1✔
64	elif constants.DB_SNAPSHOT_EXISTS in str(error):	1✔
65	waiter = rds_client.get_waiter('db_snapshot_deleted')	1✔
66	rds_client.delete_db_snapshot(	1✔
67	DBSnapshotIdentifier = snapshot_id
68	)
69	waiter.wait(DBSnapshotIdentifier = snapshot_id)	1✔
70	raise custom_exceptions.RetryDBSnapshotException(error_message)	1✔
71	else:
72	raise custom_exceptions.SnapshotCreationException(error_message)	1✔
73
74	def get_waiter_max_attempts(context):	1✔
75	# converting remaining time from milliseconds to seconds
76	remaining_time = context.get_remaining_time_in_millis() / constants.DIVISOR	1✔
77	# calculating max attempts for waiter task
78	max_attempts = math.floor(abs(remaining_time - constants.DELAY_INTERVAL) / constants.DELAY_INTERVAL)	1✔
79	return max_attempts	1✔
80
81	def get_error_message(identifier, error):	1✔
82	error_message = constants.IDENTIFIER + identifier + ' \n' + str(error)	1✔
83	return error_message	1✔
84
85	def get_aws_account_id():	1✔
86	return boto3.client('sts').get_caller_identity().get('Account')	×

intuit / Trapheus / 7ecf2541-e446-4544-a68f-78cc8fdb3252

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous