SHARE
TWEET

TestQueue.py

a guest Mar 16th, 2018 137 Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
  1. import unittest
  2. import subprocess
  3. import os
  4. import sys
  5. import pymysql
  6. from dotenv import load_dotenv, find_dotenv
  7. import json
  8.  
  9.  
  10. class TestQueue(unittest.TestCase):
  11.  
  12.     @classmethod
  13.     def setUpClass(cls):
  14.         # Set working directory to lambda-api
  15.         os.chdir(os.path.dirname(os.path.abspath(__file__)) +
  16.                  "/../../backend/lambda-api")
  17.         # Load database credentials to os.environ
  18.  
  19.         load_dotenv('./.env')
  20.         # Make database connection
  21.         connection = pymysql.connect(host='pickmo.cm8t4da2mqrz.us-west-2.rds.amazonaws.com',
  22.                                      user=os.environ['RDS_USERNAME'],
  23.                                      password=os.environ['RDS_PWD'],
  24.                                      db='pickmo_testdb')
  25.         # Clear PickerQueue
  26.         try:
  27.             with connection.cursor() as cursor:
  28.                 sql = "DELETE FROM PickerQueue"
  29.                 cursor.execute(sql)
  30.                 connection.commit()
  31.         # End connection
  32.         finally:
  33.             connection.close()
  34.  
  35.     def test_add_remove(self):
  36.         username = 'new_user1'
  37.         # Test adding picker to queue and then removing that picker
  38.         load_dotenv('./.env')
  39.         # Make database connection
  40.         connection = pymysql.connect(host='pickmo.cm8t4da2mqrz.us-west-2.rds.amazonaws.com',
  41.                                      user=os.environ['RDS_USERNAME'],
  42.                                      password=os.environ['RDS_PWD'],
  43.                                      db='pickmo_testdb')
  44.  
  45.         #-----------------------First Test-------------------------#
  46.         # Invoke lambda code
  47.         data = {
  48.             'body': {
  49.                 'username': username
  50.             },
  51.             'headers': {
  52.                 'test': 'lambda'
  53.             },
  54.             'requestContext': {
  55.                 'authorizer': {
  56.                     'claims': {
  57.                         'email': username
  58.                     }
  59.                 }
  60.             }
  61.         }
  62.         bash_command = "sls invoke local -f queuePicker --data '" + \
  63.             json.dumps(data) + "'"
  64.         self.assertEqual(subprocess.check_call(bash_command, shell=True), 0)
  65.  
  66.         # Check database for newly added picker
  67.         try:
  68.             with connection.cursor() as cursor:
  69.                 sql = "SELECT * FROM PickerQueue WHERE username = '" + username + "'"
  70.                 cursor.execute(sql)    #
  71.                 result = cursor.fetchall()
  72.                 self.assertEqual(len(result), 1)
  73.                 self.assertEqual(result[0][0], username)
  74.         finally:
  75.             connection.close()
  76.  
  77.         #-----------------------Second Test-------------------------#
  78.         data = {
  79.             'pathParameters': {
  80.                 'username': username
  81.             },
  82.             'headers': {
  83.                 'test': 'lambda'
  84.             },
  85.             'requestContext': {
  86.                 'authorizer': {
  87.                     'claims': {
  88.                         'email': username
  89.                     }
  90.                 }
  91.             }
  92.         }
  93.         bash_command = "sls invoke local -f dequeuePicker --data '" + \
  94.             json.dumps(data) + "'"
  95.         self.assertEqual(subprocess.check_call(bash_command, shell=True), 0)
  96.  
  97.         connection = pymysql.connect(host='pickmo.cm8t4da2mqrz.us-west-2.rds.amazonaws.com',
  98.                                      user=os.environ['RDS_USERNAME'],
  99.                                      password=os.environ['RDS_PWD'],
  100.                                      db='pickmo_testdb')
  101.  
  102.         # Check database for deleted picker
  103.         try:
  104.             with connection.cursor() as cursor:
  105.                 sql = "SELECT * FROM PickerQueue WHERE username = '" + username + "'"
  106.                 cursor.execute(sql)    #
  107.                 result = cursor.fetchall()
  108.                 self.assertEqual(len(result), 0)
  109.         finally:
  110.             connection.close()
  111.  
  112.     def test_error_duplicate_user(self):
  113.         username = 'new_user1'
  114.         # Test adding picker to queue and then removing that picker
  115.         load_dotenv('./.env')
  116.         # Make database connection
  117.         connection = pymysql.connect(host='pickmo.cm8t4da2mqrz.us-west-2.rds.amazonaws.com',
  118.                                      user=os.environ['RDS_USERNAME'],
  119.                                      password=os.environ['RDS_PWD'],
  120.                                      db='pickmo_testdb')
  121.  
  122.         # Invoke lambda code
  123.         data = {
  124.             'body': {
  125.                 'username': username
  126.             },
  127.             'headers': {
  128.                 'test': 'lambda'
  129.             },
  130.             'requestContext': {
  131.                 'authorizer': {
  132.                     'claims': {
  133.                         'email': username
  134.                     }
  135.                 }
  136.             }
  137.         }
  138.         # Add user
  139.         bash_command = "sls invoke local -f queuePicker --data '" + \
  140.             json.dumps(data) + "'"
  141.         self.assertEqual(subprocess.check_call(bash_command, shell=True), 0)
  142.  
  143.         # Add user again
  144.         bash_command = "sls invoke local -f queuePicker --data '" + \
  145.             json.dumps(data) + "' > QueueRes.json"
  146.         self.assertEqual(subprocess.check_call(bash_command, shell=True), 0)
  147.         with open('QueueRes.json', 'r') as f:
  148.             data = json.load(f)
  149.         self.assertEqual(data["statusCode"], 400)
  150.         # Remove the response file
  151.         os.remove('QueueRes.json')
  152.    
  153. if __name__ == '__main__':
  154.     unittest.main()
RAW Paste Data
We use cookies for various purposes including analytics. By continuing to use Pastebin, you agree to our use of cookies as described in the Cookies Policy. OK, I Understand
 
Top