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

How to use ssm_client method in localstack

Best Python code snippet using localstack_python

master.cfg

Source:master.cfg

1# -*- python -*-2# ex: set filetype=python:3import os4import boto35import subprocess6import json7from buildbot.plugins import *8from buildbot.www.hooks.github import GitHubEventHandler9from twisted.python import log10def get_default_environment(ssm_client, variable, default):11    value = os.environ.get(variable, default)12    if value.startswith("ssm://"):13        try:14            response = ssm_client.get_parameter(Name=value.replace("ssm://", ""), WithDecryption=True)15            parameter = response.get('Parameter', {})16            return parameter.get('Value', default)17        except:18            return default19    return value20class StrictHandler(GitHubEventHandler):21    def handle_pull_request(self, payload, event):22        number = payload['number']23        head_user_login = payload['pull_request']['head']['user']['login']24        author_association = payload['pull_request']['author_association']25        if (head_user_login == 'UnsafePointer' and author_association == 'OWNER'):26            log.msg("GitHub PR #{}, Processing: "27                    "head commit correct user login and author association".format(number))28            return super().handle_pull_request(payload, event)29        log.msg("GitHub PR #{}, Ignoring: "30                    "head commit wrong user login and author association".format(number))31        return [], 'git'32instance_identity = subprocess.check_output(['curl', '-s', 'http://169.254.169.254/latest/dynamic/instance-identity/document'])33identity_data = json.loads(instance_identity)34aws_region = identity_data['region']35ssm_client = boto3.client('ssm', region_name=aws_region)36c = BuildmasterConfig = {}37linux_worker_name = get_default_environment(ssm_client, "BUILDBOT_LINUX_WORKER_USERNAME", "example-worker-linux")38windows_worker_name = get_default_environment(ssm_client, "BUILDBOT_WINDOWS_WORKER_USERNAME", "example-worker-windows")39worker_password = get_default_environment(ssm_client, "BUILDBOT_WORKER_PASSWORD", "pass")40c['workers'] = []41c['protocols'] = {'pb': {'port': os.environ.get("BUILDBOT_WORKER_PORT", 9989)}}42c['change_source'] = []43pipelines = [44  {45    'project': 'ruby',46    'platform': 'linux',47    'workername': linux_worker_name,48    'command': './ci/linux.sh',49    'ami': get_default_environment(ssm_client, "BUILDBOT_LINUX_WORKER_AMI_ID", "not-a-sg-id"),50    'sgs': [get_default_environment(ssm_client, "BUILDBOT_ALLOW_SSH_SG", "not-a-sg-id")]51  },52  {53    'project' : 'ruby',54    'platform': 'windows',55    'workername': windows_worker_name,56    'command': 'call ./ci/windows.bat',57    'ami': get_default_environment(ssm_client, "BUILDBOT_WINDOWS_WORKER_AMI_ID", "not-a-sg-id"),58    'sgs': [get_default_environment(ssm_client, "BUILDBOT_ALLOW_SSH_SG", "not-a-sg-id"), get_default_environment(ssm_client, "BUILDBOT_ALLOW_RDP_SG", "not-a-sg-id")]59  },60  {61    'project': 'shinobu',62    'platform': 'linux',63    'workername': linux_worker_name,64    'command': './ci/linux.sh',65    'ami': get_default_environment(ssm_client, "BUILDBOT_LINUX_WORKER_AMI_ID", "not-a-sg-id"),66    'sgs': [get_default_environment(ssm_client, "BUILDBOT_ALLOW_SSH_SG", "not-a-sg-id")]67  },68  {69    'project' : 'shinobu',70    'platform': 'windows',71    'workername': windows_worker_name,72    'command': 'call ./ci/windows.bat',73    'ami': get_default_environment(ssm_client, "BUILDBOT_WINDOWS_WORKER_AMI_ID", "not-a-sg-id"),74    'sgs': [get_default_environment(ssm_client, "BUILDBOT_ALLOW_SSH_SG", "not-a-sg-id"), get_default_environment(ssm_client, "BUILDBOT_ALLOW_RDP_SG", "not-a-sg-id")]75  }76]77c['schedulers'] = []78c['builders'] = []79for pipeline in pipelines:80  workername = pipeline['workername']81  pipeline_name = "{}-{}".format(pipeline['project'], pipeline['platform'])82  c['workers'].append(worker.EC2LatentWorker(workername,83                                             worker_password,84                                             get_default_environment(ssm_client, "BUILDBOT_WORKER_INSTANCE_TYPE", "not-an-instance-type"),85                                             ami=pipeline['ami'],86                                             keypair_name=get_default_environment(ssm_client, "BUILDBOT_KEYPAIR_NAME", "not-a-keypair-name"),87                                             subnet_id=get_default_environment(ssm_client, "BUILDBOT_SUBNET", "not-a-subnet-id"),88                                             security_group_ids=pipeline['sgs'],89                                             build_wait_timeout=0,90                                             instance_profile_name=get_default_environment(ssm_client, "BUILDBOT_WORKERS_INSTANCE_PROFILE", "not-an-instance-profile-name")))91  c['schedulers'].append(schedulers.SingleBranchScheduler(name="webhook-pull-{}".format(pipeline_name),92                                                          change_filter=util.ChangeFilter(category='pull', repository="https://github.com/UnsafePointer/{}".format(pipeline['project'])),93                                                          treeStableTimer=None,94                                                          fileIsImportant=None,95                                                          builderNames=[pipeline_name]))96  c['schedulers'].append(schedulers.ForceScheduler(name="force-{}".format(pipeline_name),97                                                   builderNames=[pipeline_name]))98  factory = util.BuildFactory()99  factory.addStep(steps.GitHub(repourl="http://github.com/UnsafePointer/{}.git".format(pipeline['project']),100                               mode='full',101                               submodules=True))102  factory.addStep(steps.ShellCommand(command=pipeline['command'],103                                     workdir="build"))104  if pipeline['platform'] == 'windows':105      factory.addStep(steps.ShellCommand(command=["call", "self-terminate.bat"],106                                         workdir="C:\\Users\\Administrator",107                                         alwaysRun=True))108  c['builders'].append(util.BuilderConfig(name=pipeline_name,109                                          workernames=[workername],110                                          factory=factory))111context = util.Interpolate("buildbot/%(prop:buildername)s")112gs = reporters.GitHubStatusPush(token=get_default_environment(ssm_client, "BUILDBOT_GITHUB_API_TOKEN", "not-a-real-token"),113                                context=context,114                                startDescription='Build started.',115                                endDescription='Build done.')116c['services'] = []117c['services'].append(gs)118c['title'] = "UnsafePointer's emudev Buildbot"119c['titleURL'] = "https://github.com/UnsafePointer"120c['buildbotURL'] = get_default_environment(ssm_client, "BUILDBOT_WEB_URL", "http://localhost:8010/")121c['www'] = dict(port=int(get_default_environment(ssm_client, "BUILDBOT_WEB_PORT", "8010")),122                plugins=dict(waterfall_view={}, console_view={}),123                change_hook_dialects={'github': {124                    'secret': get_default_environment(ssm_client, "BUILDBOT_GITHUB_HOOK_SECRET", "not-so-secret"),125                    'strict': True,126                    'class': StrictHandler127                }})128admin_username = get_default_environment(ssm_client, "BUILDBOT_ADMIN_USERNAME", "admin")129admin_password = get_default_environment(ssm_client, "BUILDBOT_ADMIN_PASSWORD", "admin")130c['www']['authz'] = util.Authz(131        allowRules = [132            util.AnyEndpointMatcher(role="admins")133        ],134        roleMatchers = [135            util.RolesFromUsername(roles=['admins'], usernames=[admin_username])136        ]137)138c['www']['auth'] = util.UserPasswordAuth([(admin_username, admin_password)])139c['db'] = {140    'db_url' : get_default_environment(ssm_client, "BUILDBOT_DB_URL", "sqlite://").format(**os.environ),...

test_authorizer.py

Source:test_authorizer.py

1import json2import pytest3import boto34import os5from moto import mock_ssm6@pytest.fixture(scope='function')7def aws_credentials():8    # Mocked AWS Credentials for moto, this is required 9    # as per their readme: https://github.com/spulec/moto10    os.environ['AWS_ACCESS_KEY_ID'] = 'testing'11    os.environ['AWS_SECRET_ACCESS_KEY'] = 'testing'12    os.environ['AWS_SECURITY_TOKEN'] = 'testing'13    os.environ['AWS_SESSION_TOKEN'] = 'testing'14    # Disable X-Ray for unit tests15    from aws_xray_sdk import global_sdk_config16    global_sdk_config.set_sdk_enabled(False)17@pytest.fixture(scope='function')18def ssm(aws_credentials):19    mock = mock_ssm()20    mock.start()21    # There is currently a bug on moto, this line is needed as a workaround22    # Ref: https://github.com/spulec/moto/issues/192623    # boto3.setup_default_session()24    ssm_client = boto3.client('ssm', 'us-west-2')25    token_name = "TEST_TOKEN"26    token_value = "TEST_VALUE"27    response = ssm_client.put_parameter(28        Name=token_name, Description="A test parameter", Value=token_value, Type="SecureString"29    )30    yield (ssm_client, token_name, token_value)31    mock.stop()32def test_get_auth_token_success(ssm):33    ssm_client, tname, tvalue = ssm34    # Local imports are also recommended as per moto35    # Ref: https://github.com/spulec/moto#what-about-those-pesky-imports36    from functions import authorizer37    # We have already created a test token, now attempt to retrieve it38    auth_token = authorizer.get_auth_token(ssm_client, tname)39    assert auth_token == tvalue40def test_get_auth_token_failure(ssm):41    ssm_client, tname, tvalue = ssm42    invalid_tvalue = "INVALID"43    # Local imports are also recommended as per moto44    # Ref: https://github.com/spulec/moto#what-about-those-pesky-imports45    from functions import authorizer46    # We have already created a test token, now attempt to retrieve it47    auth_token = authorizer.get_auth_token(ssm_client, invalid_tvalue)48    assert auth_token is False49def test_generate_policy():50    from functions import authorizer51    test_principalId = "test-principal"52    test_effect_allow = "Allow"53    test_effect_deny = "Deny"54    test_resource = "arn:aws:execute-api:test-region:test-account:test-resource"55    allow_response = authorizer.generate_policy(56        test_principalId,57        test_effect_allow,58        test_resource59    )60    assert type(allow_response) is dict61    assert allow_response['principalId'] == test_principalId62    assert type(allow_response['policyDocument']) is dict63    assert allow_response['policyDocument']['Version'] == '2012-10-17'64    assert type(allow_response['policyDocument']['Statement']) is list65    assert allow_response['policyDocument']['Statement'][0]['Action'] == 'execute-api:Invoke'66    assert allow_response['policyDocument']['Statement'][0]['Effect'] == 'Allow'67    assert allow_response['policyDocument']['Statement'][0]['Resource'] == test_resource68    deny_response = authorizer.generate_policy(69        test_principalId,70        test_effect_deny,71        test_resource72    )73    assert type(deny_response) is dict74    assert deny_response['principalId'] == test_principalId75    assert type(deny_response['policyDocument']) is dict76    assert deny_response['policyDocument']['Version'] == '2012-10-17'77    assert type(deny_response['policyDocument']['Statement']) is list78    assert deny_response['policyDocument']['Statement'][0]['Action'] == 'execute-api:Invoke'79    assert deny_response['policyDocument']['Statement'][0]['Effect'] == 'Deny'80    assert deny_response['policyDocument']['Statement'][0]['Resource'] == test_resource81def test_validate_token(ssm):82    ssm_client, tname, tvalue = ssm83    invalid_tvalue = "INVALID"84    85    valid_test_event = {86        'type': "TOKEN",87        'authorizationToken': tvalue,88        'methodArn': "arn:aws:execute-api:test-region:test-account:test-resource"89    }90    from functions import authorizer91    allow_response = authorizer.validate_token(valid_test_event, "", ssm_client, tname)92    93    assert type(allow_response) is dict94    assert allow_response['principalId'] == 'webhook_service'95    assert type(allow_response['policyDocument']) is dict96    assert allow_response['policyDocument']['Version'] == '2012-10-17'97    assert type(allow_response['policyDocument']['Statement']) is list98    assert allow_response['policyDocument']['Statement'][0]['Action'] == 'execute-api:Invoke'99    assert allow_response['policyDocument']['Statement'][0]['Effect'] == 'Allow'100    assert allow_response['policyDocument']['Statement'][0]['Resource'] == valid_test_event['methodArn']101    invalid_test_event = {102        'type': "TOKEN",103        'authorizationToken': invalid_tvalue,104        'methodArn': "arn:aws:execute-api:test-region:test-account:test-resource"105    }106    deny_response = authorizer.validate_token(invalid_test_event, "", ssm_client, tname)107    assert type(deny_response) is dict108    assert deny_response['principalId'] == 'webhook_service'109    assert type(deny_response['policyDocument']) is dict110    assert deny_response['policyDocument']['Version'] == '2012-10-17'111    assert type(deny_response['policyDocument']['Statement']) is list112    assert deny_response['policyDocument']['Statement'][0]['Action'] == 'execute-api:Invoke'113    assert deny_response['policyDocument']['Statement'][0]['Effect'] == 'Deny'114    assert deny_response['policyDocument']['Statement'][0]['Resource'] == invalid_test_event['methodArn']115    incomplete_test_event = {116        "type":"TOKEN",117        "methodArn":"arn:aws:execute-api:test-region:test-account:test-resource"118    }119    response_403 = authorizer.validate_token(incomplete_test_event, "", ssm_client, tname)120    assert type(response_403) is dict...

test_ssm.py

Source:test_ssm.py

1import unittest2from localstack.utils.aws import aws_stack3class SSMTest(unittest.TestCase):4    def test_describe_parameters(self):5        ssm_client = aws_stack.connect_to_service('ssm')6        response = ssm_client.describe_parameters()7        assert 'Parameters' in response8        assert isinstance(response['Parameters'], list)9    def test_put_parameters(self):10        ssm_client = aws_stack.connect_to_service('ssm')11        ssm_client.put_parameter(12            Name='test_put',13            Description='test',14            Value='1',15            Type='String',16        )17        response = ssm_client.get_parameters(18            Names=['test_put'],19        )20        assert 'Parameters' in response21        assert len(response['Parameters']) > 022        assert response['Parameters'][0]['Name'] == 'test_put'...

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.