Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use describe_log_groups method in localstack

Best Python code snippet using localstack_python

test_aws_cloud.py

Source:test_aws_cloud.py

1#!/usr/bin/env python32# Copyright (c) Meta Platforms, Inc. and affiliates.3#4# This source code is licensed under the MIT license found in the5# LICENSE file in the root directory of this source tree.6# pyre-strict7import os8import unittest9from pathlib import Path10from unittest.mock import patch11from botocore.exceptions import ClientError12from fbpcs.infra.logging_service.download_logs.cloud.aws_cloud import AwsCloud13class TestAwsCloud(unittest.TestCase):14    def setUp(self) -> None:15        self.test_dir = Path(os.path.dirname(__file__))16        self.tag = "my_tag"17        with patch(18            "fbpcs.infra.logging_service.download_logs.cloud.aws_cloud.boto3"19        ), patch("fbpcs.infra.logging_service.download_logs.download_logs.Utils"):20            self.aws_container_logs = AwsCloud(self.tag)21    ##############################22    # Tests for public interface #23    ##############################24    def test_get_cloudwatch_logs(self) -> None:25        self.aws_container_logs.cloudwatch_client.get_log_events.side_effect = [26            {"events": [{"message": "123"}], "nextForwardToken": "1"},27            {"events": [{"message": "456"}], "nextForwardToken": "2"},28            {"events": [{"message": "789"}], "nextForwardToken": "3"},29            # Repeated event indicates no more data available30            {"events": [{"message": "789"}], "nextForwardToken": "3"},31        ]32        expected = ["123", "456", "789"]33        with self.subTest("basic"):34            self.assertEqual(35                expected,36                self.aws_container_logs.get_cloudwatch_logs("foo", "bar"),37            )38            # NOTE: we don't want to get *too* specific with these asserts39            # because we want to allow the internal details to change and40            # still meet the API requirements41            self.aws_container_logs.cloudwatch_client.get_log_events.assert_called()42        ####################43        # Test error cases #44        ####################45        error_cases = [46            ("InvalidParameterException", "Couldn't fetch.*"),47            ("ResourceNotFoundException", "Couldn't find.*"),48            ("SomethingElseHappenedException", "Unexpected error.*"),49        ]50        for error_code, exc_regex in error_cases:51            with self.subTest(f"get_log_events.{error_code}"):52                self.aws_container_logs.cloudwatch_client.get_log_events.reset_mock()53                self.aws_container_logs.cloudwatch_client.get_log_events.side_effect = (54                    ClientError(55                        error_response={"Error": {"Code": error_code}},56                        operation_name="get_log_events",57                    )58                )59                with self.assertRaisesRegex(Exception, exc_regex):60                    self.aws_container_logs.get_cloudwatch_logs("foo", "bar")61                    self.aws_container_logs.cloudwatch_client.get_log_events.assert_called()62    def test_create_s3_folder(self) -> None:63        self.aws_container_logs.s3_client.put_object.return_value = {64            "ResponseMetadata": {"HTTPStatusCode": 200}65        }66        with self.subTest("basic"):67            self.assertIsNone(68                self.aws_container_logs.create_s3_folder("bucket", "folder")69            )70            self.aws_container_logs.s3_client.put_object.assert_called_once_with(71                Bucket="bucket", Key="folder"72            )73        with self.subTest("put_object.Http403"):74            self.aws_container_logs.s3_client.put_object.reset_mock()75            self.aws_container_logs.s3_client.put_object.return_value = {76                "ResponseMetadata": {"HTTPStatusCode": 403}77            }78            with self.assertRaisesRegex(Exception, "Failed to create.*"):79                self.aws_container_logs.create_s3_folder("bucket", "folder")80    def test_parse_log_events(self) -> None:81        events = [82            {"message": "hello", "code": 200, "other": "ignore"},83            {"message": "world", "code": 200, "other": "ignore"},84        ]85        expected = ["hello", "world"]86        with self.subTest("basic"):87            self.assertEqual(88                expected, self.aws_container_logs._parse_log_events(events)89            )90    def test_get_s3_folder_contents(self) -> None:91        expected = {"ContinuationToken": "abc123", "Contents": ["a", "b", "c"]}92        self.aws_container_logs.s3_client.list_objects_v2.return_value = expected93        with self.subTest("basic"):94            self.assertEqual(95                expected,96                self.aws_container_logs.get_s3_folder_contents("bucket", "folder"),97            )98        # Check that continuation token is set99        with self.subTest("with_continuation_token"):100            self.aws_container_logs.s3_client.list_objects_v2.reset_mock()101            self.aws_container_logs.s3_client.list_objects_v2.return_value = expected102            self.assertEqual(103                expected,104                self.aws_container_logs.get_s3_folder_contents(105                    bucket_name="bucket",106                    folder_name="folder",107                    next_continuation_token="def678",108                ),109            )110            self.aws_container_logs.s3_client.list_objects_v2.assert_called_once_with(111                Bucket="bucket",112                Prefix="folder",113                MaxKeys=1,114                ContinuationToken="def678",115            )116        # check exception cases117        with self.subTest("list_objects_v2.InvalidParameterException"):118            self.aws_container_logs.s3_client.list_objects_v2.reset_mock()119            self.aws_container_logs.s3_client.list_objects_v2.side_effect = ClientError(120                error_response={"Error": {"Code": "InvalidParameterException"}},121                operation_name="list_objects_v2",122            )123            with self.assertRaisesRegex(Exception, "Couldn't find folder.*"):124                self.aws_container_logs.get_s3_folder_contents("bucket", "folder")125    def test_verify_log_group(self) -> None:126        self.aws_container_logs.cloudwatch_client.describe_log_groups.return_value = {127            "logGroups": ["my_log_group"]128        }129        with self.subTest("basic"):130            self.assertTrue(self.aws_container_logs._verify_log_group("my_log_group"))131        with self.subTest("describe_log_groups.InvalidParameterException"):132            self.aws_container_logs.cloudwatch_client.describe_log_groups.reset_mock()133            self.aws_container_logs.cloudwatch_client.describe_log_groups.side_effect = ClientError(134                error_response={"Error": {"Code": "InvalidParameterException"}},135                operation_name="describe_log_groups",136            )137            with self.assertRaisesRegex(Exception, "Wrong parameters.*"):138                self.aws_container_logs._verify_log_group("my_log_group")139        with self.subTest("describe_log_groups.ResourceNotFoundException"):140            self.aws_container_logs.cloudwatch_client.describe_log_groups.reset_mock()141            self.aws_container_logs.cloudwatch_client.describe_log_groups.side_effect = ClientError(142                error_response={"Error": {"Code": "ResourceNotFoundException"}},143                operation_name="describe_log_groups",144            )145            with self.assertRaisesRegex(Exception, "Couldn't find.*"):146                self.aws_container_logs._verify_log_group("my_log_group")147        with self.subTest("describe_log_groups.SomethingElseHappenedException"):148            self.aws_container_logs.cloudwatch_client.describe_log_groups.reset_mock()149            self.aws_container_logs.cloudwatch_client.describe_log_groups.side_effect = ClientError(150                error_response={"Error": {"Code": "SomethingElseHappenedException"}},151                operation_name="describe_log_groups",152            )153            with self.assertRaisesRegex(Exception, "Unexpected error.*"):154                self.aws_container_logs._verify_log_group("my_log_group")155    def test_verify_log_stream(self) -> None:156        self.aws_container_logs.cloudwatch_client.describe_log_streams.return_value = {157            "logStreams": ["my_log_stream"]158        }159        with self.subTest("basic"):160            self.assertTrue(161                self.aws_container_logs._verify_log_stream(162                    "my_log_group", "my_log_stream"163                )164            )165        with self.subTest("describe_log_streams.InvalidParameterException"):166            self.aws_container_logs.cloudwatch_client.describe_log_streams.reset_mock()167            self.aws_container_logs.cloudwatch_client.describe_log_streams.side_effect = ClientError(168                error_response={"Error": {"Code": "InvalidParameterException"}},169                operation_name="describe_log_streams",170            )171            with self.assertRaisesRegex(Exception, "Wrong parameters.*"):172                self.aws_container_logs._verify_log_stream(173                    "my_log_group", "my_log_stream"174                )175        with self.subTest("describe_log_streams.ResourceNotFoundException"):176            self.aws_container_logs.cloudwatch_client.describe_log_streams.reset_mock()177            self.aws_container_logs.cloudwatch_client.describe_log_streams.side_effect = ClientError(178                error_response={"Error": {"Code": "ResourceNotFoundException"}},179                operation_name="describe_log_streams",180            )181            with self.assertRaisesRegex(Exception, "Couldn't find.*"):182                self.aws_container_logs._verify_log_stream(183                    "my_log_group", "my_log_stream"184                )185        with self.subTest("describe_log_streams.SomethingElseHappenedException"):186            self.aws_container_logs.cloudwatch_client.describe_log_streams.reset_mock()187            self.aws_container_logs.cloudwatch_client.describe_log_streams.side_effect = ClientError(188                error_response={"Error": {"Code": "SomethingElseHappenedException"}},189                operation_name="describe_log_streams",190            )191            with self.assertRaisesRegex(Exception, "Unexpected error.*"):192                self.aws_container_logs._verify_log_stream(193                    "my_log_group", "my_log_stream"...

cloudwatch_log.py

Source:cloudwatch_log.py

...67    try:68        return client.delete_log_group(logGroupName=name)69    except botocore.exceptions.ClientError as e:70        module.fail_json(msg=str(e))71def describe_log_groups(client, module, filter_prefix):72    try:73        return client.describe_log_groups(logGroupNamePrefix=filter_prefix,74                                          limit=50)75    except botocore.exceptions.ClientError as e:76        module.fail_json(msg=str(e))77def list_groups(client, module):78    changed = False79    result = None80    filter_prefix = module.params.get('filter_prefix')81    result = describe_log_groups(client, module, filter_prefix)82    return changed, result83def setup(client, module):84    changed = False85    result = None86    name = module.params.get('log_group_name')87    filter_prefix = module.params.get('filter_prefix')88    log_groups = describe_log_groups(client, module, filter_prefix)89    if not log_groups['logGroups']:90        changed = True91        result = create_log_group(client, name)92    else:93        result = log_groups94    return changed, result95def teardown(client, module):96    changed = False97    result = None98    name = module.params.get('log_group_name')99    filter_prefix = module.params.get('filter_prefix')100    log_groups = describe_log_groups(client, module, filter_prefix)101    if log_groups['logGroups']:102        result = delete_log_group(client, module, name)103        changed = True104    return changed, result105def main():106    argument_spec = ec2_argument_spec()107    argument_spec.update(dict(108        log_group_name=dict(default=None, required=False),109        filter_prefix=dict(default=None, required=False),110        state=dict(default='present', choices=['present', 'absent', 'list'])111        )112    )113    module = AnsibleModule(argument_spec=argument_spec)114    state = module.params.get('state').lower()...

lambda_logs_collector.py

Source:lambda_logs_collector.py

1import boto32from terraform_external_data import terraform_external_data3client = boto3.client('logs')4def create_new_response(next_token):5    response = client.describe_log_groups(6        logGroupNamePrefix='/aws/lambda',7        nextToken=next_token,8        limit=509    )10    return response11@terraform_external_data12def get_log_group_names(query):13    response = client.describe_log_groups(14        logGroupNamePrefix='/aws/lambda',15        limit=5016    )17    rules = [18        # Add here the prefix of the lambdas you want to filter19    ]20    filter_flag = False21    flag = True22    log_group_names_map = ""23    while flag == True:24        log_group_list = response.get('logGroups')25        cont = 026        for group in log_group_list:27            log_group_name = group.get('logGroupName')28            first_split = log_group_name.split('-')29            second_split = first_split[0].split('/aws/lambda/')30            31            # Only filter when the filter_flag is True, and will consume de rules32            if filter_flag and second_split[1] in rules:33                group_name = group.get('logGroupName')34                log_group_names_map += f"{group_name}"35                if cont < len(log_group_list):36                    log_group_names_map += ","37                    cont += 138            else:39                group_name = group.get('logGroupName')40                log_group_names_map += f"{group_name}"41                if cont < len(log_group_list):42                    log_group_names_map += ","43                    cont += 144        nextToken = response.get('nextToken')45        if nextToken is None:46            flag = False47            response = client.describe_log_groups(48                logGroupNamePrefix='/aws/lambda',49                limit=5050            )51            comma = log_group_names_map[:-1]52            log_group_names_map = comma53        else:54            response = create_new_response(nextToken)55    return {query['group_names_list']: log_group_names_map}56if __name__ == "__main__":...

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.