Skip to content
Prev Previous commit
Next Next commit
update unit test resources and algorithm
  • Loading branch information
xiazhvera committed Dec 30, 2024
commit 70b6585392681744b01a1e8dc639680bb003135c
8 changes: 3 additions & 5 deletions test/test_mqtt.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
CUSTOM_AUTHORIZER_SIGNATURE_UNENCODED = os.environ.get("CUSTOM_AUTHORIZER_SIGNATURE_UNENCODED")
CUSTOM_AUTHORIZER_TOKEN_KEY_NAME = os.environ.get("CUSTOM_AUTHORIZER_TOKEN_KEY_NAME")
CUSTOM_AUTHORIZER_TOKEN_VALUE = os.environ.get("CUSTOM_AUTHORIZER_TOKEN_VALUE")

AWS_DEFAULT_REGION = os.environ.get("AWS_DEFAULT_REGION")

def has_custom_auth_environment():
return (CUSTOM_AUTHORIZER_ENDPOINT is not None) and (CUSTOM_AUTHORIZER_NAME_SIGNED is not None) and \
Expand Down Expand Up @@ -51,7 +51,7 @@ def get():
warnings.simplefilter('ignore', ResourceWarning)

try:
secrets = boto3.client('secretsmanager')
secrets = boto3.client('secretsmanager', region_name=AWS_DEFAULT_REGION)
response = secrets.get_secret_value(SecretId='unit-test/endpoint')
endpoint = response['SecretString']
response = secrets.get_secret_value(SecretId='ci/mqtt5/us/mqtt5_thing/cert')
Expand All @@ -60,11 +60,9 @@ def get():
key = response['SecretString'].encode('utf8')
region = secrets.meta.region_name
response = secrets.get_secret_value(SecretId='ci/Cognito/identity_id')
cognito_pool = response['SecretString']
cognito_id = response['SecretString']

cognito = boto3.client('cognito-identity')
response = cognito.get_id(IdentityPoolId=cognito_pool)
cognito_id = response['IdentityId']
response = cognito.get_credentials_for_identity(IdentityId=cognito_id)
cognito_creds = response['Credentials']

Expand Down
14 changes: 6 additions & 8 deletions test/test_mqtt5.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
CUSTOM_AUTHORIZER_SIGNATURE_UNENCODED = os.environ.get("CUSTOM_AUTHORIZER_SIGNATURE_UNENCODED")
CUSTOM_AUTHORIZER_TOKEN_KEY_NAME = os.environ.get("CUSTOM_AUTHORIZER_TOKEN_KEY_NAME")
CUSTOM_AUTHORIZER_TOKEN_VALUE = os.environ.get("CUSTOM_AUTHORIZER_TOKEN_VALUE")

AWS_DEFAULT_REGION = os.environ.get("AWS_DEFAULT_REGION")

def has_custom_auth_environment():
return (CUSTOM_AUTHORIZER_ENDPOINT is not None) and (CUSTOM_AUTHORIZER_NAME_SIGNED is not None) and \
Expand Down Expand Up @@ -53,20 +53,18 @@ def get():
warnings.simplefilter('ignore', ResourceWarning)

try:
secrets = boto3.client('secretsmanager')
secrets = boto3.client('secretsmanager', region_name=AWS_DEFAULT_REGION)
response = secrets.get_secret_value(SecretId='unit-test/endpoint')
endpoint = response['SecretString']
response = secrets.get_secret_value(SecretId='unit-test/certificate')
response = secrets.get_secret_value(SecretId='ci/mqtt5/us/mqtt5_thing/cert')
cert = response['SecretString'].encode('utf8')
response = secrets.get_secret_value(SecretId='unit-test/privatekey')
response = secrets.get_secret_value(SecretId='ci/mqtt5/us/mqtt5_thing/key')
key = response['SecretString'].encode('utf8')
region = secrets.meta.region_name
response = secrets.get_secret_value(SecretId='unit-test/cognitopool')
cognito_pool = response['SecretString']
response = secrets.get_secret_value(SecretId='ci/Cognito/identity_id')
cognito_id = response['SecretString']

cognito = boto3.client('cognito-identity')
response = cognito.get_id(IdentityPoolId=cognito_pool)
cognito_id = response['IdentityId']
response = cognito.get_credentials_for_identity(IdentityId=cognito_id)
cognito_creds = response['Credentials']

Expand Down
Loading