TestQueue.py

import unittest
import subprocess
import os
import sys
import pymysql
from dotenv import load_dotenv, find_dotenv
import json

class TestQueue(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        # Set working directory to lambda-api
        os.chdir(os.path.dirname(os.path.abspath(__file__)) + "/../../backend/lambda-api")
        # Load database credentials to os.environ

        load_dotenv('./.env')
        # Make database connection
        connection = pymysql.connect(host='pickmo.cm8t4da2mqrz.us-west-2.rds.amazonaws.com',
                                 user=os.environ['RDS_USERNAME'],
                                 password=os.environ['RDS_PWD'],
                                 db='pickmo_testdb')
        # Clear PickerQueue
        try:
            with connection.cursor() as cursor:
                sql = "DELETE FROM PickerQueue"
                cursor.execute(sql)
                connection.commit()
        # End connection
        finally:
            connection.close()

    def test_queue_1(self):
        # Test adding picker to queue
        load_dotenv('./.env')
        # Make database connection
        connection = pymysql.connect(host='pickmo.cm8t4da2mqrz.us-west-2.rds.amazonaws.com',
                                 user=os.environ['RDS_USERNAME'],
                                 password=os.environ['RDS_PWD'],
                                 db='pickmo_testdb')
        # Invoke lambda code
        data = {
            'body':{
                'username':'new_user1'
            },
            'headers':{
                'test':'lambda'
            },
            'requestContext': {
                'authorizer': {
                    'claims': {
                        'email': 'new_user1'
                    }
                }
            }
        }
        bashCommand = "sls invoke local -f queuePicker --data '" + json.dumps(data) + "'"
        self.assertEqual(subprocess.check_call( bashCommand, shell=True), 0)

        # Check database for newly added picker
        try:
            with connection.cursor() as cursor:
                sql = "SELECT * FROM PickerQueue WHERE username = 'new_user1'"
                cursor.execute(sql)    #
                result = cursor.fetchall()
                self.assertEqual(len(result), 1)
                self.assertEqual(result[0][0], 'new_user1')
        finally:
            connection.close()

    def test_queue_2(self):
        # Test removing picker from queue
        connection = pymysql.connect(host='pickmo.cm8t4da2mqrz.us-west-2.rds.amazonaws.com',
                                 user=os.environ['RDS_USERNAME'],
                                 password=os.environ['RDS_PWD'],
                                 db='pickmo_testdb')
        # Invoke lambda code
        data = {
            'pathParameters':{
                'username':'new_user1'
            },
            'headers':{
                'test':'lambda'
            },
            'requestContext': {
                'authorizer': {
                    'claims': {
                        'email': 'new_user1'
                    }
                }
            }
        }
        bashCommand = "sls invoke local -f dequeuePicker --data '" + json.dumps(data) + "'"
        self.assertEqual(subprocess.check_call( bashCommand, shell=True), 0)

        # Check database for deleted picker
        try:
            with connection.cursor() as cursor:
                sql = "SELECT * FROM PickerQueue WHERE username = 'new_user1'"
                cursor.execute(sql)    #
                result = cursor.fetchall()
                self.assertEqual(len(result), 0)
        finally:
            connection.close()

if __name__ == '__main__':
    unittest.main()