Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use aws_logger method in localstack

Best Python code snippet using localstack_python

log.py

Source:log.py

1# Copyright (c) 2017 David Preece, All rights reserved.2#3# Permission to use, copy, modify, and/or distribute this software for any4# purpose with or without fee is hereby granted.5#6# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES7# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF8# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR9# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES10# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN11# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF12# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.13"""AWS CloudWatch compatible logger - logs to stdout if not running under AWS"""14import json.decoder15import requests16import boto317import logging18import json19import time20import signal21import os22import botocore.errorfactory23from botocore.exceptions import ClientError, EndpointConnectionError24from threading import Thread25from queue import Queue, Empty26class LogHandler(logging.Handler):27    def __init__(self, group, stream, blacklist=None):28        try:29            # see if we are logging verbosely30            dynamic_data_text = requests.get('http://169.254.169.254/latest/dynamic/instance-identity/document').text31            if '404 Not Found' in dynamic_data_text:32                raise requests.exceptions.ConnectionError33            ssm = boto3.client('ssm', region_name=json.loads(dynamic_data_text)['region'])34            try:35                verbose_string = ssm.get_parameter(Name='/20ft/verbose')['Parameter']['Value']36            except ClientError:  # string not written yet37                verbose_string = 'False'38            level = logging.DEBUG if verbose_string == 'True' or verbose_string == 'true' else logging.INFO39            # find instance ID40            iid = requests.get('http://169.254.169.254/latest/meta-data/instance-id').text41            stream_name = stream + '/' + iid42        except requests.exceptions.ConnectionError:  # not running under AWS43            level = logging.INFO44            stream_name = None45        super().__init__(level)46        self.blacklist = blacklist if blacklist is not None else []47        self.formatter = logging.Formatter(fmt='%(levelname)-8s %(message)s')48        logging.basicConfig(level=level, handlers=[self])49        # delivery thread (aws only)50        self.queue = None51        if stream_name is not None:52            self.queue = Queue()53            self.thread = Thread(target=self.background, args=(group, stream_name, self.queue))54            self.thread.start()55    def emit(self, record):56        # urllib and boto try to log themselves doing all sorts57        if record.name.startswith('urllib3') or record.name.startswith('botocore'):58            return59        # is the log on the blacklist?60        text = self.formatter.format(record)61        for string in self.blacklist:62            if string in text:63                return64        # otherwise enqueue the record65        print(self.formatter.format(record))66        if self.queue is not None:67            self.queue.put(record)68    def stop(self):69        """Posts a message on the queue telling the logging thread to stop."""70        if self.queue is not None:71            self.queue.put(None)72    def background(self, group, stream, queue):73        """Runs as a background thread delivering the logs as they arrive (to avoid stalling the event loop)"""74        # create a cloud watch log client75        dynamic_data_text = requests.get('http://169.254.169.254/latest/dynamic/instance-identity/document').text76        aws_logger = boto3.client('logs', region_name=json.loads(dynamic_data_text)['region'])77        # create the group and stream if need be78        groups = aws_logger.describe_log_groups(79            logGroupNamePrefix=group80        )['logGroups']81        groups_dict = {g['logGroupName']: g for g in groups}82        if group not in groups_dict.keys():83            aws_logger.create_log_group(84                    logGroupName=group85            )86        stream_desc = aws_logger.describe_log_streams(87                logGroupName=group,88                logStreamNamePrefix=stream89        )90        sequence_token = '0'91        streams = {s['logStreamName']: s for s in stream_desc['logStreams']}92        if stream not in streams:93            aws_logger.create_log_stream(94                    logGroupName=group,95                    logStreamName=stream96            )97        else:98            sequence_token = streams[stream]['uploadSequenceToken']99        # loop picking logs off the queue and delivering100        running = True101        while running:102            log_events = []103            try:104                while True:105                    # fetch the next record, unless there isn't one for one second106                    record = queue.get(timeout=1)107                    # None causes the loop to exit cleanly but still deliver the queue of log events108                    if record is None:109                        running = False110                    else:111                        text = self.formatter.format(record)112                        log_events.append({'timestamp': int(record.created * 1000), 'message': text})113                    # if there are very many log events fake a timeout hit so they get delivered114                    if len(log_events) == 256:115                        raise Empty116            # timeout hit117            except Empty:118                pass  # timeout hit119            # loop back if nothing to deliver120            if len(log_events) == 0:121                continue122            # loop until delivered or bad things happen123            sent = False124            while not sent:125                try:126                    result = aws_logger.put_log_events(127                            logGroupName=group,128                            logStreamName=stream,129                            logEvents=log_events,130                            sequenceToken=sequence_token131                    )132                    sequence_token = result['nextSequenceToken']133                    sent = True134                except ClientError as e:135                    print("....LogHandler error: " + str(e))136                    time.sleep(2)137                except EndpointConnectionError:138                    print("...Name resolution has failed")139                    self.queue.put(None)140                except BaseException as e:141                    print("...A bad thing happened with the logging thread: " + str(e))...

aws_mailer.py

Source:aws_mailer.py

1import boto32from botocore.exceptions import ClientError3from content import html, subject, text4from dotenv import dotenv_values5from logger_config import configLogger6import csv7aws_logger = configLogger(__name__)8config = dotenv_values(dotenv_path=".env")9try:10    AWS_REGION = config["AWS_REGION"]11    FROM_NAME = config["FROM_NAME"]12    FROM_EMAIL = config["FROM_EMAIL"]13    DOMAIN = config["DOMAIN"]14except:15    aws_logger.exception("Improperly Configured Environment")16    exit({"error": "Improperly Configured Environment"})17SENDER = f"{FROM_NAME} <{FROM_EMAIL}@{DOMAIN}>"18CHARSET = "UTF-8"19client = boto3.client('ses', region_name=AWS_REGION)20def sendEmail(email):21    email = email.strip()22    try:23        response = client.send_email(24            Destination={25                'ToAddresses': [26                    email,27                ],28            },29            Message={30                'Body': {31                    'Html': {32                        'Charset': CHARSET,33                        'Data': html,34                    },35                    'Text': {36                        'Charset': CHARSET,37                        'Data': text,38                    },39                },40                'Subject': {41                    'Charset': CHARSET,42                    'Data': subject,43                },44            },45            Source=SENDER,46        )47    except ClientError as e:48        aws_logger.error(f"Email not sent ({email})")49        print({e.response['Error']['Message']})50        with open("unsent.csv", "a") as unsent_csv:51            csv_w = csv.writer(unsent_csv)52            csv_w.writerow([email])53    else:54        aws_logger.debug(f"Email - {email}")55        aws_logger.debug(f"Sent - Message ID = {response['MessageId']}")56# only to make sure that the code is working57def main():58    sendEmail("example@example.com")59if __name__ == "__main__":...

test_logging.py

Source:test_logging.py

1from unittest import TestCase2from models import setuplogging3class MockAWSLogging():4    @staticmethod5    def info(message: str) -> None:6        assert message is not None7class TestLogging(TestCase):8    def test_setup_logging_mocked(self):9        setuplogging.AWS_LOGGER = None10        setuplogging.LOGGING_HANDLER = None11        setuplogging.initialize_logging(mocking=True)12        assert setuplogging.AWS_LOGGER is None13        assert setuplogging.LOGGING_HANDLER == setuplogging.mock_logging_handler14    def test_setup_logging_prod(self):15        setuplogging.AWS_LOGGER = None16        setuplogging.LOGGING_HANDLER = None17        setuplogging.initialize_logging(mocking=False)18        assert setuplogging.AWS_LOGGER is not None19        assert setuplogging.LOGGING_HANDLER == setuplogging.prod_logging_handler20    def test_logging_to_prod(self):21        """In this test we do all the prod setup but we change22        out where PROD logs to and log to a locally defined function"""23        setuplogging.AWS_LOGGER = None24        setuplogging.LOGGING_HANDLER = None25        setuplogging.initialize_logging(mocking=False)26        setuplogging.AWS_LOGGER = MockAWSLogging()27        string_to_log = "log this string"28        setuplogging.LOGGING_HANDLER(string_to_log)29    def test_logging_to_mocked(self):30        """Our mocked log is just a string, so make sure we stash it in the right place"""31        setuplogging.AWS_LOGGER = None32        setuplogging.LOGGING_HANDLER = None33        setuplogging.initialize_logging(mocking=True)34        string_to_log = "log this string"35        setuplogging.LOGGING_HANDLER(string_to_log)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.