more pylint changes

Author: rpcme

src/bulk_importer/main.py

@@ -5,7 +5,7 @@
 Lambda function to import certificate, construct IoT Thing, and associate
 the Thing, Policy, Certificate, Thing Type, and Thing Group
 """
-
+import ast
 import base64
 import json
 import binascii
@@ -163,7 +163,7 @@ def process_certificate(config, requeue_cb):
        TODO: This should be simplified"""
     iot_client = boto3client('iot')
     payload = config['certificate']
-    certificate_text = base64.b64decode(eval(payload))
+    certificate_text = base64.b64decode(ast.literal_eval(payload))
 
     # See if the certificate has already been registered.  If so, bail.
     certificate_obj = x509.load_pem_x509_certificate(data=certificate_text,

src/provider_espressif/main.py

@@ -1,12 +1,17 @@
+"""
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: MIT-0
+
+Lambda function to decompose Espressif based certificate manifest(s) and begin
+the import processing pipeline
+"""
 import os
 import io
 import json
 import csv
 from base64 import b64encode
 import botocore
-from boto3 import resource, client, s3
-from moto import mock_aws, settings
-from aws_lambda_powertools.utilities.validation import validate
+from boto3 import resource, client
 from cryptography import x509
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import serialization
@@ -62,6 +67,6 @@ def lambda_handler(event, context):
     """Lambda function main entry point"""
     queue_url = os.environ['QUEUE_TARGET']
     bucket_name = event['Records'][0]['s3']['bucket']['name']
-    manifest_name = event['Records'][0]['s3']['object']['key'] 
+    manifest_name = event['Records'][0]['s3']['object']['key']
 
     invoke_export(bucket_name, manifest_name, queue_url)

src/provider_infineon/main.py

@@ -1,70 +1,82 @@
+"""
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: MIT-0
+
+Lambda function to decompose Infineon based certificate manifest(s) and begin
+the import processing pipeline
+"""
 import os
 import io
 import json
+from xml.etree import ElementTree
+from base64 import b64encode
 from botocore import exceptions as botoexceptions
 from boto3 import resource as boto3resource, client as boto3client
-import binascii
-from xml.etree import ElementTree
 from cryptography import x509
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import serialization
-from base64 import b64encode
 
-# Given a bucket and object, verify its existence and return the resource.
 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')
     res = s3.Object(bucket_name=bucket_name, key=object_name)
-    try: 
+    try:
         fs = io.BytesIO()
         res.download_fileobj(fs)
         return fs
     except botoexceptions.ClientError as ce:
         raise ce
 
-# Given a bucket name and object name, return bytes representing
-# the object content.
+
 def s3_filebuf_bytes(bucket_name: str, object_name: str):
+    """ Given a bucket name and object name, return bytes representing
+        the object content."""
     object_stream = s3_object_stream(bucket_name=bucket_name,
                                      object_name=object_name)
     return object_stream.getvalue()
 
-def format_certificate(certString):
-    encodedCert = certString.encode('ascii')
+def format_certificate(cert_string):
+    """Encode certificate so that it can safely travel via sqs"""
+    cert_encoded = cert_string.encode('ascii')
 
-    pem_obj = x509.load_pem_x509_certificate(encodedCert,
+    pem_obj = x509.load_pem_x509_certificate(cert_encoded,
                                              backend=default_backend())
     block = pem_obj.public_bytes(encoding=serialization.Encoding.PEM).decode('ascii')
-    return {'certificate': str(b64encode(block.encode('ascii')))}
+    return str(b64encode(block.encode('ascii')))
 
-    
+def queue_certificate(identity, certificate, queue_url):
+    """Send the thing name and certificate to sqs queue"""
+    sqs_client = boto3client("sqs")
+    payload = {
+        'thing': identity,
+        'certificate': certificate
+    }
+    sqs_client.send_message( QueueUrl=queue_url,
+                             MessageBody=json.dumps(payload) )
 
-def invoke_export(manifest, queueUrl):
-    client = boto3client("sqs")
-    
+def invoke_export(manifest, queue_url):
+    """Function to Iterate through the certificate list and queue for processing"""
     root = ElementTree.fromstring(manifest)
 
     for group in root.findall('group'): # /binaryhex
         thing_name = ''
 
+        # TODO: Evaluate what happens when this fails
         for hex_element in group.findall('hex'):
             if hex_element.get('name') == 'TpmMAC':
                 thing_name = hex_element.get('value')
 
-        # There can be more than one certificate
         for hexdata_element in group.findall('binaryhex'):
             certificate_data = format_certificate(hexdata_element.text)
-            # Need to send each certificate separately
-            certificate_data['thing'] = thing_name
-            print(certificate_data)
-            client.send_message( QueueUrl=queueUrl,
-                                 MessageBody=json.dumps(certificate_data) )
-        
+            queue_certificate(thing_name, certificate_data, queue_url)
+
 def lambda_handler(event, context):
-    queueUrl = os.environ['QUEUE_TARGET']
+    """Lambda function main entry point"""
+    queue_url = os.environ['QUEUE_TARGET']
 
     bucket = event['Records'][0]['s3']['bucket']['name']
-    manifest = event['Records'][0]['s3']['object']['key'] 
+    manifest = event['Records'][0]['s3']['object']['key']
 
-    manifestContent = s3_filebuf_bytes(bucket, manifest)
+    manifest_content = s3_filebuf_bytes(bucket, manifest)
 
-    invoke_export(manifestContent, queueUrl)
+    invoke_export(manifest_content, queue_url)

test/unit/src/test_bulk_importer.py

@@ -4,29 +4,23 @@
 
 Unit tests for bulk_importer
 """
-import sys
 import os
-#import io
+import base64
 from unittest import TestCase
-#from unittest.mock import MagicMock, patch
-import pytest
-
-import botocore
+from moto import mock_aws
+from cryptography import x509
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization
 
 from boto3 import resource, client
-from moto import mock_aws, settings
-from moto.settings import iot_use_valid_cert
+from src.bulk_importer.testable import LambdaSQSClass
+from src.bulk_importer.main import get_certificate_fingerprint, requeue, process_certificate
+#    from src.bulk_importer.main import lambda_handler, get_certificate, get_thing, get_policy
+#    from src.bulk_importer.main import get_certificate_arn, get_thing_group, get_thing_type
+#    from src.bulk_importer.main import process_policy, process_thing
+#    from src.bulk_importer.main import process_thing_group, get_name_from_certificate, process_sqs
 
-from aws_lambda_powertools.utilities.validation import validate
-
-sys.path.append('./src/bulk_importer')
 os.environ['AWS_DEFAULT_REGION'] = "us-east-1"
-from src.bulk_importer.testable import LambdaSQSClass   # pylint: disable=wrong-import-position
-from src.bulk_importer.main import lambda_handler, get_certificate, get_certificate_fingerprint, get_certificate_arn, get_thing, get_policy, get_thing_group, get_thing_type, process_policy, process_thing, requeue, process_certificate, process_thing_group, get_name_from_certificate, process_sqs # pylint: disable=wrong-import-position
-from cryptography import x509
-from cryptography.hazmat.backends import default_backend
-from cryptography.hazmat.primitives import serialization
-import base64
 
 @mock_aws(config={
     "core": {
@@ -36,6 +30,7 @@
     },
     'iot': {'use_valid_cert': True}})
 class TestBulkImporter(TestCase):
+    """Test cases for bulk importer lambda function"""
     def setUp(self):
         self.test_sqs_queue_name = "provider"
         sqs_client = client('sqs', region_name="us-east-1")
@@ -46,14 +41,15 @@ def setUp(self):
         self.mocked_sqs_class = LambdaSQSClass(mocked_sqs_resource)
 
     def test_pos_process_certificate(self):
+        """Positive test case for processing certificate"""
         with open('./test/artifacts/single.pem', 'rb') as data:
             pem_obj = x509.load_pem_x509_certificate(data.read(),
                                                      backend=default_backend())
             block = pem_obj.public_bytes(encoding=serialization.Encoding.PEM).decode('ascii')
             cert = str(base64.b64encode(block.encode('ascii')))
             c = {'certificate': cert}
             r = process_certificate(c, requeue)
-            assert (r == get_certificate_fingerprint(pem_obj))
+            assert r == get_certificate_fingerprint(pem_obj)
 
     def tearDown(self):
         sqs_resource = resource("sqs", region_name="us-east-1")