def cert_validation_test()

in provisioningHandler.py [0:0]

• 33 lines of code
• 3 McCabe index (conditional complexity)

    def cert_validation_test(self):
        event_loop_group = io.EventLoopGroup(1)
        host_resolver = io.DefaultHostResolver(event_loop_group)
        client_bootstrap = io.ClientBootstrap(event_loop_group, host_resolver)
        cpath = self.secure_cert_path.format(unique_id=self.unique_id)
        print("Connecting to production with credentials ({}, {}). ".format(self.new_key_name, self.new_cert_name))
        certpath = "{}/{}".format(cpath, self.new_cert_name)
        keypath = "{}/{}".format(cpath, self.new_key_name)
        
        cert = OpenSSL.crypto.load_certificate(
            OpenSSL.crypto.FILETYPE_PEM, 
            open(certpath).read()
        )
        print("t: {}".format(datetime.today().strftime('%Y-%m-%d-%H:%M:%S')))
        print("c: {}".format(cert.get_notBefore()))
        if os.path.isfile(certpath) and os.path.isfile(keypath):
            self.test_MQTTClient = mqtt_connection_builder.mtls_from_path(
                endpoint=self.iot_endpoint,
                cert_filepath=certpath,
                pri_key_filepath=keypath,
                client_bootstrap=client_bootstrap,
                ca_filepath="{}/{}".format(self.root_cert_path, self.root_cert),
                client_id=self.unique_id,
                clean_session=False,
                on_connection_interrupted=self.on_connection_interrupted,
                on_connection_resumed=self.on_connection_resumed,
                verify_peer=False,
                keep_alive_secs=6)
        else:
            exit()
        print("Connecting with Prod certs to {} with client ID '{}'...".format(self.iot_endpoint, self.unique_id))
        connect_future = self.test_MQTTClient.connect()
        # Future.result() waits until a result is available
        connect_future.result()
        print("Connected with Prod certs!")