def process_logs()

in deviceadvisor/script/DATestRun.py [0:0]

• 21 lines of code
• 3 McCabe index (conditional complexity)

def process_logs(log_group, log_stream, thing_name):
    logs_client = boto3.client('logs')
    response = logs_client.get_log_events(
        logGroupName=log_group,
        logStreamName=log_stream
    )
    log_file = "DA_Log_Javascript_" + thing_name + ".log"
    f = open(log_file, 'w')
    for event in response["events"]:
        f.write(event['message'])
    f.close()

    try:
        secrets_client = boto3.client(
            "secretsmanager", region_name=os.environ["AWS_DEFAULT_REGION"])
        s3_bucket_name = secrets_client.get_secret_value(SecretId="ci/DeviceAdvisor/s3bucket")["SecretString"]
        s3.Bucket(s3_bucket_name).upload_file(log_file, log_file)
        print("[Device Advisor] Device Advisor Log file uploaded to "+ log_file)

    except Exception:
        print ("[Device Advisor] Error: could not store log in S3 bucket!")

    os.remove(log_file)
    print("[Device Advisor] Device Advisor Log file uploaded to "+ log_file)