Added token and user tests.

portal
jprochazka 2024-07-18 16:22:56 -04:00
rodzic 182b689490
commit 3b5c546db5
7 zmienionych plików z 400 dodań i 15 usunięć

Wyświetl plik

@ -32,7 +32,10 @@ def post_user():
db=get_db()
cursor=db.cursor()
cursor.execute(
"INSERT INTO users (name, email, password, administrator) VALUES (%s, %s, %s, %s)",
"INSERT INTO users (name, email, password, administrator) VALUES (?, ?, ?, ?)",
#"INSERT INTO users (name, email, password, administrator) VALUES (%s, %s, %s, %s)",
(payload['name'], payload['email'], payload['password'], payload['administrator'])
)
db.commit()
@ -48,11 +51,17 @@ def delete_user(email):
try:
db=get_db()
cursor=db.cursor()
cursor.execute("SELECT COUNT(*) FROM users WHERE email = %s", (email,))
cursor.execute("SELECT COUNT(*) FROM users WHERE email = ?", (email,))
#cursor.execute("SELECT COUNT(*) FROM users WHERE email = %s", (email,))
if cursor.fetchone()[0] == 0:
return "Not Found", 404
else:
cursor.execute("DELETE FROM users WHERE email = %s", (email,))
cursor.execute("DELETE FROM users WHERE email = ?", (email,))
#cursor.execute("DELETE FROM users WHERE email = %s", (email,))
db.commit()
except Exception as ex:
logging.error(f"Error encountered while trying to delete user related to email {email}", exc_info=ex)
@ -68,7 +77,10 @@ def get_user(email):
try:
db=get_db()
cursor=db.cursor()
cursor.execute("SELECT * FROM users WHERE email = %s", (email,))
cursor.execute("SELECT * FROM users WHERE email = ?", (email,))
#cursor.execute("SELECT * FROM users WHERE email = %s", (email,))
columns=[x[0] for x in cursor.description]
results = cursor.fetchall()
for result in results:
@ -93,12 +105,18 @@ def put_user(email):
try:
db=get_db()
cursor=db.cursor()
cursor.execute("SELECT COUNT(*) FROM users WHERE email = %s", (email,))
cursor.execute("SELECT COUNT(*) FROM users WHERE email = ?", (email,))
#cursor.execute("SELECT COUNT(*) FROM users WHERE email = %s", (email,))
if cursor.fetchone()[0] == 0:
return "Not Found", 404
else:
cursor.execute(
"UPDATE users SET name = %s, password = %s, administrator = %s WHERE email = %s",
"UPDATE users SET name = ?, password = ?, administrator = ? WHERE email = ?",
#"UPDATE users SET name = %s, password = %s, administrator = %s WHERE email = %s",
(payload['name'], payload['password'], payload['administrator'], email)
)
db.commit()
@ -112,8 +130,8 @@ def put_user(email):
@jwt_required()
def get_users():
offset = request.args.get('offset', default=0, type=int)
limit = request.args.get('limit', default=100, type=int)
if offset < 0 or limit < 1 or limit > 1000:
limit = request.args.get('limit', default=50, type=int)
if offset < 0 or limit < 1 or limit > 100:
abort(400, description="Bad Request")
users=[]
@ -121,7 +139,10 @@ def get_users():
try:
db=get_db()
cursor=db.cursor()
cursor.execute("SELECT * FROM users ORDER BY name LIMIT %s, %s", (offset, limit))
cursor.execute("SELECT * FROM users ORDER BY name LIMIT ?, ?", (offset, limit))
#cursor.execute("SELECT * FROM users ORDER BY name LIMIT %s, %s", (offset, limit))
columns=[x[0] for x in cursor.description]
result=cursor.fetchall()
for result in result:

Wyświetl plik

@ -76,7 +76,6 @@ CREATE TABLE `users` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`name` varchar(100) NOT NULL,
`email` varchar(75) NOT NULL,
`login` varchar(25) NOT NULL,
`password` varchar(255) NOT NULL,
`administrator` bit DEFAULT 0,
PRIMARY KEY (`id`)

Wyświetl plik

@ -76,7 +76,6 @@ CREATE TABLE users (
`id` int NOT NULL GENERATED ALWAYS AS IDENTITY,
`name` varchar(100) NOT NULL,
`email` varchar(75) NOT NULL,
`login` varchar(25) NOT NULL,
`password` varchar(255) NOT NULL,
`administrator` bit DEFAULT 0,
PRIMARY KEY (id)

Wyświetl plik

@ -69,7 +69,6 @@ CREATE TABLE users (
`id` INTEGER PRIMARY KEY AUTOINCREMENT,
`name` TEXT NOT NULL,
`email` TEXT NOT NULL,
`login` TEXT NOT NULL,
`password` TEXT,
`administrator` INTEGER DEFAULT 0
);

Wyświetl plik

@ -6,6 +6,13 @@ VALUES
('icao04', '2024-07-17 04:40:44', '2024-06-17 04:44:04'),
('icao05', '2024-07-17 05:50:55', '2024-06-17 05:55:05');
INSERT INTO blog_posts (`title`, `date`, `author`, `content`)
VALUES
('Title One', '2024-07-03 13:00:01', 'User One', 'Content for blog post one.'),
('Title Two', '2024-07-04 14:30:02', 'User One', 'Content for blog post two.'),
('Title Three', '2024-07-05 15:00:03', 'User Three', 'Content for blog post three.'),
('Title Four', '2024-07-06 16:30:04', 'User Two', 'Content for blog post four.');
INSERT INTO flights (`aircraft`, `flight`, `first_seen`, `last_seen`)
VALUES
(1, 'FLT0001', '2024-07-17 01:10:11', '2024-06-17 01:11:01'),
@ -13,6 +20,18 @@ VALUES
(3, 'FLT0003', '2024-07-17 03:30:33', '2024-06-17 03:33:03'),
(5, 'FLT0005', '2024-07-17 04:40:44', '2024-06-17 04:44:04');
INSERT INTO links (`name`, `address`)
VALUES
('Link One', 'https://adsbportal.com/one'),
('Link Two', 'https://adsbportal.com/two'),
('Link Three', 'https://adsbportal.com/four');
INSERT INTO notifications (`flight`)
VALUES
('FLT0011'),
('FLT0012'),
('FLT0013');
INSERT INTO positions (`flight`, `aircraft`, `time`, `message`, `squawk`, `latitude`, `longitude`, `track`, `altitude`, `verticle_rate`, `speed`)
VALUES
(1, 1, '2024-06-17 01:11:01', 204, 6523, 42.649292, -84.960896, 98, 46975, 0, 477),
@ -30,7 +49,14 @@ VALUES
(4, 5, '2024-07-17 05:50:35', 417, 1317, 41.773464, -83.749737, 91, 36825, 768, 487),
(4, 5, '2024-07-17 05:50:55', 504, 1317, 41.772903, -83.690727, 91, 37225, 1216, 484);
INSERT INTO users (`name`, `email`, `login`, `password`, `administrator`)
INSERT INTO settings (`name`, `value`)
VALUES
('User One', 'noreply@adsbportal.com', 'login_one', '$2y$0htWdxS7PxTvIwJNo2COJ7Rywgif4En0TmJbDvrjLRfWZOBX526yJUKW', 1),
('User Two', 'noreply@adsbreceiver.net', 'login_two', '$2y$ui7QK047JldTekx828J2rfSVQ7N5yo6ETQIYGoBqpfFRbNr3EvWzQzt6', 0);
('SettingOne', 'ValueOne'),
('SettingTwo', 'ValueTwo'),
('SettingThree', 'ValueThree');
INSERT INTO users (`name`, `email`, `password`, `administrator`)
VALUES
('Name One', 'noreply@email-one.com', '$2y$0htWdxS7PxTvIwJNo2COJ7Rywgif4En0TmJbDvrjLRfWZOBX526yJUKW', 1),
('Name Two', 'noreply@email-two.com', '$2y$ui7QK047JldTekx828J2rfSVQ7N5yo6ETQIYGoBqpfFRbNr3EvWzQzt6', 0),
('Name Three', 'noreply@email-three.com', '$2y$7jiYNNoUa1zNu6dCLxv2mIurCG8nuDgOeUCeCPO9pkjiQ1zr8jfTzdEe', 0);

Wyświetl plik

@ -0,0 +1,24 @@
from flask_jwt_extended import create_refresh_token
# POST /token/login
def test_post_token_login_200(client):
response = client.post('/api/token/login')
content = response.get_json(silent=True)
assert response.status_code == 200
assert len(content['access_token']) > 0
assert len(content['refresh_token']) > 0
# POST /token/refresh
def test_post_token_refresh_200(client, app):
with app.app_context():
refresh_token = create_refresh_token(identity="developer")
request_headers = {
'Authorization': 'Bearer {}'.format(refresh_token),
'accept': 'application/json'
}
response = client.post('/api/token/refresh', headers=request_headers)
content = response.get_json(silent=True)
assert response.status_code == 200
assert len(content['access_token']) > 0

Wyświetl plik

@ -0,0 +1,317 @@
from flask_jwt_extended import create_access_token
# POST /user
def test_post_user_200(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': 'Bearer {}'.format(access_token),
}
request_json = {
'name': 'Name Four',
'email': 'noreply@email-four.com',
'password': '$2y$LFLCJxrho1eGVPp9p9ygc5fuK1XWzLWS6nsWVJvJNbtZOeMVkuBJzTXG',
'administrator': False
}
response = client.post('/api/user', headers=request_headers, json=request_json)
assert response.status_code == 201
def test_post_user_200_as_administrator(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': 'Bearer {}'.format(access_token),
}
request_json = {
'name': 'Name Five',
'email': 'noreply@email-five.com',
'password': '$2y$fqVywEatbLgW8p5QMiLVwyxc1fBcHw9nB7x2MEJ0QRo8QHlQccwvbW1S',
'administrator': 1
}
response = client.post('/api/user', headers=request_headers, json=request_json)
assert response.status_code == 201
def test_post_user_400_missing_name(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': 'Bearer {}'.format(access_token),
}
request_json = {
'email': 'noreply@email-six.com',
'password': '$2y$NFlFTvVQGuE4KXGj4PX5ekoZ64BAcT75xHGgyi7piJ8BC37vYCnMsKHS',
'administrator': 0
}
response = client.post('/api/user', headers=request_headers, json=request_json)
assert response.status_code == 400
def test_post_user_400_missing_email(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': 'Bearer {}'.format(access_token),
}
request_json = {
'name': 'Name Seven',
'password': '$2y$kLWiDakW52fH96pRHlPGFyOGL7xUD4JMcAFQvGy2HnnvBG5WzJhtCNyD',
'administrator': 0
}
response = client.post('/api/user', headers=request_headers, json=request_json)
assert response.status_code == 400
def test_post_user_400_missing_password(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': 'Bearer {}'.format(access_token),
}
request_json = {
'name': 'Name Eight',
'email': 'noreply@email-eight.com',
'administrator': 0
}
response = client.post('/api/user', headers=request_headers, json=request_json)
assert response.status_code == 400
def test_post_user_400_missing_administrator(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': 'Bearer {}'.format(access_token),
}
request_json = {
'name': 'Name Eight',
'email': 'noreply@email-eight.com',
'password': '$2y$oatXHr5ov3xW0KiPimp6UB7n76Mlb8futmO4A11285zMlPqo26I3uO2V'
}
response = client.post('/api/user', headers=request_headers, json=request_json)
assert response.status_code == 400
# DELETE /user/{email}
def test_delete_user_204(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': 'Bearer {}'.format(access_token),
}
response = client.delete('/api/user/noreply@email-three.com', headers=request_headers)
assert response.status_code == 204
def test_delete_user_404(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': 'Bearer {}'.format(access_token),
}
response = client.delete('/api/user/noreply@email-four.com', headers=request_headers)
assert response.status_code == 404
# GET /user/{email}
def test_get_user_200(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': 'Bearer {}'.format(access_token),
}
response = client.get('/api/user/noreply@email-one.com', headers=request_headers)
assert response.status_code == 200
assert response.json['id'] == 1
assert response.json['name'] == "Name One"
assert response.json['email'] == "noreply@email-one.com"
assert response.json['password'] == "$2y$0htWdxS7PxTvIwJNo2COJ7Rywgif4En0TmJbDvrjLRfWZOBX526yJUKW"
assert response.json['administrator'] == 1
def test_get_user_404(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': 'Bearer {}'.format(access_token),
}
response = client.get('/api/user/noreply@email-four.com', headers=request_headers)
assert response.status_code == 404
# PUT /user/{email}
def test_put_user_204(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': 'Bearer {}'.format(access_token),
}
request_json = {
'name': 'Name Two',
'password': '$2y$VxTtlJcPlXFj3eHzZTAvGKHXVyHWqK12TXXdUT9SHaAXKC6l7spI7sqv',
'administrator': 1
}
response = client.put('/api/user/noreply@email-two.com', headers=request_headers, json=request_json)
assert response.status_code == 204
def test_put_user_400_missing_name(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': 'Bearer {}'.format(access_token),
}
request_json = {
'password': '$2y$VxTtlJcPlXFj3eHzZTAvGKHXVyHWqK12TXXdUT9SHaAXKC6l7spI7sqv',
'administrator': 1
}
response = client.put('/api/user/noreply@email-two.com', headers=request_headers, json=request_json)
assert response.status_code == 400
def test_put_user_400_missing_password(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': 'Bearer {}'.format(access_token),
}
request_json = {
'name': 'Name Four',
'administrator': 1
}
response = client.put('/api/user/noreply@notregistered.com', headers=request_headers, json=request_json)
assert response.status_code == 400
def test_put_user_400_missing_administrator(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': 'Bearer {}'.format(access_token),
}
request_json = {
'name': 'Name Four',
'password': '$2y$52frWp0QeGJA9JZ0RZ0oQkBaDWcCWnJnUj759kmTWtUeWEjjNQMfzxo0'
}
response = client.put('/api/user/noreply@notregistered.com', headers=request_headers, json=request_json)
assert response.status_code == 400
def test_put_user_404(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': 'Bearer {}'.format(access_token),
}
request_json = {
'name': 'New Eleven',
'password': '$2y$VKHySRyCtvds21lEnqkSgvlqe4dBLSkQX1cDd32el8IgDHTnEQahoD2P',
'administrator': 1
}
response = client.put('/api/user/noreply@notregistered.com', headers=request_headers, json=request_json)
assert response.status_code == 404
# GET /users
def test_get_users_200(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': f"Bearer {access_token}",
'accept': 'application/json'
}
response = client.get('/api/users', headers=request_headers)
assert response.status_code == 200
assert response.json['offset'] == 0
assert response.json['limit'] == 50
assert response.json['count'] == 3
assert response.json['users'][0]['id'] == 1
assert response.json['users'][0]['name'] == "Name One"
assert response.json['users'][0]['email'] == "noreply@email-one.com"
assert response.json['users'][0]['password'] == "$2y$0htWdxS7PxTvIwJNo2COJ7Rywgif4En0TmJbDvrjLRfWZOBX526yJUKW"
assert response.json['users'][0]['administrator'] == 1
assert response.json['users'][1]['id'] == 3
assert response.json['users'][1]['name'] == "Name Three"
assert response.json['users'][1]['email'] == "noreply@email-three.com"
assert response.json['users'][1]['password'] == "$2y$7jiYNNoUa1zNu6dCLxv2mIurCG8nuDgOeUCeCPO9pkjiQ1zr8jfTzdEe"
assert response.json['users'][1]['administrator'] == 0
assert response.json['users'][2]['id'] == 2
assert response.json['users'][2]['name'] == "Name Two"
assert response.json['users'][2]['email'] == "noreply@email-two.com"
assert response.json['users'][2]['password'] == "$2y$ui7QK047JldTekx828J2rfSVQ7N5yo6ETQIYGoBqpfFRbNr3EvWzQzt6"
assert response.json['users'][2]['administrator'] == 0
def test_get_users_200_offset(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': f"Bearer {access_token}",
'accept': 'application/json'
}
response = client.get('/api/users?offset=2', headers=request_headers)
assert response.status_code == 200
assert response.json['offset'] == 2
assert response.json['limit'] == 50
assert response.json['count'] == 1
assert response.json['users'][0]['id'] == 2
assert response.json['users'][0]['name'] == "Name Two"
assert response.json['users'][0]['email'] == "noreply@email-two.com"
assert response.json['users'][0]['password'] == "$2y$ui7QK047JldTekx828J2rfSVQ7N5yo6ETQIYGoBqpfFRbNr3EvWzQzt6"
assert response.json['users'][0]['administrator'] == 0
def test_get_users_200_limit(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': f"Bearer {access_token}",
'accept': 'application/json'
}
response = client.get('/api/users?limit=1', headers=request_headers)
assert response.status_code == 200
assert response.json['offset'] == 0
assert response.json['limit'] == 1
assert response.json['count'] == 1
assert response.json['users'][0]['id'] == 1
assert response.json['users'][0]['name'] == "Name One"
assert response.json['users'][0]['email'] == "noreply@email-one.com"
assert response.json['users'][0]['password'] == "$2y$0htWdxS7PxTvIwJNo2COJ7Rywgif4En0TmJbDvrjLRfWZOBX526yJUKW"
assert response.json['users'][0]['administrator'] == 1
def test_get_users_200_offset_and_limit(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': f"Bearer {access_token}",
'accept': 'application/json'
}
response = client.get('/api/users?offset=1&limit=1', headers=request_headers)
assert response.status_code == 200
assert response.json['offset'] == 1
assert response.json['limit'] == 1
assert response.json['count'] == 1
assert response.json['users'][0]['id'] == 3
assert response.json['users'][0]['name'] == "Name Three"
assert response.json['users'][0]['email'] == "noreply@email-three.com"
assert response.json['users'][0]['password'] == "$2y$7jiYNNoUa1zNu6dCLxv2mIurCG8nuDgOeUCeCPO9pkjiQ1zr8jfTzdEe"
assert response.json['users'][0]['administrator'] == 0
def test_get_users_400_offset_less_than_0(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': f"Bearer {access_token}",
'accept': 'application/json'
}
response = client.get('/api/users?offset=-1', headers=request_headers)
assert response.status_code == 400
def test_get_users_400_limit_less_than_0(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': f"Bearer {access_token}",
'accept': 'application/json'
}
response = client.get('/api/users?limit=-1', headers=request_headers)
assert response.status_code == 400
def test_get_users_400_limit_greater_than_100(client, app):
with app.app_context():
access_token = create_access_token(identity="developer")
request_headers = {
'Authorization': f"Bearer {access_token}",
'accept': 'application/json'
}
response = client.get('/api/users?limit=101', headers=request_headers)
assert response.status_code == 400