Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use delete_metric_filter method in localstack

Best Python code snippet using localstack_python

cloudwatch_metric_filter.py

Source:cloudwatch_metric_filter.py

...91        )92        module.exit_json(msg="Create Metric Filter operation complete.",changed=True)93    except BotoServerError as e:94        module.fail_json(msg=str(e))95def delete_metric_filter(connection, module):96    log_group_name = module.params.get('log_group_name')97    filter_name = module.params.get('filter_name')98    filters = connection.describe_metric_filters(99                  logGroupName=log_group_name,100                  filterNamePrefix=filter_name101              )102    if filters:103        try:104            connection.delete_metric_filter(105                logGroupName=log_group_name,106                filterName=filter_name107            )108            module.exit_json(changed=True)109        except BotoServerError as e:110            module.fail_json(msg=str(e))111    else:112        module.exit_json(changed=False)113def main():114    argument_spec = ec2_argument_spec()115    argument_spec.update(116        dict(117            log_group_name=dict(required=True, type='str'),118            filter_name=dict(required=True, type='str'),119            filter_pattern=dict(required=True, type='str'),120            metric_transformations=dict(required=True, type='list'),121            state=dict(default='present', choices=['present', 'absent']),122            )123    )124    module = AnsibleModule(argument_spec=argument_spec)125    if not HAS_BOTO3:126        module.fail_json(msg='boto3 required for this module')127    state = module.params.get('state')128    region, ec2_url, aws_connect_kwargs = get_aws_connection_info(module, boto3=True)129    if region:130        connection = boto3_conn(module, conn_type='client', resource='logs', region=region, endpoint=ec2_url, **aws_connect_kwargs)131    else:132        module.fail_json(msg="region must be specified")133    if state == 'present':134        create_metric_filter(connection, module)135    elif state == 'absent':136        delete_metric_filter(connection, module)137if __name__ == '__main__':...

test_disco_log_metrics.py

Source:test_disco_log_metrics.py

1"""Tests of disco_log_metrics"""2from unittest import TestCase3from mock import MagicMock, call, PropertyMock4from disco_aws_automation import DiscoLogMetrics5from tests.helpers.patch_disco_aws import get_mock_config6class DiscoLogMetricsTests(TestCase):7    """Test DiscoLogMetrics"""8    def setUp(self):9        self.log_metrics = DiscoLogMetrics('test-env')10        self.log_metrics.logs = MagicMock()11        config_mock = PropertyMock(return_value=get_mock_config({12            'mhcdummy.metric_name': {13                'log_file': '/error_log',14                'filter_pattern': 'error',15                'metric_value': 116            }17        }))18        type(self.log_metrics).config = config_mock19        # pylint: disable=C010320        def _describe_log_groups(logGroupNamePrefix):21            if logGroupNamePrefix == 'test-env/':  # getting all log metrics in env22                return {'logGroups': [{'logGroupName': 'test-env/mhcdummy/info_log'},23                                      {'logGroupName': 'test-env/mhcbanana/warning_log'}]}24            # getting all log metrics for hostclass25            return {'logGroups': [{'logGroupName': 'test-env/mhcdummy/info_log'}]}26        # pylint: disable=C010327        def _describe_metric_filters(logGroupName):28            if logGroupName == 'test-env/mhcdummy/info_log':29                return {'metricFilters': [{'filterName': 'mhcdummy_metric'}]}30            elif logGroupName == 'test-env/mhcbanana/warning_log':31                return {'metricFilters': [{'filterName': 'mhcbanana_metric'}]}32        self.log_metrics.logs.describe_log_groups.side_effect = _describe_log_groups33        self.log_metrics.logs.describe_metric_filters.side_effect = _describe_metric_filters34    def test_list_log_groups(self):35        """Test listing log groups"""36        log_groups = self.log_metrics.list_log_groups('mhcdummy')37        self.assertEqual(len(log_groups), 1)38        self.assertEqual(log_groups[0]['logGroupName'], 'test-env/mhcdummy/info_log')39    def test_list_metrics(self):40        """Test list log metric filters for hostclass"""41        metrics = self.log_metrics.list_metric_filters('mhcdummy')42        self.assertEqual(len(metrics), 1)43        self.assertEqual(metrics[0]['filterName'], 'mhcdummy_metric')44    def test_delete_metrics(self):45        """Test delete metric filters for hostclass"""46        self.log_metrics.delete_metrics('mhcdummy')47        self.log_metrics.logs.delete_metric_filter.assert_called_once_with(48            logGroupName='test-env/mhcdummy/info_log',49            filterName='mhcdummy_metric')50    def test_delete_log_groups(self):51        """Test delete log groups for hostclass"""52        self.log_metrics.delete_log_groups('mhcdummy')53        self.log_metrics.logs.delete_log_group.assert_called_once_with(54            logGroupName='test-env/mhcdummy/info_log')55    def test_delete_all_metrics(self):56        """Test delete all metric filters in env"""57        self.log_metrics.delete_all_metrics()58        expected = [call(logGroupName='test-env/mhcdummy/info_log',59                         filterName='mhcdummy_metric'),60                    call(logGroupName='test-env/mhcbanana/warning_log',61                         filterName='mhcbanana_metric')]62        self.log_metrics.logs.delete_metric_filter.assert_has_calls(expected, any_order=True)63    def test_delete_all_log_groups(self):64        """Test delete all log groups in environment"""65        self.log_metrics.delete_all_log_groups()66        expected = [call(logGroupName='test-env/mhcdummy/info_log'),67                    call(logGroupName='test-env/mhcbanana/warning_log')]68        self.log_metrics.logs.delete_log_group.assert_has_calls(expected, any_order=True)69    def test_update(self):70        """Test deleting and creating metric filter from config"""71        self.log_metrics.update('mhcdummy')72        self.log_metrics.logs.delete_metric_filter.assert_called_once_with(73            logGroupName='test-env/mhcdummy/info_log',74            filterName='mhcdummy_metric')75        self.log_metrics.logs.put_metric_filter.assert_called_once_with(76            logGroupName='test-env/mhcdummy/error_log',77            filterName='mhcdummy-metric_name',78            filterPattern='error',79            metricTransformations=[{80                'metricName': 'mhcdummy-metric_name',81                'metricNamespace': 'LogMetrics/test-env',82                'metricValue': 1...

metricFilterHandler.py

Source:metricFilterHandler.py

...31    return response32'''33Deletes specified metric filter34'''35def delete_metric_filter(request, logGroupName):36    describeResponse = {}37    deleteResponse = None38    deleteAlarmRequest = request.copy()39    deleteAlarmRequest['logGroupName'] = logGroupName40    try:41        if 'filterName' in request:42            describeResponse = client.describe_metric_filters(43                logGroupName=logGroupName,44                filterNamePrefix=request['filterName']45            )46        else:47            describeResponse = client.describe_metric_filters(48                logGroupName=logGroupName49            )50        if len(describeResponse['metricFilters']) > 0:51            for metricFilter in describeResponse['metricFilters']:52                deleteResponse = client.delete_metric_filter(53                    logGroupName=metricFilter['logGroupName'],54                    filterName=metricFilter['filterName']55                )56                logger.info(deleteResponse)57                if 'deleteMetricAlarms' in request and request['deleteMetricAlarms']:58                    # Start deleting alarms59                    for metricTransformation in metricFilter['metricTransformations']:60                        deleteAlarmRequest['metricName'] = metricTransformation['metricName']61                        deleteAlarmRequest['metricNamespace'] = metricTransformation['metricNamespace']62                        metricAlarmHandler.delete_metric_alarm(deleteAlarmRequest)63        else:64            deleteResponse = 'Metric filter not found: '+request['filterName']65            logger.info(deleteResponse)66    except Exception as e:...

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.