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

Copied to Clipboard!

How to use arn method in localstack

Best Python code snippet using localstack_python

test_sns.py

Source:test_sns.py

1# Copyright 2017 Capital One Services, LLC2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7# http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14from __future__ import absolute_import, division, print_function, unicode_literals15import json, time16from .common import BaseTest, functional17class TestSNS(BaseTest):18    @functional19    def test_sns_remove_matched(self):20        session_factory = self.replay_flight_data('test_sns_remove_matched')21        client = session_factory().client('sns')22        name = 'test-sns-remove-matched'23        topic_arn = client.create_topic(Name=name)['TopicArn']24        self.addCleanup(client.delete_topic, TopicArn=topic_arn)25        client.set_topic_attributes(26            TopicArn=topic_arn,27            AttributeName="Policy",28            AttributeValue=json.dumps({29                "Version": "2012-10-17",30                "Statement": [31                    {32                        "Sid": "SpecificAllow",33                        "Effect": "Allow",34                        "Principal": {35                            "AWS": "arn:aws:iam::644160558196:root"36                        },37                        "Action": [38                            "SNS:Subscribe"39                        ],40                        "Resource": topic_arn41                    },42                    {43                        "Sid": "Public",44                        "Effect": "Allow",45                        "Principal": "*",46                        "Action": [47                            "SNS:GetTopicAttributes"48                        ],49                        "Resource": topic_arn50                    }51                ]52            })53        )54        p = self.load_policy({55            'name': 'sns-rm-matched',56            'resource': 'sns',57            'filters': [58                {'TopicArn': topic_arn},59                {'type': 'cross-account',60                 'whitelist': ["123456789012"]}61            ],62            'actions': [63                {'type': 'remove-statements',64                 'statement_ids': 'matched'}]65            },66            session_factory=session_factory)67        resources = p.run()68        self.assertEqual([r['TopicArn'] for r in resources], [topic_arn])69        data = json.loads(client.get_topic_attributes(70            TopicArn=resources[0]['TopicArn'])['Attributes']['Policy'])71        self.assertEqual(72            [s['Sid'] for s in data.get('Statement', ())],73            ['SpecificAllow'])74    @functional75    def test_sns_remove_named(self):76        session_factory = self.replay_flight_data('test_sns_remove_named')77        client = session_factory().client('sns')78        name = 'test-sns-remove-named'79        topic_arn = client.create_topic(Name=name)['TopicArn']80        self.addCleanup(client.delete_topic, TopicArn=topic_arn)81        client.set_topic_attributes(82            TopicArn=topic_arn,83            AttributeName="Policy",84            AttributeValue=json.dumps({85                "Version": "2012-10-17",86                "Statement": [87                    {88                        "Sid": "SpecificAllow",89                        "Effect": "Allow",90                        "Principal": "*",91                        "Action": ["SNS:Subscribe"],92                        "Resource": topic_arn93                    },94                    {95                        "Sid": "RemoveMe",96                        "Effect": "Allow",97                        "Principal": "*",98                        "Action": ["SNS:GetTopicAttributes"],99                        "Resource": topic_arn100                    }101                ]102            })103        )104        p = self.load_policy({105            'name': 'sns-rm-named',106            'resource': 'sns',107            'filters': [{'TopicArn': topic_arn}],108            'actions': [109                {'type': 'remove-statements',110                 'statement_ids': ['RemoveMe']}]111            },112            session_factory=session_factory)113        resources = p.run()114        self.assertEqual(len(resources), 1)115        data = json.loads(client.get_topic_attributes(116            TopicArn=resources[0]['TopicArn'])['Attributes']['Policy'])117        self.assertTrue('RemoveMe' not in118            [s['Sid'] for s in data.get('Statement', ())])119    @functional120    def test_sns_modify_replace_policy(self):121        session_factory = self.replay_flight_data('test_sns_modify_replace_policy')122        client = session_factory().client('sns')123        name = 'test_sns_modify_replace_policy'124        topic_arn = client.create_topic(Name=name)['TopicArn']125        self.addCleanup(client.delete_topic, TopicArn=topic_arn)126        client.set_topic_attributes(127            TopicArn=topic_arn,128            AttributeName="Policy",129            AttributeValue=json.dumps({130                "Version": "2012-10-17",131                "Statement": [132                    {133                        "Sid": "SpecificAllow",134                        "Effect": "Allow",135                        "Principal": "*",136                        "Action": ["SNS:Subscribe"],137                        "Resource": topic_arn138                    }139                ]140            })141        )142        p = self.load_policy({143            'name': 'sns-modify-replace-policy',144            'resource': 'sns',145            'filters': [{'TopicArn': topic_arn}],146            'actions': [147                {'type': 'modify-policy',148                    'add-statements': [{149                        "Sid": "ReplaceWithMe",150                        "Effect": "Allow",151                        "Principal": "*",152                        "Action": ["SNS:GetTopicAttributes"],153                        "Resource": topic_arn154                    }],155                    'remove-statements': "*"156                }]157            },158            session_factory=session_factory)159        resources = p.run()160        self.assertEqual(len(resources), 1)161        data = json.loads(client.get_topic_attributes(162            TopicArn=resources[0]['TopicArn'])['Attributes']['Policy'])163        self.assertTrue('ReplaceWithMe' in164            [s['Sid'] for s in data.get('Statement', ())])165    @functional166    def test_sns_account_id_template(self):167        session_factory = self.replay_flight_data('test_sns_account_id_template')168        client = session_factory().client('sns')169        name = 'test_sns_account_id_template'170        topic_arn = client.create_topic(Name=name)['TopicArn']171        self.addCleanup(client.delete_topic, TopicArn=topic_arn)172        client.set_topic_attributes(173            TopicArn=topic_arn,174            AttributeName="Policy",175            AttributeValue=json.dumps({176                "Version": "2012-10-17",177                "Statement": [178                    {179                        "Sid": "SpecificAllow",180                        "Effect": "Allow",181                        "Principal": "*",182                        "Action": ["SNS:Subscribe"],183                        "Resource": topic_arn184                    }185                ]186            })187        )188        p = self.load_policy({189            'name': 'sns-modify-replace-policy',190            'resource': 'sns',191            'filters': [{'TopicArn': topic_arn}],192            'actions': [193                {'type': 'modify-policy',194                    'add-statements': [{195                        "Sid": "__default_statement_ID_{account_id}",196                        "Effect": "Allow",197                        "Principal": {198                            "Service": "s3.amazonaws.com"199                        },200                        "Action": "SNS:Publish",201                        "Resource": topic_arn,202                        "Condition": {203                            "StringEquals": {204                                "AWS:SourceAccount": "{account_id}"205                            },206                            "ArnLike": {207                                "aws:SourceArn": "arn:aws:s3:*:*:*"208                            }209                        }210                    }],211                    'remove-statements': "*"212                }]213            },214            session_factory=session_factory)215        resources = p.run()216        self.assertEqual(len(resources), 1)217        data = json.loads(client.get_topic_attributes(218            TopicArn=resources[0]['TopicArn'])['Attributes']['Policy'])219        self.assertTrue('__default_statement_ID_' + self.account_id in220            [s['Sid'] for s in data.get('Statement', ())])221    @functional222    def test_sns_modify_remove_policy(self):223        session_factory = self.replay_flight_data('test_sns_modify_remove_policy')224        client = session_factory().client('sns')225        name = 'test_sns_modify_remove_policy'226        topic_arn = client.create_topic(Name=name)['TopicArn']227        self.addCleanup(client.delete_topic, TopicArn=topic_arn)228        client.set_topic_attributes(229            TopicArn=topic_arn,230            AttributeName="Policy",231            AttributeValue=json.dumps({232                "Version": "2012-10-17",233                "Statement": [234                    {235                        "Sid": "SpecificAllow",236                        "Effect": "Allow",237                        "Principal": "*",238                        "Action": ["SNS:Subscribe"],239                        "Resource": topic_arn240                    },241                    {242                        "Sid": "RemoveMe",243                        "Effect": "Allow",244                        "Principal": "*",245                        "Action": ["SNS:GetTopicAttributes"],246                        "Resource": topic_arn247                    }248                ]249            })250        )251        p = self.load_policy({252            'name': 'sns-modify-remove-policy',253            'resource': 'sns',254            'filters': [{'TopicArn': topic_arn}],255            'actions': [256                {'type': 'modify-policy',257                    'add-statements': [],258                    'remove-statements': ['RemoveMe']259                }]260            },261            session_factory=session_factory)262        resources = p.run()263        self.assertEqual(len(resources), 1)264        data = json.loads(client.get_topic_attributes(265            TopicArn=resources[0]['TopicArn'])['Attributes']['Policy'])266        self.assertTrue('RemoveMe' not in267            [s['Sid'] for s in data.get('Statement', ())])268    @functional269    def test_sns_modify_add_policy(self):270        session_factory = self.replay_flight_data('test_sns_modify_add_policy')271        client = session_factory().client('sns')272        name = 'test_sns_modify_add_policy'273        topic_arn = client.create_topic(Name=name)['TopicArn']274        self.addCleanup(client.delete_topic, TopicArn=topic_arn)275        client.set_topic_attributes(276            TopicArn=topic_arn,277            AttributeName="Policy",278            AttributeValue=json.dumps({279                "Version": "2012-10-17",280                "Statement": [281                    {282                        "Sid": "SpecificAllow",283                        "Effect": "Allow",284                        "Principal": "*",285                        "Action": ["SNS:Subscribe"],286                        "Resource": topic_arn287                    }288                ]289            })290        )291        p = self.load_policy({292            'name': 'sns-modify-add-policy',293            'resource': 'sns',294            'filters': [{'TopicArn': topic_arn}],295            'actions': [296                {'type': 'modify-policy',297                    'add-statements': [{298                        "Sid": "AddMe",299                        "Effect": "Allow",300                        "Principal": "*",301                        "Action": ["SNS:GetTopicAttributes"],302                        "Resource": topic_arn303                    }],304                    'remove-statements': []305                }]306            },307            session_factory=session_factory)308        resources = p.run()309        self.assertEqual(len(resources), 1)310        data = json.loads(client.get_topic_attributes(311            TopicArn=resources[0]['TopicArn'])['Attributes']['Policy'])312        self.assertTrue('AddMe' in313            [s['Sid'] for s in data.get('Statement', ())])314    @functional315    def test_sns_modify_add_and_remove_policy(self):316        session_factory = self.replay_flight_data('test_sns_modify_add_and_remove_policy')317        client = session_factory().client('sns')318        name = 'test_sns_modify_add_and_remove_policy'319        topic_arn = client.create_topic(Name=name)['TopicArn']320        self.addCleanup(client.delete_topic, TopicArn=topic_arn)321        client.set_topic_attributes(322            TopicArn=topic_arn,323            AttributeName="Policy",324            AttributeValue=json.dumps({325                "Version": "2012-10-17",326                "Statement": [327                    {328                        "Sid": "SpecificAllow",329                        "Effect": "Allow",330                        "Principal": "*",331                        "Action": ["SNS:Subscribe"],332                        "Resource": topic_arn333                    },334                    {335                        "Sid": "RemoveMe",336                        "Effect": "Allow",337                        "Principal": "*",338                        "Action": ["SNS:GetTopicAttributes"],339                        "Resource": topic_arn340                    }341                ]342            })343        )344        p = self.load_policy({345            'name': 'sns-modify-add-and-remove-policy',346            'resource': 'sns',347            'filters': [{'TopicArn': topic_arn}],348            'actions': [349                {'type': 'modify-policy',350                    'add-statements': [{351                        "Sid": "AddMe",352                        "Effect": "Allow",353                        "Principal": "*",354                        "Action": ["SNS:GetTopicAttributes"],355                        "Resource": topic_arn356                    }],357                    'remove-statements': ['RemoveMe']358                }]359            },360            session_factory=session_factory)361        resources = p.run()362        self.assertEqual(len(resources), 1)363        data = json.loads(client.get_topic_attributes(364            TopicArn=resources[0]['TopicArn'])['Attributes']['Policy'])365        statement_ids = {s['Sid'] for s in data.get('Statement', ())}366        self.assertTrue('AddMe' in statement_ids)367        self.assertTrue('RemoveMe' not in statement_ids)...

DeviceFarm.py

Source:DeviceFarm.py

...11WEB_URL_TEMPLATE = 'https://us-west-2.console.aws.amazon.com/devicefarm/home#/projects/%s/runs/%s'12device_farm = boto3.client('devicefarm', region_name=REGION)13s3 = boto3.client('s3', region_name=REGION)14logger = logging.getLogger(__name__)15def get_project_arn(name):16    for project in device_farm.list_projects()['projects']:17        if project['name'] == name:18            return project['arn']19    raise KeyError('Could not find project %r' % name)20def get_device_pool(project_arn, name):21    for device_pool in device_farm.list_device_pools(arn=project_arn)['devicePools']:22        if device_pool['name'] == name:23            return device_pool['arn']24    raise KeyError('Could not find device pool %r' % name)25def _upload_presigned_url(url, file_path):26    with open(file_path, encoding='latin1') as fp:27        data = fp.read()28        result = requests.put(url, data=data, headers={'content-type': 'application/octet-stream'})29        assert result.status_code == 20030def create_upload(project_arn, upload_type, name, file_path):31    # name needs to be a file name like app-releaseProduction.apk, not "Android App"32    logger.info('Uploading %s %r' % (upload_type, file_path))33    result = device_farm.create_upload(34        projectArn=project_arn,35        name=name,36        type=upload_type,37        contentType='application/octet-stream',38    )39    upload = result['upload']40    _upload_presigned_url(upload['url'], file_path)41    return upload['arn']42def schedule_run(project_arn, name, device_pool_arn, app_arn, test_package_arn):43    logger.info('Scheduling test run %r' % name)44    result = device_farm.schedule_run(45        projectArn=project_arn,46        appArn=app_arn,47        devicePoolArn=device_pool_arn,48        name=name,49        test={50            'type': 'INSTRUMENTATION',51            'testPackageArn': test_package_arn,52            'parameters':{53            'video_recording': 'false', 'app_performance_monitoring': 'false'54            }55        }56    )57    run = result['run']58    return run['arn']59def _poll_until(method, arn, get_status_callable, success_statuses, timeout_seconds=10):60    check_every_seconds = 10 if timeout_seconds == RUN_TIMEOUT_SECONDS else 161    start = time.time()62    while True:63        result = method(arn=arn)64        current_status = get_status_callable(result)65        if current_status in success_statuses:66            return result67        logger.info('Waiting for %r status %r to be in %r' % (arn, current_status, success_statuses))68        now = time.time()69        if now - start > timeout_seconds:70            raise StopIteration('Time out waiting for %r to be done' % arn)71        time.sleep(check_every_seconds)72def wait_for_upload(arn):73    return _poll_until(74        device_farm.get_upload,75        arn,76        get_status_callable=lambda x: x['upload']['status'],77        success_statuses=('SUCCEEDED', ),78    )79def wait_for_run(test_package_arn):80    result = _poll_until(81        device_farm.get_run,82        test_package_arn,83        get_status_callable=lambda x: x['run']['status'],84        success_statuses=('COMPLETED', ),85        timeout_seconds=RUN_TIMEOUT_SECONDS,86    )87    final_run = result['run']88    logger.info('Final run counts: %(counters)s' % final_run)89    return final_run['result'] == 'PASSED'90def get_run_web_url(project_arn, test_run_arn):91    # project_arn = arn:aws:devicefarm:us-west-2:foo:project:NEW-ARN-HERE92    # test_run_arn = arn:aws:devicefarm:us-west-2:foo:run:project-arn/NEW-ARN-HERE93    project_arn_id = project_arn.split(':')[6]94    test_run_arid = test_run_arn.split('/')[1]95    return WEB_URL_TEMPLATE % (96        project_arn_id,97        test_run_arid,98    )99if __name__ == '__main__':100    logging.basicConfig(format='%(message)s')101    logger.setLevel(logging.INFO)102    project_arn = get_project_arn(PROJECT_NAME)103    logger.info('Project: %r' % project_arn)104    device_pool_arn = get_device_pool(project_arn, DEVICE_POOL_NAME)105    logger.info('Device pool: %r' % device_pool_arn)106    app_arn = create_upload(107        project_arn,108        'ANDROID_APP',109        'app-debug.apk',110        'app/build/outputs/apk/app-debug.apk',111    )112    wait_for_upload(app_arn)113    logger.info('App: %s' % app_arn)114    test_package_arn = create_upload(115        project_arn,116        'INSTRUMENTATION_TEST_PACKAGE',...

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.