KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use sns_subscription_arn method in localstack

Best Python code snippet using localstack_python

PagerDuty-AWS-CloudWatch-SNS-Integration.py

Source:PagerDuty-AWS-CloudWatch-SNS-Integration.py

1import boto32import requests3import json4# Update to match a locally stored AWS profile5session = boto3.session.Session(profile_name='augmenthor')6PDAccountAccessKey ='SwEYWx-i3x24JMZhThxt'7# Update to match your PagerDuty email address8PD_ACCT_OWNER_EMAIL = 'cristian.i.balan@icloud.com'9sns_clnt = session.client('sns')10sns_topic_arn =None11sns_subscription_pd_protocol = 'HTTPS'12sns_subscription_arn =None13def create_escalation_policy(team_id):14    url = 'https://api.pagerduty.com/escalation_policies'15    headers = {16        'Accept': 'application/vnd.pagerduty+json;version=2',17        'Authorization': 'Token token={token}'.format(token=PDAccountAccessKey),18        'Content-type': 'application/json',19        'From': PD_ACCT_OWNER_EMAIL20    }21    payload = {22    "escalation_policy": {23        "type": "escalation_policy",24        "name": "DevOpsAPI",25        "escalation_rules": [26            {27                "escalation_delay_in_minutes": 30,28                "targets": [29                    {30                        "id": "PCHN6LF",31                            "type": "user_reference",32                            "summary": "Ionel Balan",33                            "self": "https://api.pagerduty.com/users/PCHN6LF",34                            "html_url": "https://cbalan.pagerduty.com/users/PCHN6LF"35                    }36                ]37            }38        ],39        "num_loops": 0,40        "on_call_handoff_notifications": "if_has_services",41        "teams": [42            {43                "id": "{teamid}".format(teamid=team_id),44                "type": "team_reference",45                "summary": "ACME Hotel Back Office - DevOps",46                "self": "https://api.pagerduty.com/teams/PJKUNJK",47                "html_url": "https://cbalan.pagerduty.com/teams/PJKUNJK"48            }49        ],50        "description": "Here is the ep for the devops team."51    }52}53    try:54        r = requests.post(url, headers=headers, data=json.dumps(payload))55        print('Status Code: {code}'.format(code=r.status_code))56        print('escalation ploicy id: '+ r.json()['escalation_policy']['id'])57        return r.json()['escalation_policy']['id']58    except:59        print ("Error processing the create_escalation_policy's POST request")60def create_service(escalation_policy_id):61    url = 'https://api.pagerduty.com/services'62    headers = {63        'Accept': 'application/vnd.pagerduty+json;version=2',64        'Authorization': 'Token token={token}'.format(token=PDAccountAccessKey),65        'Content-type': 'application/json',66        'From': PD_ACCT_OWNER_EMAIL67    }68    payload = {69    "service": {70        "type": "service",71        "name": "AWS CloudWatch API",72        "description": "AWS CloudWatch service (created via API)",73        "status": "active",74        "escalation_policy": {75            "id": escalation_policy_id,76            "type": "escalation_policy_reference"77        },78        "incident_urgency_rule": {79            "type": "constant",80            "urgency": "high"81        },82        "alert_creation": "create_alerts_and_incidents"83    }84}85    try:86        r = requests.post(url, headers=headers, data=json.dumps(payload))87        print('Status Code: {code}'.format(code=r.status_code))88        print('service id: '+r.json()['service']['id'])89        return r.json()['service']['id']90    except:91        print ("Error processing the create_service's POST request")92def create_integration(service_id, subdomain):93    url = 'https://api.pagerduty.com/services/{id}/integrations'.format(id=service_id)94    headers = {95        'Accept': 'application/vnd.pagerduty+json;version=2',96        'Authorization': 'Token token={token}'.format(token=PDAccountAccessKey),97        'Content-type': 'application/json',98        'From': PD_ACCT_OWNER_EMAIL99    }100    payload = {101        "integration": {102            "type": "event_transformer_api_inbound_integration",103            "name": "Amazon CloudWatch API",104            "service": {105                "id": service_id,106                "type": "service_reference",107                "summary": "Amazon CloudWatch API",108                "self": "https://api.pagerduty.com/services/{id}".format(id=service_id),109                "html_url": "https://{sd}.pagerduty.com/services/{id}".format(sd=subdomain,id=service_id)110            },111            "vendor": {112              "id": "PZQ6AUS",113              "type": "vendor_reference"114            }115        }116    }117    try:118        r = requests.post(url, headers=headers, data=json.dumps(payload))119        print('Status Code: {code}'.format(code=r.status_code))120        print('integration key: '+r.json()['integration']['integration_key'])121        return r.json()['integration']['integration_key']122    except Exception as error:123        print(error)124def create_sns_topic(topicname):125    # create an SNS topic126    try:127        response = sns_clnt.create_topic(Name=topicname)128        return response['TopicArn']129    except Exception as error:130        print(error)131def create_sns_subcription_endpoint(snstopicarn):132    # Subcribe to a PD endpoint (previously created in PD's admin console)133    try:134        response = sns_clnt.subscribe(135            TopicArn=snstopicarn,136            Protocol=sns_subscription_pd_protocol,137            Endpoint=sns_subscription_pd_endpoint,138            ReturnSubscriptionArn=True139        )140        return response['SubscriptionArn']141    except Exception as error:142        print(error)143if __name__ == '__main__':144    #create in PagerDuty sbdomain:145    # - an escalation policy for a single team, a new service146    # - a new service which will create alerts and incidents147    # - a new integration with AWS CloudWatch for the service created148    ep_team_id="PX7BN8L"149    service_subdomain="cbalan"150    ik=create_integration(create_service(create_escalation_policy(ep_team_id)), service_subdomain)151    #use the integration key to create an AWS SNS topic and subscription152    sns_subscription_pd_endpoint = "https://events.pagerduty.com/integration/{integration_key}/enqueue".format(153        integration_key=ik)154    #CloudWatch to pagerDuty Topic and Subscription creation155    sns_topic_arn=create_sns_topic("CloudWatch-to-PagerDuty")156    print("sns_topic_arn: {}".format(sns_topic_arn))157    sns_subscription_arn= create_sns_subcription_endpoint(sns_topic_arn)...

sns.py

Source:sns.py

...73        # TODO: use get_subscription_attributes to compare FilterPolicy74        return result[0] if result else None75    @staticmethod76    def get_deploy_templates():77        def sns_subscription_arn(params, resources, resource_id, **kwargs):78            resource = resources[resource_id]79            return resource["PhysicalResourceId"]80        def sns_subscription_params(params, **kwargs):81            def attr_val(val):82                return json.dumps(val) if isinstance(val, (dict, list)) else str(val)83            attrs = [84                "DeliveryPolicy",85                "FilterPolicy",86                "RawMessageDelivery",87                "RedrivePolicy",88            ]89            result = dict([(a, attr_val(params[a])) for a in attrs if a in params])90            return result91        return {...

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.