-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
9d35cd8
commit ead5ccf
Showing
6 changed files
with
123 additions
and
35 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,42 @@ | ||
import boto3 | ||
|
||
|
||
def get(*paths, decryption=False): | ||
ssm = boto3.client('ssm') | ||
path = _join_slashes(paths) | ||
response = ssm.get_parameters_by_path(Path=path, Recursive=True, WithDecryption=decryption) | ||
params = map(lambda p: _remove_prefix(p, path), response['Parameters']) | ||
return _convert_to_dict(params) | ||
|
||
|
||
def _join_slashes(paths): | ||
path = '/'.join(filter(lambda e: len(e), map(lambda e: _remove_slashes_on_edge(e), paths))) | ||
if not path: | ||
return '/' | ||
else: | ||
return '/' + path + '/' | ||
|
||
|
||
def _remove_slashes_on_edge(string): | ||
if _is_led_by_slash(string): | ||
string = string[1:] | ||
if _is_followed_by_slash(string): | ||
string = string[:-1] | ||
return string | ||
|
||
|
||
def _is_led_by_slash(string): | ||
return string[:1] == '/' | ||
|
||
|
||
def _is_followed_by_slash(string): | ||
return string[-1:] == '/' | ||
|
||
|
||
def _remove_prefix(param, prefix): | ||
param['Name'] = param['Name'][len(prefix):] | ||
return param | ||
|
||
|
||
def _convert_to_dict(params): | ||
return {p['Name']: p['Value'] for p in params} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Empty file.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,54 @@ | ||
import unittest | ||
from unittest.mock import patch | ||
|
||
import aws_paramstore_py as paramstore | ||
|
||
|
||
@patch('aws_paramstore_py.main.boto3') | ||
class TestMain(unittest.TestCase): | ||
def test_get(self, mock): | ||
method = mock.client('ssm').get_parameters_by_path | ||
method.return_value = {'Parameters': [ | ||
{'Name': '/path/to/params/key1', 'Value': "value1"}, | ||
{'Name': '/path/to/params/key2', 'Value': "value2"} | ||
]} | ||
|
||
params = paramstore.get('path', 'to', 'params') | ||
|
||
method.assert_called_with(Path='/path/to/params/', Recursive=True, WithDecryption=False) | ||
self.assertDictEqual({"key1": "value1", "key2": "value2"}, params) | ||
|
||
def test_get_root(self, mock): | ||
method = mock.client('ssm').get_parameters_by_path | ||
|
||
paramstore.get() | ||
|
||
method.assert_called_with(Path='/', Recursive=True, WithDecryption=False) | ||
|
||
def test_get_slash(self, mock): | ||
method = mock.client('ssm').get_parameters_by_path | ||
|
||
paramstore.get('/') | ||
|
||
method.assert_called_with(Path='/', Recursive=True, WithDecryption=False) | ||
|
||
def test_get_leading_slash(self, mock): | ||
method = mock.client('ssm').get_parameters_by_path | ||
|
||
paramstore.get('/path/to') | ||
|
||
method.assert_called_with(Path='/path/to/', Recursive=True, WithDecryption=False) | ||
|
||
def test_get_following_slash(self, mock): | ||
method = mock.client('ssm').get_parameters_by_path | ||
|
||
paramstore.get('path/to/') | ||
|
||
method.assert_called_with(Path='/path/to/', Recursive=True, WithDecryption=False) | ||
|
||
def test_get_with_decryption(self, mock): | ||
method = mock.client('ssm').get_parameters_by_path | ||
|
||
paramstore.get('path/to/params', decryption=True) | ||
|
||
method.assert_called_with(Path='/path/to/params/', Recursive=True, WithDecryption=True) |