More pylint tidying

Author: rpcme

src/bulk_importer/main.py

@@ -18,6 +18,8 @@
 from cryptography.x509.oid import NameOID
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import hashes
+from aws_lambda_powertools.utilities.typing import LambdaContext
+from aws_lambda_powertools.utilities.data_classes import SQSEvent
 
 logger = logging.getLogger()
 logger.setLevel("INFO")
@@ -58,7 +60,8 @@ def get_thing(thing_name):
         response = iot_client.describe_thing(thingName=thing_name)
         return response.get("thingArn")
     except ClientError as error:
-        assert error.response['Error']['Code'] == 'ResourceNotFoundException'
+        error_code = error.response['Error']['Code']
+        assert error_code == 'ResourceNotFoundException'
         return None
 
 def get_policy(policy_name):
@@ -67,12 +70,13 @@ def get_policy(policy_name):
     try:
         response = iot_client.get_policy(policyName=policy_name)
         return response.get('policyArn')
-    except botocore.exceptions.ClientError as error:
-        if error.response['Error']['Code'] == 'ResourceNotFoundException':
-            print("ERROR: You need to configure the policy [" + policy_name +
-                  "] in your target region first.")
-        if error.response['Error']['Code'] == 'UnauthorizedException':
-            print("ERROR: There is a deployment problem with the attached Role."
+    except ClientError as error:
+        error_code = error.response['Error']['Code']
+        if error_code == 'ResourceNotFoundException':
+            logger.error("ERROR: You need to configure the policy {policy_name} "
+                         "in your target region first.")
+        if error_code == 'UnauthorizedException':
+            logger.error("There is a deployment problem with the attached Role."
                   "Unable to reach IoT Core object.")
         return None
 
@@ -83,13 +87,14 @@ def get_thing_group(thing_group_name):
     try:
         response = iot_client.describe_thing_group(thingGroupName=thing_group_name)
         return response.get('thingGroupArn')
-    except botocore.exceptions.ClientError as error:
-        if error.response['Error']['Code'] == 'ResourceNotFoundException':
-            print("ERROR: You need to configure the Thing Group [" + thing_group_name +
-                  "] in your target region first.")
-        if error.response['Error']['Code'] == 'UnauthorizedException':
-            print("ERROR: There is a deployment problem with the attached Role. Unable"
-                  "to reach IoT Core object.")
+    except ClientError as error:
+        error_code = error.response['Error']['Code']
+        if 'ResourceNotFoundException' == error_code:
+            logger.error("You need to configure the Thing Group {thing_group_name} "
+                         "in your target region first.")
+        if 'UnauthorizedException' == error_code:
+            logger.error("There is a deployment problem with the attached Role. Unable"
+                         "to reach IoT Core object.")
         return None
 
 def get_thing_type(type_name):
@@ -99,12 +104,13 @@ def get_thing_type(type_name):
         response = iot_client.describeThingType(thingTypeName=type_name)
         return response.get('thingTypeArn')
     except ClientError as error:
-        if error.response['Error']['Code'] == 'ResourceNotFoundException':
-            print("ERROR: You need to configure the Thing Type [" + type_name +
-                  "] in your target region first.")
-        if error.response['Error']['Code'] == 'UnauthorizedException':
-            print("ERROR: There is a deployment problem with the attached Role."
-                  "Unable to reach IoT Core object.")
+        error_code = error.response['Error']['Code']
+        if 'ResourceNotFoundException' == error_code:
+            logger.error("You need to configure the Thing Type {type_name} "
+                         "in your target region first.")
+        if 'UnauthorizedException' == error_code:
+            logger.error("There is a deployment problem with the attached Role."
+                         "Unable to reach IoT Core object.")
         return None
 
 def process_policy(policy_name, certificate_id):
@@ -121,7 +127,8 @@ def process_thing(thing_name, certificate_id, thing_type_name):
         iot_client.describe_thing(thingName=thing_name)
         return None
     except ClientError as error:
-        print("Thing not found (" + error.response['Error']['Code'] + "). Creating.")
+        error_code = error.response['Error']['Code']
+        logger.info("Thing not found {error_code}. Creating.")
 
     # Create thing
     try:
@@ -132,8 +139,8 @@ def process_thing(thing_name, certificate_id, thing_type_name):
                                     thingTypeName=thing_type_name)
 
     except ClientError as error:
-        print("ERROR Thing creation failed.")
-        print(error)
+        error_code = error.response['Error']['Code']
+        logger.error("ERROR Thing creation failed: {error_code}")
         return None
 
     try:
@@ -182,11 +189,9 @@ def process_certificate(config, requeue_cb):
                         fingerprint, error.response['Error']['Code'])
 
     try:
-        print("Importing certificate.")
         response = iot_client.register_certificate_without_ca(
             certificatePem=certificate_text.decode('ascii'),
             status='ACTIVE')
-        print("Certificate imported.")
         return response.get("certificateId")
     except botocore.exceptions.ClientError as e:
         if e.response['Error']['Code'] == 'ThrottlingException':
@@ -248,7 +253,7 @@ def process_sqs(config):
     process_policy(policy_name, certificate_id)
     process_thing_group(thing_group_name, thing_name)
 
-def lambda_handler(event, context):
+def lambda_handler(event: SQSEvent, context: LambdaContext):
     """Lambda function main entry point"""
     if event.get('Records') is None:
         print("ERROR: Configuration incorrect: no event record on invoke")
@@ -264,9 +269,4 @@ def lambda_handler(event, context):
             config = json.loads(record["body"])
             process_sqs(config)
 
-    return {
-        "statusCode": 204,
-        "body": json.dumps({
-            "message": "Processing succeeded."
-        })
-    }
+    return event

src/provider_infineon/main.py

@@ -18,8 +18,6 @@
 from aws_lambda_powertools.utilities.typing import LambdaContext
 from aws_lambda_powertools.utilities.data_classes import S3Event, event_source
 
-from . import schemas
-
 def s3_object_stream(bucket_name: str, object_name: str):
     """Given a bucket and object, verify its existence and return the resource."""
     s3 = boto3resource('s3')

test/unit/src/test_provider_espressif.py

@@ -22,9 +22,9 @@
 from moto import mock_aws, settings
 from aws_lambda_powertools.utilities.validation import validate
 
-from src.provider_espressif.testable import LambdaS3Class, LambdaSQSClass   # pylint: disable=wrong-import-position
 from src.provider_espressif.main import lambda_handler, s3_filebuf_bytes, invoke_export  # pylint: disable=wrong-import-position
 from src.provider_espressif.main import s3_object_stream
+from .model_provider_espressif import LambdaS3Class, LambdaSQSClass   # pylint: disable=wrong-import-position
 
 @mock_aws(config={
     "core": {
@@ -63,18 +63,21 @@ def test_pos_s3_object_resource(self):
 
     def test_neg_s3_object_resource(self):
         with pytest.raises(botocore.exceptions.ClientError) as e:
+            # Although this returns a value, no need to define var for it
             r = s3_object_stream("unit_test_s3_buckets", "manifest")
-        assert str(e.value) == "An error occurred (NoSuchBucket) when calling the HeadObject operation: The specified bucket does not exist"
+        assert str(e.value) == "An error occurred (NoSuchBucket) when calling " \
+                               "the HeadObject operation: The specified bucket " \
+                               "does not exist"
 
     def test_pos_s3_filebuf_bytes(self):
-        # The bytes should equal to the object in the bucket
+        """ The bytes should equal to the object in the bucket """
         v = s3_filebuf_bytes("unit_test_s3_bucket", "manifest.csv")
         assert v == self.test_s3_object_content.read()
 
     def test_pos_invoke_export(self):
+        """ The number of items in the queue should be 7 since there are
+            seven certificates in the test file """
         invoke_export("unit_test_s3_bucket", "manifest.csv", "provider")
-        # The number of items in the queue should be 7 since there are
-        # seven certificates in the test file
         sqs_client = client("sqs", "us-east-1")
         sqs_queue_url_r = sqs_client.get_queue_url(QueueName=self.test_sqs_queue_name)
         sqs_queue_url = sqs_queue_url_r['QueueUrl']

test/unit/src/test_provider_infineon.py

@@ -18,9 +18,9 @@
 #from moto import mock_aws, settings
 #from unittest.mock import MagicMock, patch
 
-from src.provider_infineon.testable import LambdaS3Class, LambdaSQSClass
 from src.provider_infineon.main import s3_object_stream, s3_filebuf_bytes
 from src.provider_infineon.main import lambda_handler, invoke_export
+from .model_provider_infineon import LambdaS3Class, LambdaSQSClass
 
 @mock_aws(config={
     "core": {
@@ -39,7 +39,8 @@ def setUp(self):
         s3_client.create_bucket(Bucket = self.test_s3_bucket_name )
         with open('./test/artifacts/manifest-espressif.csv', 'rb') as data:
             s3_client.put_object(Bucket=self.test_s3_bucket_name, Key="manifest.csv", Body=data)
-            self.test_s3_object_content = s3_client.get_object(Bucket=self.test_s3_bucket_name, Key="manifest.csv")['Body']
+            self.test_s3_object_content = s3_client.get_object(Bucket=self.test_s3_bucket_name,
+                                                               Key="manifest.csv")['Body']
         mocked_s3_resource = resource("s3")
         mocked_s3_resource = { "resource" : resource('s3'),
                                "bucket_name" : self.test_s3_bucket_name }
@@ -54,15 +55,21 @@ def setUp(self):
         self.mocked_sqs_class = LambdaSQSClass(mocked_sqs_resource)
 
     def test_pos_s3_object_resource(self):
+        """Basic pos test case for object resource"""
         r = s3_object_stream("unit_test_s3_bucket", "manifest.csv")
         assert isinstance(r, io.BytesIO)
 
     def test_neg_s3_object_resource(self):
+        """Basic neg test case for object resource"""
         with pytest.raises(botocore.exceptions.ClientError) as e:
-            r = s3_object_stream("unit_test_s3_buckets", "manifest")
-        assert str(e.value) == "An error occurred (NoSuchBucket) when calling the HeadObject operation: The specified bucket does not exist"
+            # Although this returns a value, no need to define var for it
+            s3_object_stream("unit_test_s3_buckets", "manifest")
+        errstr = "An error occurred (NoSuchBucket) when calling the " \
+                 "HeadObject operation: The specified bucket does not exist"
+        assert str(e.value) == errstr
 
     def test_pos_s3_filebuf_bytes(self):
+        """Basic pos test case for byte buffer handling"""
         # The bytes should equal to the object in the bucket
         v = s3_filebuf_bytes("unit_test_s3_bucket", "manifest.csv")
         assert v == self.test_s3_object_content.read()