Skip to content

Commit

Permalink
Fix Skipped Tests (#604)
Browse files Browse the repository at this point in the history
* testing unit test ci

* update unit test resources and algorithm

* print errors

* try add cognito unit test

* setup client id starting with test-

* setup auth

* print id result

* fix cog test

* update gg test args

* test websocket default

* test for mqtt5 default auth

* add log

* add test log

* clean up
  • Loading branch information
xiazhvera authored Jan 6, 2025
1 parent 22ea68f commit 119ad10
Show file tree
Hide file tree
Showing 3 changed files with 36 additions and 25 deletions.
17 changes: 8 additions & 9 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,32 +51,31 @@ 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']

Config.cache = Config(endpoint, cert, key, region, cognito_creds)
except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as ex:
print(ex)
raise unittest.SkipTest("No credentials")

return Config.cache


def create_client_id():
return 'aws-iot-device-sdk-python-v2-unit-test-{0}'.format(uuid.uuid4())
return 'test-aws-iot-device-sdk-python-v2-unit-test-{0}'.format(uuid.uuid4())


class MqttBuilderTest(unittest.TestCase):
Expand Down
17 changes: 8 additions & 9 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,32 +53,31 @@ 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']

Config.cache = Config(endpoint, cert, key, region, cognito_creds)
except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as ex:
print(ex)
raise unittest.SkipTest("No credentials")

return Config.cache


def create_client_id():
return 'aws-iot-device-sdk-python-v2-unit-test-{0}'.format(uuid.uuid4())
return 'test-aws-iot-device-sdk-python-v2-unit-test-{0}'.format(uuid.uuid4())

class Mqtt5TestCallbacks():
def __init__(self):
Expand Down
27 changes: 20 additions & 7 deletions test/test_samples.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
class Config:
cache = None

def __init__(self, endpoint, cert, key, region):
def __init__(self, endpoint, cert, key, gg_cert, gg_key, region):
self.endpoint = endpoint
self.region = region
self.cert_bytes = cert
Expand All @@ -31,6 +31,14 @@ def __init__(self, endpoint, cert, key, region):
with open(self.key_filepath, 'wb') as key_file:
key_file.write(key)

self.gg_cert_filepath = os.path.join(self._tmp_dirpath, 'gg_certificate.pem')
with open(self.gg_cert_filepath, 'wb') as cert_file:
cert_file.write(gg_cert)

self.gg_key_filepath = os.path.join(self._tmp_dirpath, 'gg_privatekey.pem')
with open(self.gg_key_filepath, 'wb') as key_file:
key_file.write(gg_key)

def __del__(self):
shutil.rmtree(self._tmp_dirpath)

Expand All @@ -45,23 +53,28 @@ def get():
warnings.simplefilter('ignore', ResourceWarning)

try:
secrets = boto3.client('secretsmanager')
secrets = boto3.client('secretsmanager',region_name="us-east-1")
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')
response = secrets.get_secret_value(SecretId='ci/GreengrassDiscovery/cert')
gg_cert = response['SecretString'].encode('utf8')
response = secrets.get_secret_value(SecretId='ci/GreengrassDiscovery/key')
gg_key = response['SecretString'].encode('utf8')
region = secrets.meta.region_name
Config.cache = Config(endpoint, cert, key, region)
Config.cache = Config(endpoint, cert, key, region, gg_cert, gg_key)
except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as ex:
print(ex)
raise unittest.SkipTest("No credentials")

return Config.cache


def create_client_id():
return 'aws-crt-python-unit-test-{0}'.format(uuid.uuid4())
return 'aws-iot-device-sdk-python-v2-unit-test-{0}'.format(uuid.uuid4())


class SamplesTest(unittest.TestCase):
Expand Down Expand Up @@ -115,7 +128,7 @@ def test_basic_discovery_response_only(self):
"--region", config.region,
"--cert", config.cert_filepath,
"--key", config.key_filepath,
"--thing_name", "aws-sdk-crt-unit-test",
"--thing_name", "CI_Greengrass_Discovery_Thing",
"--verbosity", "Trace",
]

Expand Down

0 comments on commit 119ad10

Please sign in to comment.