Skip to content

Instantly share code, notes, and snippets.

@adhorn
Forked from kapilt/fault.py
Created July 24, 2019 16:04
Show Gist options
  • Save adhorn/22e8d79c205d2f1918b9559f6b10d7af to your computer and use it in GitHub Desktop.
Save adhorn/22e8d79c205d2f1918b9559f6b10d7af to your computer and use it in GitHub Desktop.
import boto3
import random
def getconfig(key):
return {
'api-delay-services': ['ec2'],
'api-delay-range': [0, 3],
'api-delay-rate': 10,
'aws-fault-services': ['s3', 'dynamodb'],
'api-fault-rate': 10
}.get(key)
class BotoFaultInjector(object):
def __init__(self):
self._regid = None
self.delay_services = getconfig('api-delay-services')
self.delay_range = getconfig('api-delay-range')
self.delay_rate = getconfig('api-delay-rate')
self.fault_service = getconfig('api-fault-services')
self.fault_rate = getconfig('api-fault-rate')
def attach(self, s):
self._regid = s.register('after-call.*.*', self.invoke)
def invoke(self, http_response, parsed, model, **kwargs):
if model.name in delay_services:
if random.randrange(0, 100) < self.delay_rate:
time.sleep(random.randrange(*self.delay_range))
if model.name in fault_services:
if random.rangrange(0, 100) < self.delay_rate:
http_response["Error"] = {
"Message": "Not Found",
"Code": "ResourceNotFoundException"
}
http_response['status_code'] = 400
if __name__ == '__main__':
boto3.client('ec2')
fault_injector = BotoFaultInjector()
fault_injector.attach(boto3.DEFAULT_SESSION)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment