TestGetFlaggedImages.py

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

# The tests I wrote extended Jacob's test. I wrote the tests test_get_flagged_images_3, test_get_flagged_images_4, test_get_flagged_images_6. But I also came up with the way of testing serverless lambda functions locally, which uses the python subprocess library. This makes it so we emulate exactly how the lambda functions will be called by the API Gateway.

class TestGetImage(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 Tables
        try:
            with connection.cursor() as cursor:
                sql = "DELETE FROM Connections"
                cursor.execute(sql)
                connection.commit()
            with connection.cursor() as cursor:
                sql = "DELETE FROM Robots"
                cursor.execute(sql)
                connection.commit()
            with connection.cursor() as cursor:
                sql = "DELETE FROM FlaggedImages"
                cursor.execute(sql)
                connection.commit()
            with connection.cursor() as cursor:
                sql = "DELETE FROM Flags"
                cursor.execute(sql)
                connection.commit()
            with connection.cursor() as cursor:
                sql = "DELETE FROM FlagCoordinates"
                cursor.execute(sql)
                connection.commit()
            with connection.cursor() as cursor:
                sql = "DELETE FROM RandomSamplingInfo"
                cursor.execute(sql)
                connection.commit()
        # Write test data to use for tests
            with connection.cursor() as cursor:
                sql = "INSERT INTO RandomSamplingInfo " \
                      "(id, percentage) " \
                      "VALUES (0, 50)"
                cursor.execute(sql)
                connection.commit()
            with connection.cursor() as cursor:
                sql = "INSERT INTO Robots (id) VALUES ('Robot1')"
                cursor.execute(sql)
                connection.commit()
            with connection.cursor() as cursor:
                sql = "INSERT INTO Connections (username, robotID) VALUES ('new_user1','Robot1')"
                cursor.execute(sql)
                connection.commit()
            with connection.cursor() as cursor:
                sql = "INSERT INTO Flags " \
                      "(flagName) " \
                      "VALUES ('flag1')"
                cursor.execute(sql)
                connection.commit()
            with connection.cursor() as cursor:
                sql = "SELECT id FROM Flags WHERE flagName = 'flag1'"
                cursor.execute(sql)
                flag_id = cursor.fetchone()[0]
            with connection.cursor() as cursor:
                sql = "INSERT INTO FlagCoordinates " \
                      "(line1AX, line1AY, line1BX, line1BY, line2AX, line2AY, line2BX, line2BY, circleX, circleY) " \
                      "VALUES (1, 2, 3, 4, 5, 6, 7, 8, 9, 10)"
                cursor.execute(sql)
                connection.commit()
            with connection.cursor() as cursor:
                sql = "SELECT id FROM FlagCoordinates"
                cursor.execute(sql)
                flag_coords_id = cursor.fetchone()[0]
            with connection.cursor() as cursor:
                sql = "INSERT INTO FlaggedImages (username, timestamp, flagID, imgPath, flagCoordinatesID, lat, lon) " \
                      "VALUES ('new_user1', 12345, " + str(flag_id) + ", 'Robot1/2018 07 07 - 17:02:1518023271.png', " + str(flag_coords_id)+ ", 5, 7)"
                cursor.execute(sql)
                connection.commit()
            with connection.cursor() as cursor:
                sql = "INSERT INTO FlaggedImages (username, timestamp, flagID, imgPath, flagCoordinatesID, lat, lon) " \
                      "VALUES ('new_user2', 32345, " + str(flag_id) + ", 'Robot1/2018 08 06 - 17:02:1518023271.png', " + str(flag_coords_id)+ ", 5, 7)"
                cursor.execute(sql)
                connection.commit()
        # End connection
        finally:
            connection.close()

    def test_get_flagged_images_3(self):
        # Test getting flagged images of a certain sample percentage
        data = {
            "headers": {
                "test": "lambda"
            },
            "queryStringParameters": {
                "sample": 50
            }
        }
        bash_command = "sls invoke local -f getFlaggedImages --data '" + json.dumps(data) + "' > getFImagesRes.json"
        self.assertEqual(subprocess.check_call( bash_command, shell=True), 0)

        with open('getFImagesRes.json', 'r') as f:
            data = json.load(f)
        self.assertEqual(len(data["body"]), 1)
        # Remove the response file
        os.remove('getFImagesRes.json')

    def test_get_flagged_images_4(self):
        # Test getting flagged images of a certain sample percentage
        data = {
            "headers": {
                "test": "lambda"
            },
            "queryStringParameters": {
                "flag": "flag1"
            }
        }
        bash_command = "sls invoke local -f getFlaggedImages --data '" + json.dumps(data) + "' > getFImagesRes.json"
        self.assertEqual(subprocess.check_call( bash_command, shell=True), 0)

        with open('getFImagesRes.json', 'r') as f:
            data = json.load(f)
        self.assertEqual(len(data["body"]), 2)
        # Remove the response file
        os.remove('getFImagesRes.json')

    def test_get_flagged_images_5(self):
        # Test getting flagged images of a certain sample percentage
        data = {
            "headers": {
                "test": "lambda"
            },
            "queryStringParameters": {
                "flag": "flag2"
            }
        }
        bash_command = "sls invoke local -f getFlaggedImages --data '" + json.dumps(data) + "' > getFImagesRes.json"
        self.assertEqual(subprocess.check_call( bash_command, shell=True), 0)

        with open('getFImagesRes.json', 'r') as f:
            data = json.load(f)
        self.assertEqual(len(data["body"]), 0)
        # Remove the response file
        os.remove('getFImagesRes.json')

    def test_get_flagged_images_6(self):
        # Test getting flagged images of a certain sample percentage
        data = {
            "headers": {
                "test": "lambda"
            },
            "queryStringParameters": {
                "sample": 50,
                "flag": "flag1"
            }
        }
        bash_command = "sls invoke local -f getFlaggedImages --data '" + json.dumps(data) + "' > getFImagesRes.json"
        self.assertEqual(subprocess.check_call( bash_command, shell=True), 0)

        with open('getFImagesRes.json', 'r') as f:
            data = json.load(f)
        self.assertEqual(len(data["body"]), 1)
        # Remove the response file
        os.remove('getFImagesRes.json')

    @classmethod
    def tearDownClass(cls):
        # 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 test data
        try:
            with connection.cursor() as cursor:
                sql = "DELETE FROM Connections"
                cursor.execute(sql)
                connection.commit()
            with connection.cursor() as cursor:
                sql = "DELETE FROM Robots"
                cursor.execute(sql)
                connection.commit()
            with connection.cursor() as cursor:
                sql = "DELETE FROM FlaggedImages"
                cursor.execute(sql)
                connection.commit()
            with connection.cursor() as cursor:
                sql = "DELETE FROM Flags"
                cursor.execute(sql)
                connection.commit()
            with connection.cursor() as cursor:
                sql = "DELETE FROM FlagCoordinates"
                cursor.execute(sql)
                connection.commit()
        # End connection
        finally:
            connection.close()

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