CERC/api_v1.4
58
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Added tests for user

Browse Source
This commit is contained in:
Peter Yefi 2023-02-01 15:48:14 -05:00
parent 4b322148ba
commit d961241079
7 changed files with 227 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
bootstrap.py
View File

@ -54,4 +54,5 @@ def home():
return Response(headers={'Access-Control-Allow-Origin': '*'})
app.run(port=15789, host="0.0.0.0", debug=False)
if __name__ == '__main__':
app.run(port=15789, host="0.0.0.0", debug=False)

12
hub_api/city_info.py
View File

@ -15,6 +15,8 @@ from pathlib import Path
import os
from hub_api.config import Config
headers = {'Content-Type': 'application/json'}
class CityInfo(Resource, Config):
def __init__(self):
@ -32,8 +34,8 @@ class CityInfo(Resource, Config):
'upper_corner': city.upper_corner, 'created': city.created, 'updated': city.updated,
'user': {'id': city.user.id, 'name': city.user.name, 'email': city.user.email,
'role': city.user.role.value}
}, default=str), status=200)
return Response(response=json.dumps({'err_msg': 'City not found'}), status=404)
}, default=str), status=200, headers=headers)
return Response(response=json.dumps({'err_msg': 'City not found'}), status=404, headers=headers)
class City(Resource, Config):
@ -71,9 +73,9 @@ class City(Resource, Config):
'user': {'id': saved_city.user.id, 'name': saved_city.user.name, 'email': saved_city.user.email,
'role': saved_city.user.role.value}
}, default=str), status=201)
return Response(response=json.dumps(saved_city), status=200)
return Response(response=json.dumps(saved_city), status=200, headers=headers)
else:
return Response(response=json.dumps({'err_msg': 'Unknown city file type'}), status=400)
return Response(response=json.dumps({'err_msg': 'Unknown city file type'}), status=400, headers=headers)
except Exception as err:
logger.error(err)
return Response(response=json.dumps({'err_msg': 'Sorry an error occurred while creating city'}), status=400)
return Response(response=json.dumps({'err_msg': 'Sorry an error occurred while creating city'}), status=400, headers=headers)

20
hub_api/config.py
View File

@ -6,15 +6,31 @@ Copyright © 2023 Project Peter Yefi peteryefi@gmail.com
from hub.exports.db_factory import DBFactory as CityExportFactory
from hub.imports.db_factory import DBFactory
import os
from hub.imports.user_factory import UserFactory
from hub.exports.user_factory import UserFactory as ExUserFactory
class Config:
def __init__(self):
self.export_db_factory = CityExportFactory(db_name='hub_prod', app_env='PROD',
db_name = None
app_env = None
if os.getenv("FLASK_ENV") == 'production':
db_name = 'hub_prod'
app_env = 'PROD'
elif os.getenv("FLASK_ENV") == 'testing':
db_name = 'test_db'
app_env = 'TEST'
self.export_db_factory = CityExportFactory(db_name=db_name, app_env=app_env,
dotenv_path="{}/.env".format(os.path.expanduser('~')))
self.import_db_factory = DBFactory(db_name='hub_prod', app_env='PROD',
self.import_db_factory = DBFactory(db_name=db_name, app_env=app_env,
dotenv_path="{}/.env".format(os.path.expanduser('~')))
self.user_factory = UserFactory(db_name=db_name, app_env=app_env,
dotenv_path="{}/.env".format(os.path.expanduser('~')))
self.ex_user_factory = ExUserFactory(db_name=db_name, app_env=app_env,
dotenv_path="{}/.env".format(os.path.expanduser('~')))
def get_city(self, city_id):
return self.export_db_factory.get_city(city_id)

36
hub_api/user.py
View File

@ -6,18 +6,17 @@ Copyright © 2023 Project Author Peter Yefi peteryefi@gmail.com
import json
from flask import Response, request
from flask_restful import Resource
from hub.imports.user_factory import UserFactory
from hub.exports.user_factory import UserFactory as ExUserFactory
import os
from hub.hub_logger import logger
from hub_api.helpers.auth import generate_auth_token, role_required
from hub.persistence.models import UserRoles
from hub_api.config import Config
headers = {'Content-Type': 'application/json'}
class User(Resource):
class User(Resource, Config):
def __init__(self):
self.user_factory = UserFactory(db_name='hub_prod', app_env='PROD',
dotenv_path="{}/.env".format(os.path.expanduser('~')))
super().__init__()
@role_required([UserRoles.Admin.value])
def post(self):
@ -26,12 +25,14 @@ class User(Resource):
user = self.user_factory.create_user(name=payload["name"], email=payload["email"], password=payload["password"],
role=payload["role"])
if type(user) is dict:
return Response(response=json.dumps(user), status=400)
return Response(response=json.dumps(user), status=400, headers=headers)
return Response(response=json.dumps({'user': {'id': user.id, 'name': user.name, 'email': user.email,
'password': user.password, 'role': user.role.value}}), status=201)
'password': user.password, 'role': user.role.value}}), status=201,
headers=headers)
except Exception as err:
logger.error(err)
return Response(response=json.dumps({'err_msg': 'Sorry an error occurred while creating user'}), status=400)
return Response(response=json.dumps({'err_msg': 'Sorry an error occurred while creating user'}), status=400,
headers=headers)
@role_required([UserRoles.Admin.value])
def put(self):
@ -40,25 +41,24 @@ class User(Resource):
res = self.user_factory.update_user(user_id=payload['id'], name=payload['name'], password=payload['password'],
role=payload['role'], email=payload['email'])
if res:
return Response(response=json.dumps(res), status=400)
return Response(response=json.dumps({'success': 'user updated successfully'}), status=200)
return Response(response=json.dumps(res), status=400, headers=headers)
return Response(response=json.dumps({'success': 'user updated successfully'}), status=200, headers=headers)
except Exception as err:
logger.error(err)
return Response(response=json.dumps({'err_msg': 'Sorry, an error occurred while updating user'}),
status=400)
status=400, headers=headers)
class UserLogin(Resource):
class UserLogin(Resource, Config):
def __init__(self):
self.user_factory = ExUserFactory(db_name='hub_prod', app_env='PROD',
dotenv_path="{}/.env".format(os.path.expanduser('~')))
super().__init__()
def post(self):
try:
payload = request.get_json()
user = self.user_factory.login_user(email=payload["email"], password=payload["password"])
user = self.ex_user_factory.login_user(email=payload["email"], password=payload["password"])
if type(user) is dict:
return Response(response=json.dumps(user), status=400)
return Response(response=json.dumps(user), status=400, headers=headers)
user = user[0]
user_dict = {
'user': {
@ -70,7 +70,7 @@ class UserLogin(Resource):
}
}
user_dict['token'] = generate_auth_token(user_dict)
return Response(response=json.dumps(user_dict), status=200)
return Response(response=json.dumps(user_dict), status=200, headers=headers)
except Exception as err:
logger.error(err)
return Response(response=json.dumps({'err_msg': 'An error occurred while authenticating user'}), status=400)

0
tests/__init__.py Normal file
View File

58
tests/base_test.py Normal file
View File

@ -0,0 +1,58 @@
import os
from unittest import TestCase
from bootstrap import app
from hub.persistence.base_repo import BaseRepo
from sqlalchemy import create_engine
from hub.persistence.models import City
from hub.persistence.models import User
from sqlalchemy.exc import ProgrammingError
# function to ensure tests run in order shown in fle
def arrange():
order = {}
def ordered(f):
order[f.__name__] = len(order)
return f
def compare(a, b):
return [1, -1][order[a] < order[b]]
return ordered, compare
class BaseTest(TestCase):
"""
Tests for payment resource
"""
@classmethod
def setUpClass(cls):
os.environ['FLASK_ENV'] = 'testing'
cls.app = app
cls.client = cls.app.test_client()
# Create test database
repo = BaseRepo(db_name='test_db', app_env='TEST', dotenv_path="{}/.env".format(os.path.expanduser('~')))
eng = create_engine(f'postgresql://{repo.config.get_db_user()}@/{repo.config.get_db_user()}')
try:
# delete test database if it exists
conn = eng.connect()
conn.execute('commit')
conn.execute('DROP DATABASE test_db')
conn.close()
except ProgrammingError as err:
print(f'Database does not exist. Nothing to delete')
cnn = eng.connect()
cnn.execute('commit')
cnn.execute("CREATE DATABASE test_db")
cnn.close()
User.__table__.create(bind=repo.engine, checkfirst=True)
City.__table__.create(bind=repo.engine, checkfirst=True)
@classmethod
def tearDownClass(cls) -> None:
pass

124
tests/test_user.py Normal file
View File

@ -0,0 +1,124 @@
import json
from .base_test import BaseTest, arrange
import unittest
from hub_api.helpers.auth import generate_auth_token
ordered, compare = arrange()
unittest.defaultTestLoader.sortTestMethodsUsing = compare
class UserTest(BaseTest):
"""
Tests for User API endpoints
"""
@classmethod
def setUpClass(cls):
# Call setUp from parent
super().setUpClass()
cls.user_dict = {"user": {
"name": "Test User",
"email": "testuser@gmail.com",
"password": "TestUser@12345",
"role": "Admin",
}}
cls.token = generate_auth_token(cls.user_dict)
@ordered
def test_create_user_by_non_admin(self):
# When
res = self.client.post('/v1.4/user', data=json.dumps(self.user_dict['user']))
# Then
self.assertEqual('Invalid payload', res.json['messages'])
self.assertEqual(400, res.status_code)
@ordered
def test_create_user_by_admin(self):
# When
res = self.client.post('/v1.4/user',
headers={
'Content-Type': 'application/json', 'Authorization': 'Bearer ' + str(self.token)
},
data=json.dumps(self.user_dict['user']))
# Then
user = res.json
self.assertEqual(201, res.status_code)
self.assertEqual(type(user['user']), dict)
self.assertEqual(user['user']['email'], self.user_dict['user']['email'])
self.assertEqual(user['user']['role'], self.user_dict['user']['role'])
@ordered
def test_create_user_with_existing_email(self):
# When
res = self.client.post('/v1.4/user',
headers={
'Content-Type': 'application/json', 'Authorization': 'Bearer ' + str(self.token)
},
data=json.dumps(self.user_dict['user']))
# Then
self.assertEqual('user with testuser@gmail.com email already exists', res.json['message'])
self.assertEqual(400, res.status_code)
@ordered
def test_create_user_with_weak_password(self):
# When
self.user_dict['user']['password'] = '1234'
self.user_dict['user']['email'] = 'new@gmail.com'
res = self.client.post('/v1.4/user',
headers={
'Content-Type': 'application/json', 'Authorization': 'Bearer ' + str(self.token)
},
data=json.dumps(self.user_dict['user']))
# Then
self.assertEqual('Sorry an error occurred while creating user', res.json['err_msg'])
self.assertEqual(400, res.status_code)
@ordered
def test_login_user_with_wrong_credential(self):
# When
res = self.client.post('/v1.4/user/login',
headers={
'Content-Type': 'application/json'
},
data=json.dumps({'email': 'wrong@gmail.com', 'password': 'wrong'}))
# Then
message = res.json
self.assertEqual('user not found', message['message'])
self.assertEqual(400, res.status_code)
@ordered
def test_login_user_with_correct_credential(self):
# When
self.user_dict['user']['password'] = 'TestUser@12345'
self.user_dict['user']['email'] = 'testuser@gmail.com'
login_data = {
"email": self.user_dict['user']['email'],
"password": self.user_dict['user']['password']
}
res = self.client.post('/v1.4/user/login', headers={'Content-Type': 'application/json'},
data=json.dumps(login_data))
# Then
user = res.json
self.assertEqual(user['user']['email'], self.user_dict['user']['email'])
self.assertEqual(user['user']['role'], self.user_dict['user']['role'])
self.assertIsNotNone(user['token'])
self.assertEqual(200, res.status_code)
@classmethod
def tearDownClass(cls) -> None:
# Call tearDown from parent
super().tearDownClass()
def suite():
test_suite = unittest.TestSuite()
test_suite.addTest(unittest.TestLoader(
).loadTestsFromTestCase(UserTest))
if __name__ == '__main__':
unittest.TextTestRunner(verbosity=2).run(suite())