test_samples.py

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0.
importawsiot
importboto3
importbotocore.exceptions
importos.path
importshutil
importsubprocess
importsys
importtempfile
importunittest
importuuid
importwarnings


classConfig:
cache=None

def__init__(self, endpoint, cert, key, gg_cert, gg_key, region):
self.endpoint=endpoint
self.region=region
self.cert_bytes=cert
self.key_bytes=key

self._tmp_dirpath=tempfile.mkdtemp()
self.cert_filepath=os.path.join(self._tmp_dirpath, 'certificate.pem')
withopen(self.cert_filepath, 'wb') ascert_file:
cert_file.write(cert)

self.key_filepath=os.path.join(self._tmp_dirpath, 'privatekey.pem')
withopen(self.key_filepath, 'wb') askey_file:
key_file.write(key)

self.gg_cert_filepath=os.path.join(self._tmp_dirpath, 'gg_certificate.pem')
withopen(self.gg_cert_filepath, 'wb') ascert_file:
cert_file.write(gg_cert)

self.gg_key_filepath=os.path.join(self._tmp_dirpath, 'gg_privatekey.pem')
withopen(self.gg_key_filepath, 'wb') askey_file:
key_file.write(gg_key)

def__del__(self):
shutil.rmtree(self._tmp_dirpath)

@staticmethod
defget():
"""Raises SkipTest if credentials aren't set up correctly"""
ifConfig.cache:
returnConfig.cache

# boto3 caches the HTTPS connection for the API calls, which appears to the unit test
# framework as a leak, so ignore it, that's not what we're testing here
warnings.simplefilter('ignore', ResourceWarning)

try:
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='ci/mqtt5/us/mqtt5_thing/cert')
cert=response['SecretString'].encode('utf8')
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, gg_cert, gg_key)
except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) asex:
print(ex)
raiseunittest.SkipTest("No credentials")

returnConfig.cache


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


classSamplesTest(unittest.TestCase):

def_run(self, args, stdout_checker):
process=subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout, stderr=process.communicate()
stdout=stdout.decode()
try:
self.assertEqual(0, process.returncode)
stdout_checker(stdout)
exceptExceptionase:
# print output and rethrow exception
print(subprocess.list2cmdline(args))
print("--- stdout ---")
forlineinstdout.splitlines():
print(line)
print("--- stderr ---")
forlineinstderr.splitlines():
print(line)
print("--- end ---")

raisee

deftest_pubsub(self):
config=Config.get()
args= [
sys.executable,
"samples/pubsub.py",
"--endpoint", config.endpoint,
"--cert", config.cert_filepath,
"--key", config.key_filepath,
"--client_id", create_client_id(),
"--count", "1",
"--verbosity", "Trace",
 ]

defstdout_checker(stdout):
# check for last line printed by sample
last_line=stdout.splitlines()[-1]
self.assertTrue(last_line.startswith("Disconnected!"))

self._run(args, stdout_checker)

deftest_basic_discovery_response_only(self):
config=Config.get()
args= [
sys.executable,
"samples/basic_discovery.py",
"--print_discover_resp_only",
"--region", config.region,
"--cert", config.cert_filepath,
"--key", config.key_filepath,
"--thing_name", "CI_Greengrass_Discovery_Thing",
"--verbosity", "Trace",
 ]

defstdout_checker(stdout):
# check for last line printed by sample
last_line=stdout.splitlines()[-1]
self.assertTrue(last_line.startswith("awsiot.greengrass_discovery.DiscoverResponse("))

self._run(args, stdout_checker)