Merge pull request #24 from markscamilleri/TestingBackendDavid

Backend Testing script

.travis.yml

@@ -1,6 +1,8 @@
 language: node_js
 node_js:
-- 10.17.0
+  - 10.17.0
+python:
+  - 3.6.7
 env:
   matrix:
     secure: gSb+5hzjOEsBsrJRIb8x2pQF/Y/yB4b1d1OGW2BkXCLECwFfk/UCCpE1LIWLeK2XJpuF9H/yewZFc2lrgYsM1NI1Wn33BcOFWuvCq3uyLFlpdiB5jKNXvLYA6xmudwbvWPh9AjubN2IJplatF2EAMyf1JjgOX+M3lEWOGqiaOYMLjP+8PphInhbscMDWJb2ni601Q+iBDtEkhwLPWVIzz0gY3EyXta0YbtbiWvjhsFE9NYf2MuB/s9xb3UK09jq2bBtD4M80+ppzYUTmq8s3GzTPqpPm0vww67xkMUPA4tkgygFNwYYETeytLL9bdyDY8PitJfGn26qoZTPSy6z2TOJEmEVhf077Tk25FrJcIjFltQ3Nne/NYczTwFqUrrXNoLPgplm4zMy3LZZKweL11juMmbXmyTZ3fywGqJ8RwKPEKO3Qeuv0xtsPiAc1qmh6bC5Y7E65G5rDj77ei+7BFc12nSjqBCa0rLx1iC47fzeGTwWKCIb6A6tJbr6mRVkRbcP7M9rQ/UMWYskrFLgGbvrgHX0Hn28uHS76CNBAwvh0eQk3iG2iOIZ9GVC32Lh66daLL9noDhTiDGUdmp6q7ofG8jtNDPvnqF6CXgUBgpE8GGDd6paTmHT4rIiKLUI42sQwOzfIhwt7exckspqyZvvZbnkYg/Yk7bR8vTHLZ0M=
@@ -11,6 +13,7 @@ cache:
 before_install:
 - npm install -g npm@latest
 - npm install -g expo-cli
+
 stages:
 - build
 - test
@@ -21,10 +24,23 @@ stages:
 jobs:
   include:
   - stage: test
+    before_install:
+      - cd ./backend
+      - pip3 install --upgrade pip
+      - pip3 install -r requirements.txt
+      - pip3 install pytest
+      - pip3 install pytest-timeout
+      - pip3 install codecov
     script:
-    - cd frontend
-    - npm ci
-    - npx jest --ci --passWithNoTests
+      - cd frontend
+      - npm ci
+      - npx jest --ci --passWithNoTests
+      - cd ../backend
+      - python3 -m pytest --timeout 600 test/test_SQLQueue.py
+    after_success:
+        - codecov # submit coverage
+
+
 #  - stage: deploy staging
 #    script:
 #    - cd frontend

backend/requirements.txt

@@ -0,0 +1,20 @@
+certifi==2019.11.28
+chardet==3.0.4
+Click==7.0
+coverage==4.5.4
+deprecation==2.0.7
+Flask==1.1.1
+idna==2.8
+itsdangerous==1.1.0
+janus==0.4.0
+Jinja2==2.10.3
+MarkupSafe==1.1.1
+mock==3.0.5
+mysql-connector==2.2.9
+packaging==19.2
+postcodes-io-api==0.0.4
+pyparsing==2.4.5
+requests==2.22.0
+six==1.13.0
+urllib3==1.25.7
+Werkzeug==0.16.0

backend/src/app.py

@@ -0,0 +1,174 @@
+from flask import Flask, request, jsonify
+import postcodes_io_api
+import database
+import hashlib
+import mysql.connector
+import secrets
+
+#gpsDatabase = mysql.connector.connect(  # connect to db
+#        host="34.89.126.252",  # won't change
+#        user="root",  # username for rwx access
+#        password="[email protected]=100%",
+#        database="price_paid_data"  # schema name
+#)
+
+#dbex  = gpsDatabase.cursor()
+
+
+app = Flask(__name__)
+
+@app.route('/getHouses', methods=['POST'])
+def postcodesIO():
+    frontData = request.get_json()
+    latitude = frontData['lat']
+    longitude = frontData['lon']
+    radius = frontData['radius']
+    houseLimit = frontData['limit']
+    listOfPostcodes = callAPI(latitude, longitude, houseLimit, radius)
+    # houseLimit used here to limit len(listOfPostcodes) amount of values in the SQL WHERE clause.
+    varseq = ','.join(['%s']*len(listOfPostcodes))
+
+    statement = (
+        f"""SELECT post.id, house.paon, house.saon, post.street, post.postcode, props.initial, trans.price
+        FROM postcodes AS post
+        INNER JOIN houses AS house
+        ON house.postcode_id = post.id
+        INNER JOIN transactions AS trans
+        ON trans.house_id = house.id
+        INNER JOIN property_types AS props
+        ON props.id = house.property_type_id
+        WHERE post.postcode IN ({varseq});"""
+    )
+
+    print(statement)
+    print(str(listOfPostcodes))
+
+    result = db.select(query=statement, parameters=listOfPostcodes)
+    # --py-pseudo:
+    # for postcode in listOfPostcodes (not necessarily a for loop)
+    # --sql-pseudo:
+    # SELECT (postcode, id, street, county)
+    # FROM postcodes WHERE postcode IN str(tuple(listOfPostcodes))
+    # SELECT (paon, saon) FROM houses WHERE houses.id = postcodes.id (JOIN. DOUBLE CAUTION.)
+    # SELECT price FROM transactions WHERE transactions.id = houses.id (DOUBLE JOIN. TRIPLE CAUTION.)
+    # SELECT initial FROM property_types WHERE property_types.id = houses.id (If you don't get it now..)
+
+    print(str(result))
+
+    return jsonify(result)
+
+
+def callAPI(lat, lon, lim, rad):
+    api = postcodes_io_api.Api(debug_http=True)
+    listPostcodes = api.get_nearest_postcodes_for_coordinates(
+        latitude=lat, longitude=lon, limit=lim, radius=rad)
+    onlyPostcodes = []
+    for i in range(len(listPostcodes["result"])):
+        print(str(i))
+        onlyPostcodes.append(listPostcodes["result"][i]["postcode"])
+    return onlyPostcodes
+
+
+# this function is used when creating a new user so a salt can be made
+def passwordHash(password):
+    salt = secrets.token_hex(16)
+    saltedPass = password + salt
+    n = hashlib.sha256()
+    n.update(str.encode(saltedPass))
+    hash2 = n.hexdigest()
+    return {'hash2': hash2, 'salt': salt}
+
+
+def passwordCheckHash(password, salt):  # this function is used when checking hash2
+    m = hashlib.sha256()
+    saltedPass = password + salt
+    m.update(str.encode(saltedPass))
+    hash2 = m.hexdigest()
+    return hash2
+
+
+def getSalt(username):  # get the salt of a password from database
+    command = "SELECT salt FROM users WHERE username = %s"
+    result = db.select(query=command, parameters=(username,))
+    if not result:
+        return "EMPTY LIST"
+    else:
+        return result[0]['salt']  # TODO use dict cursor or SQLQueue and refer from there
+
+
+# This function should be used to check if a username has been taken or not on signup
+def usernameExists(username):
+    command = "SELECT * FROM users WHERE username = %s"
+    result = db.select(query=command, parameters=(username,))
+    print(result)
+    if len(result):
+        return True  # username has been taken
+    else:
+        return False  # username has't been taken
+
+
+def checkLogin(username, password):  # this checks for login details in table
+    command = "SELECT username, hash2 FROM users WHERE username = %s AND hash2 = %s"
+    result = db.select(query=command, parameters=[username, password])
+    if len(result):  # username and/or hash2 are correct as found in table
+        return True
+    else:  # username and/or hash2 are incorrect as not found in table
+        return False
+
+
+def addNewUser(username, hash2, salt):  # adds a new user into the table
+    command = "INSERT INTO users (username, hash2, salt) VALUES (%s, %s, %s)"
+    db.execute(query=command, parameters=(username, hash2, salt))
+   # newCommand = f"INSERT INTO users (username, hash2, salt) VALUES ('{username}', '{hash2}', '{salt}')"
+   # dbex.execute(newCommand)
+   # gpsDatabase.commit()
+   # command2 = "SELECT username, hash2, salt FROM users WHERE username = %s AND hash2 = %s AND salt = %s"
+   # result = db.select(query=command2, parameters=(username, hash2, salt))
+   # if(len(result)):
+   #     return True
+   # else:
+   #     return False
+
+
+@app.route('/login', methods=['POST'])
+def login():
+    data = request.get_json()
+    username = data['username']
+    hash1 = data['hashedPassword']
+    salt = getSalt(username)
+    hash2 = passwordCheckHash(hash1, salt)
+    if checkLogin(username, hash2):
+        # user proceeds to next screen as login details are correct
+        res = jsonify(response="True")  # {'response': 'True'}
+        return res  # login successful
+    else:
+        # show that the user has used incorrect details and needs to try again
+        res = jsonify(response="False")  # {'response': 'False'}
+        return res  # notification needed saying incorrect login details
+
+
+@app.route('/signup', methods=['POST'])
+def signup():
+    data = request.get_json()
+    username = data['username']
+    hash1 = data['hashedPassword']
+    if usernameExists(username):
+        # this block shows that the username already exists in the database and the user needs a different one
+        # ' {'response': 'True'}'  # notification needed saying try another username
+        res = jsonify(response="True")
+        return res
+    else:
+        # this block shows the username hasn't been taken and the new details are being added into the database
+        hashDict = passwordHash(hash1)
+        hash2 = hashDict.get('hash2')
+        salt = hashDict.get('salt')
+        addNewUser(username, hash2, salt)
+        # '{'response': 'False'}'  # notification needed saying account made
+        res = jsonify(response="False")
+        return res
+
+
+if __name__ == '__main__':
+    db = database.SQLQueue.get_instance(
+        host="34.89.126.252", user="root", password={change}, database="price_paid_data")
+    app.run(host='0.0.0.0', port=80)

backend/test/backendTesting.py

@@ -0,0 +1,61 @@
+import requests
+import json
+import unittest
+
+
+class TestRunningBackend(unittest.TestCase):
+    @staticmethod
+    def get_request(json):
+        headers = {'Content-type': 'application/json', 'Accept': 'application/json'}
+        r = requests.post("http://34.89.126.252/getHouses", data=json, headers=headers)
+        return r
+
+    def testCaseCorrectParams(self):
+        json_dicc = {"lat": 50.82838, "lon": -0.13947, "limit": 4, "radius": 2000}
+        json_data = json.dumps(json_dicc)
+        r = TestRunningBackend.get_request(json_data)
+        self.assertEquals(r.status_code, 200)
+
+    def testCaseCorrectParams2(self):
+        json_dicc = {"lat": 50.82838, "lon": -0.13947, "limit": 6, "radius": 4000}
+        json_data = json.dumps(json_dicc)
+        r = TestRunningBackend.get_request(json_data)
+        self.assertEquals(r.status_code, 200)
+
+    def testCaseLatOut(self):
+        json_dicc = {"lat": 50.0, "lon": -0.13947, "limit": 4, "radius": 2000}
+        json_data = json.dumps(json_dicc)
+        r = TestRunningBackend.get_request(json_data)
+        self.assertTrue(r.status_code == 200 or "error" in r.json().keys())
+
+    def testCaseLonOut(self):
+        json_dicc = {"lat": 50.82838, "lon": 0.081089, "limit": 4, "radius": 2000}
+        json_data = json.dumps(json_dicc)
+        r = TestRunningBackend.get_request(json_data)
+        self.assertTrue(r.status_code == 200 or "error" in r.json().keys())
+
+    def testCaseSmallRadius(self):
+        json_dicc = {"lat": 50.82838, "lon": -0.13947, "limit": 4, "radius": 2}
+        json_data = json.dumps(json_dicc)
+        r = TestRunningBackend.get_request(json_data)
+        self.assertTrue(r.status_code == 200 or "error" in r.json().keys())
+
+
+    def testCaseImpossibleLat(self):
+        json_dicc = {"lat": 91.0, "lon": -0.13947, "limit": 4, "radius": 2000}
+        json_data = json.dumps(json_dicc)
+        r = TestRunningBackend.get_request(json_data)
+        self.assertTrue(r.status_code == 200 or "error" in r.json().keys())
+
+    def testCaseImpossibleLon(self):
+        json_dicc = {"lat": 50.82838, "lon": -190.5, "limit": 4, "radius": 2000}
+        json_data = json.dumps(json_dicc)
+        r = TestRunningBackend.get_request(json_data)
+        self.assertTrue(r.status_code == 200 or "error" in r.json().keys())
+
+    def testCaseChangedParameters(self):
+        json_dicc = {"lat": 2000, "lon": -0.13947, "limit": 50.82838, "radius": 4}
+        json_data = json.dumps(json_dicc)
+        r = TestRunningBackend.get_request(json_data)
+        print(r)
+        self.assertTrue(r.status_code == 200 or "error" in r.json().keys())

backend/test/test_SQLQueue.py

@@ -0,0 +1,34 @@
+from unittest import TestCase, mock
+
+from backend.src.database import SQLQueue
+from backend.src.exceptions import SingletonException
+
+
+class TestSQLQueue(TestCase):
+    # @mock.patch('mysql.connector')
+    # def test_constructor_raises_exception(self, mock_connection):
+    #     sql_queue = SQLQueue(host='localhost')
+    #     self.assertRaises(SingletonException, SQLQueue, host="localhost")
+    #     del sql_queue
+
+    @mock.patch('mysql.connector')
+    def test_get_instance_first_time_gets_new_instance(self, mock_connection):
+        sql_queue = SQLQueue.get_instance(host="localhost", user="root", password="root")
+        self.assertIsInstance(sql_queue, SQLQueue)
+        del sql_queue
+
+    @mock.patch('mysql.connector')
+    def test_get_instance_second_time_same_gets_same_instance(self, mock_connection):
+        sql_queue_1 = SQLQueue.get_instance(host="localhost", user="root", password="root")
+        sql_queue_2 = SQLQueue.get_instance(host="localhost", user="root", password="root")
+        self.assertEqual(sql_queue_1, sql_queue_2)
+        del sql_queue_1
+        del sql_queue_2
+
+    @mock.patch('mysql.connector')
+    def test_get_instance_second_time_different_gets_new_instance(self, mock_connection):
+        sql_queue_1 = SQLQueue.get_instance(host="localhost", user="root", password="root")
+        sql_queue_2 = SQLQueue.get_instance(host="localhost", user="test", password="root")
+        self.assertNotEqual(sql_queue_1, sql_queue_2)
+        del sql_queue_1
+        del sql_queue_2