kopia lustrzana https://github.com/jprochazka/adsb-receiver
Added additional test.
Joseph Prochazka
rodzic
3b5c546db5
commit
24eeb4498b
|
|
@ -3,4 +3,5 @@
|
|||
__pycache__
|
||||
instance
|
||||
.coverage
|
||||
htmlcov
|
||||
htmlcov
|
||||
.pytest_cache
|
||||
|
|
@ -31,7 +31,10 @@ def post_blog_post():
|
|||
db=get_db()
|
||||
cursor=db.cursor()
|
||||
cursor.execute(
|
||||
"INSERT INTO blog_posts (date, title, author, content) VALUES (%s, %s, %s, %s)",
|
||||
|
||||
"INSERT INTO blog_posts (date, title, author, content) VALUES (?, ?, ?, ?)",
|
||||
#"INSERT INTO blog_posts (date, title, author, content) VALUES (%s, %s, %s, %s)",
|
||||
|
||||
(datetime.now(), payload['title'], payload['author'], payload['content'])
|
||||
)
|
||||
db.commit()
|
||||
|
|
@ -47,11 +50,17 @@ def delete_blog_post(blog_post_id):
|
|||
try:
|
||||
db=get_db()
|
||||
cursor=db.cursor()
|
||||
cursor.execute("SELECT COUNT(*) FROM blog_posts WHERE id = %s", (blog_post_id,))
|
||||
|
||||
cursor.execute("SELECT COUNT(*) FROM blog_posts WHERE id = ?", (blog_post_id,))
|
||||
#cursor.execute("SELECT COUNT(*) FROM blog_posts WHERE id = %s", (blog_post_id,))
|
||||
|
||||
if cursor.fetchone()[0] == 0:
|
||||
return "Not Found", 404
|
||||
else:
|
||||
cursor.execute("DELETE FROM blog_posts WHERE id = %s", (blog_post_id,))
|
||||
|
||||
cursor.execute("DELETE FROM blog_posts WHERE id = ?", (blog_post_id,))
|
||||
#cursor.execute("DELETE FROM blog_posts WHERE id = %s", (blog_post_id,))
|
||||
|
||||
db.commit()
|
||||
except Exception as ex:
|
||||
logging.error(f"Error encountered while trying to delete blog post id {blog_post_id}", exc_info=ex)
|
||||
|
|
@ -66,7 +75,10 @@ def get_blog_post(blog_post_id):
|
|||
try:
|
||||
db=get_db()
|
||||
cursor=db.cursor()
|
||||
cursor.execute("SELECT * FROM blog_posts WHERE id = %s", (blog_post_id,))
|
||||
|
||||
cursor.execute("SELECT * FROM blog_posts WHERE id = ?", (blog_post_id,))
|
||||
#cursor.execute("SELECT * FROM blog_posts WHERE id = %s", (blog_post_id,))
|
||||
|
||||
columns=[x[0] for x in cursor.description]
|
||||
results = cursor.fetchall()
|
||||
for result in results:
|
||||
|
|
@ -93,12 +105,18 @@ def put_blog_post(blog_post_id):
|
|||
try:
|
||||
db=get_db()
|
||||
cursor=db.cursor()
|
||||
cursor.execute("SELECT COUNT(*) FROM blog_posts WHERE id = %s", (blog_post_id,))
|
||||
|
||||
cursor.execute("SELECT COUNT(*) FROM blog_posts WHERE id = ?", (blog_post_id,))
|
||||
#cursor.execute("SELECT COUNT(*) FROM blog_posts WHERE id = %s", (blog_post_id,))
|
||||
|
||||
if cursor.fetchone()[0] == 0:
|
||||
return "Not Found", 404
|
||||
else:
|
||||
cursor.execute(
|
||||
"UPDATE blog_posts SET date = %s, title = %s, content = %s WHERE id = %s",
|
||||
|
||||
"UPDATE blog_posts SET date = ?, title = ?, content = ? WHERE id = ?",
|
||||
#"UPDATE blog_posts SET date = %s, title = %s, content = %s WHERE id = %s",
|
||||
|
||||
(datetime.now(), payload['title'], payload['content'], blog_post_id)
|
||||
)
|
||||
db.commit()
|
||||
|
|
@ -120,7 +138,10 @@ def get_blog_posts():
|
|||
try:
|
||||
db=get_db()
|
||||
cursor=db.cursor()
|
||||
cursor.execute("SELECT * FROM blog_posts ORDER BY date DESC LIMIT %s, %s", (offset, limit))
|
||||
|
||||
cursor.execute("SELECT * FROM blog_posts ORDER BY date DESC LIMIT ?, ?", (offset, limit))
|
||||
#cursor.execute("SELECT * FROM blog_posts ORDER BY date DESC LIMIT %s, %s", (offset, limit))
|
||||
|
||||
columns=[x[0] for x in cursor.description]
|
||||
result=cursor.fetchall()
|
||||
for result in result:
|
||||
|
|
|
|||
|
|
@ -29,7 +29,10 @@ def post_link():
|
|||
db=get_db()
|
||||
cursor=db.cursor()
|
||||
cursor.execute(
|
||||
"INSERT INTO links (name, address) VALUES (%s, %s)",
|
||||
|
||||
"INSERT INTO links (name, address) VALUES (?, ?)",
|
||||
#"INSERT INTO links (name, address) VALUES (%s, %s)",
|
||||
|
||||
(payload['name'], payload['address'])
|
||||
)
|
||||
db.commit()
|
||||
|
|
@ -45,11 +48,17 @@ def delete_link(link_id):
|
|||
try:
|
||||
db=get_db()
|
||||
cursor=db.cursor()
|
||||
cursor.execute("SELECT COUNT(*) FROM links WHERE id = %s", (link_id,))
|
||||
|
||||
cursor.execute("SELECT COUNT(*) FROM links WHERE id = ?", (link_id,))
|
||||
#cursor.execute("SELECT COUNT(*) FROM links WHERE id = %s", (link_id,))
|
||||
|
||||
if cursor.fetchone()[0] == 0:
|
||||
return "Not Found", 404
|
||||
else:
|
||||
cursor.execute("DELETE FROM links WHERE id = %s", (link_id,))
|
||||
|
||||
cursor.execute("DELETE FROM links WHERE id = ?", (link_id,))
|
||||
#cursor.execute("DELETE FROM links WHERE id = %s", (link_id,))
|
||||
|
||||
db.commit()
|
||||
except Exception as ex:
|
||||
logging.error(f"Error encountered while trying to delete link id {link_id}", exc_info=ex)
|
||||
|
|
@ -65,7 +74,10 @@ def get_link(link_id):
|
|||
try:
|
||||
db=get_db()
|
||||
cursor=db.cursor()
|
||||
cursor.execute("SELECT * FROM links WHERE id = %s", (link_id,))
|
||||
|
||||
cursor.execute("SELECT * FROM links WHERE id = ?", (link_id,))
|
||||
#cursor.execute("SELECT * FROM links WHERE id = %s", (link_id,))
|
||||
|
||||
columns=[x[0] for x in cursor.description]
|
||||
results = cursor.fetchall()
|
||||
for result in results:
|
||||
|
|
@ -82,24 +94,27 @@ def get_link(link_id):
|
|||
@links.route('/api/link/<int:id>', methods=['PUT'])
|
||||
@jwt_required()
|
||||
def put_link(id):
|
||||
payload = request.json
|
||||
payload_schema = UpdateLinkRequestSchema
|
||||
|
||||
try:
|
||||
payload_object = payload_schema.load(payload)
|
||||
payload = UpdateLinkRequestSchema().load(request.json)
|
||||
except ValidationError as err:
|
||||
return jsonify(err.messages), 400
|
||||
|
||||
try:
|
||||
db=get_db()
|
||||
cursor=db.cursor()
|
||||
cursor.execute("SELECT COUNT(*) FROM links WHERE id = %s", (id))
|
||||
|
||||
cursor.execute("SELECT COUNT(*) FROM links WHERE id = ?", (id,))
|
||||
#cursor.execute("SELECT COUNT(*) FROM links WHERE id = %s", (id,))
|
||||
|
||||
if cursor.fetchone()[0] == 0:
|
||||
abort(404, description="Not Found")
|
||||
return "Not Found", 404
|
||||
else:
|
||||
cursor.execute(
|
||||
"UPDATE links SET name = %s, address = %s WHERE id = %s",
|
||||
(payload_object['name'], payload_object['address'], id)
|
||||
|
||||
"UPDATE links SET name = ?, address = ? WHERE id = ?",
|
||||
#"UPDATE links SET name = %s, address = %s WHERE id = %s",
|
||||
|
||||
(payload['name'], payload['address'], id)
|
||||
)
|
||||
db.commit()
|
||||
except Exception as ex:
|
||||
|
|
@ -120,7 +135,10 @@ def get_links():
|
|||
try:
|
||||
db=get_db()
|
||||
cursor=db.cursor()
|
||||
cursor.execute("SELECT * FROM links ORDER BY name LIMIT %s, %s", (offset, limit))
|
||||
|
||||
cursor.execute("SELECT * FROM links ORDER BY name LIMIT ?, ?", (offset, limit))
|
||||
#cursor.execute("SELECT * FROM links ORDER BY name LIMIT %s, %s", (offset, limit))
|
||||
|
||||
columns=[x[0] for x in cursor.description]
|
||||
result=cursor.fetchall()
|
||||
for result in result:
|
||||
|
|
|
|||
|
|
@ -13,11 +13,17 @@ def delete_notification(flight):
|
|||
try:
|
||||
db=get_db()
|
||||
cursor=db.cursor()
|
||||
cursor.execute("SELECT COUNT(*) FROM notifications WHERE flight = %s", (flight,))
|
||||
|
||||
cursor.execute("SELECT COUNT(*) FROM notifications WHERE flight = ?", (flight,))
|
||||
#cursor.execute("SELECT COUNT(*) FROM notifications WHERE flight = %s", (flight,))
|
||||
|
||||
if cursor.fetchone()[0] == 0:
|
||||
return "Not Found", 404
|
||||
else:
|
||||
cursor.execute("DELETE FROM notifications WHERE flight = %s", (flight,))
|
||||
|
||||
cursor.execute("DELETE FROM notifications WHERE flight = ?", (flight,))
|
||||
#cursor.execute("DELETE FROM notifications WHERE flight = %s", (flight,))
|
||||
|
||||
db.commit()
|
||||
except Exception as ex:
|
||||
logging.error(f"Error encountered while trying to delete blog post id {flight}", exc_info=ex)
|
||||
|
|
@ -31,12 +37,18 @@ def post_notification(flight):
|
|||
try:
|
||||
db=get_db()
|
||||
cursor=db.cursor()
|
||||
cursor.execute("SELECT COUNT(*) FROM notifications WHERE flight = %s", (flight,))
|
||||
|
||||
cursor.execute("SELECT COUNT(*) FROM notifications WHERE flight = ?", (flight,))
|
||||
#cursor.execute("SELECT COUNT(*) FROM notifications WHERE flight = %s", (flight,))
|
||||
|
||||
if cursor.fetchone()[0] > 0:
|
||||
return "Bad Request", 400
|
||||
else:
|
||||
cursor.execute(
|
||||
"INSERT INTO notifications (flight) VALUES (%s)",
|
||||
|
||||
"INSERT INTO notifications (flight) VALUES (?)",
|
||||
#"INSERT INTO notifications (flight) VALUES (%s)",
|
||||
|
||||
(flight,)
|
||||
)
|
||||
db.commit()
|
||||
|
|
@ -47,7 +59,6 @@ def post_notification(flight):
|
|||
return "Created", 201
|
||||
|
||||
@notifications.route('/api/notifications', methods=['GET'])
|
||||
@jwt_required()
|
||||
def get_notifications():
|
||||
offset = request.args.get('offset', default=0, type=int)
|
||||
limit = request.args.get('limit', default=100, type=int)
|
||||
|
|
@ -59,7 +70,10 @@ def get_notifications():
|
|||
try:
|
||||
db=get_db()
|
||||
cursor=db.cursor()
|
||||
cursor.execute("SELECT * FROM notifications ORDER BY flight LIMIT %s, %s", (offset, limit))
|
||||
|
||||
cursor.execute("SELECT * FROM notifications ORDER BY flight LIMIT ?, ?", (offset, limit))
|
||||
#cursor.execute("SELECT * FROM notifications ORDER BY flight LIMIT %s, %s", (offset, limit))
|
||||
|
||||
columns=[x[0] for x in cursor.description]
|
||||
result=cursor.fetchall()
|
||||
for result in result:
|
||||
|
|
|
|||
|
|
@ -24,12 +24,18 @@ def put_setting():
|
|||
try:
|
||||
db=get_db()
|
||||
cursor=db.cursor()
|
||||
cursor.execute("SELECT COUNT(*) FROM settings WHERE name = %s", (payload['name'],))
|
||||
|
||||
cursor.execute("SELECT COUNT(*) FROM settings WHERE name = ?", (payload['name'],))
|
||||
#cursor.execute("SELECT COUNT(*) FROM settings WHERE name = %s", (payload['name'],))
|
||||
|
||||
if cursor.fetchone()[0] == 0:
|
||||
abort(404, description="Not Found")
|
||||
return "Not Found", 404
|
||||
else:
|
||||
cursor.execute(
|
||||
"UPDATE settings SET value = %s WHERE name = %s",
|
||||
|
||||
"UPDATE settings SET value = ? WHERE name = ?",
|
||||
#"UPDATE settings SET value = %s WHERE name = %s",
|
||||
|
||||
(payload['value'], payload['name'])
|
||||
)
|
||||
db.commit()
|
||||
|
|
@ -40,14 +46,16 @@ def put_setting():
|
|||
return "No Content", 204
|
||||
|
||||
@settings.route('/api/setting/<string:name>', methods=['GET'])
|
||||
@jwt_required()
|
||||
def get_setting(name):
|
||||
data=[]
|
||||
|
||||
try:
|
||||
db=get_db()
|
||||
cursor=db.cursor()
|
||||
cursor.execute("SELECT * FROM settings WHERE name = %s", (name,))
|
||||
|
||||
cursor.execute("SELECT * FROM settings WHERE name = ?", (name,))
|
||||
#cursor.execute("SELECT * FROM settings WHERE name = %s", (name,))
|
||||
|
||||
columns=[x[0] for x in cursor.description]
|
||||
results = cursor.fetchall()
|
||||
for result in results:
|
||||
|
|
@ -78,4 +86,5 @@ def get_settings():
|
|||
logging.error(f"Error encountered while trying to get settings", exc_info=ex)
|
||||
abort(500, description="Internal Server Error")
|
||||
|
||||
return jsonify(settings), 200
|
||||
return jsonify(settings), 200
|
||||
|
||||
|
|
|
|||
|
|
@ -677,8 +677,6 @@ paths:
|
|||
get:
|
||||
tags:
|
||||
- notifications
|
||||
security:
|
||||
- bearerAuth: []
|
||||
summary: Get data related to multiple notifications
|
||||
description: Returns data for all currently available notifications
|
||||
operationId: getNotifications
|
||||
|
|
@ -746,8 +744,6 @@ paths:
|
|||
get:
|
||||
tags:
|
||||
- settings
|
||||
security:
|
||||
- bearerAuth: []
|
||||
summary: Get data related to the specified setting name
|
||||
description: Returns all data related to the setting assigned the supplied name
|
||||
operationId: getSetting
|
||||
|
|
@ -1072,7 +1068,7 @@ components:
|
|||
type: integer
|
||||
count:
|
||||
type: integer
|
||||
blogPosts:
|
||||
blog_posts:
|
||||
type: array
|
||||
items:
|
||||
$ref: '#/components/schemas/BlogPost'
|
||||
|
|
|
|||
Plik binarny nie jest wyświetlany.
|
|
@ -24,7 +24,7 @@ INSERT INTO links (`name`, `address`)
|
|||
VALUES
|
||||
('Link One', 'https://adsbportal.com/one'),
|
||||
('Link Two', 'https://adsbportal.com/two'),
|
||||
('Link Three', 'https://adsbportal.com/four');
|
||||
('Link Three', 'https://adsbportal.com/three');
|
||||
|
||||
INSERT INTO notifications (`flight`)
|
||||
VALUES
|
||||
|
|
@ -51,9 +51,9 @@ VALUES
|
|||
|
||||
INSERT INTO settings (`name`, `value`)
|
||||
VALUES
|
||||
('SettingOne', 'ValueOne'),
|
||||
('SettingTwo', 'ValueTwo'),
|
||||
('SettingThree', 'ValueThree');
|
||||
('setting_one', 'Value One'),
|
||||
('setting_two', 'Value Two'),
|
||||
('setting_three', 'Value Three');
|
||||
|
||||
INSERT INTO users (`name`, `email`, `password`, `administrator`)
|
||||
VALUES
|
||||
|
|
|
|||
|
|
@ -0,0 +1,226 @@
|
|||
from flask_jwt_extended import create_access_token
|
||||
|
||||
# POST /blog/post
|
||||
|
||||
def test_post_blog_post_200(client, app):
|
||||
with app.app_context():
|
||||
access_token = create_access_token(identity="developer")
|
||||
request_headers = {
|
||||
'Authorization': 'Bearer {}'.format(access_token),
|
||||
}
|
||||
request_json = {
|
||||
'title': 'Title Five',
|
||||
'author': 'User Two',
|
||||
'content': 'Content for blog post five.'
|
||||
}
|
||||
response = client.post('/api/blog/post', headers=request_headers, json=request_json)
|
||||
assert response.status_code == 201
|
||||
|
||||
def test_post_blog_post_400_missing_title(client, app):
|
||||
with app.app_context():
|
||||
access_token = create_access_token(identity="developer")
|
||||
request_headers = {
|
||||
'Authorization': 'Bearer {}'.format(access_token),
|
||||
}
|
||||
request_json = {
|
||||
'author': 'User Two',
|
||||
'content': 'Content for blog post five.'
|
||||
}
|
||||
response = client.post('/api/blog/post', headers=request_headers, json=request_json)
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_post_blog_post_400_missing_author(client, app):
|
||||
with app.app_context():
|
||||
access_token = create_access_token(identity="developer")
|
||||
request_headers = {
|
||||
'Authorization': 'Bearer {}'.format(access_token),
|
||||
}
|
||||
request_json = {
|
||||
'title': 'Title Five',
|
||||
'content': 'Content for blog post five.'
|
||||
}
|
||||
response = client.post('/api/blog/post', headers=request_headers, json=request_json)
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_post_blog_post_400_missing_content(client, app):
|
||||
with app.app_context():
|
||||
access_token = create_access_token(identity="developer")
|
||||
request_headers = {
|
||||
'Authorization': 'Bearer {}'.format(access_token),
|
||||
}
|
||||
request_json = {
|
||||
'title': 'Title Five',
|
||||
'author': 'User Two',
|
||||
}
|
||||
response = client.post('/api/blog/post', headers=request_headers, json=request_json)
|
||||
assert response.status_code == 400
|
||||
|
||||
# DELETE /blog/post/{id}
|
||||
|
||||
def test_delete_blog_post_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/blog/post/2', headers=request_headers)
|
||||
assert response.status_code == 204
|
||||
|
||||
def test_delete_blog_post_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/blog/post/5', headers=request_headers)
|
||||
assert response.status_code == 404
|
||||
|
||||
# GET /blog/post/{id}
|
||||
|
||||
def test_get_blog_post_200(client):
|
||||
response = client.get('/api/blog/post/3')
|
||||
assert response.status_code == 200
|
||||
assert response.json['id'] == 3
|
||||
assert response.json['title'] == "Title Three"
|
||||
assert response.json['date'] == "2024-07-05 15:00:03"
|
||||
assert response.json['author'] == "User Three"
|
||||
assert response.json['content'] == "Content for blog post three."
|
||||
|
||||
def test_get_blog_post_404(client):
|
||||
response = client.get('/api/blog/post/5')
|
||||
assert response.status_code == 404
|
||||
|
||||
# PUT /blog/post/{id}
|
||||
|
||||
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 = {
|
||||
'title': 'Updated Title One',
|
||||
'content': 'Updated content for blog post one.'
|
||||
}
|
||||
response = client.put('/api/blog/post/1', headers=request_headers, json=request_json)
|
||||
assert response.status_code == 204
|
||||
|
||||
def test_put_user_400_missing_title(client, app):
|
||||
with app.app_context():
|
||||
access_token = create_access_token(identity="developer")
|
||||
request_headers = {
|
||||
'Authorization': 'Bearer {}'.format(access_token),
|
||||
}
|
||||
request_json = {
|
||||
'content': 'Updated content for blog post one.'
|
||||
}
|
||||
response = client.put('/api/blog/post/1', headers=request_headers, json=request_json)
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_put_user_400_missing_content(client, app):
|
||||
with app.app_context():
|
||||
access_token = create_access_token(identity="developer")
|
||||
request_headers = {
|
||||
'Authorization': 'Bearer {}'.format(access_token),
|
||||
}
|
||||
request_json = {
|
||||
'title': 'Updated Title One'
|
||||
}
|
||||
response = client.put('/api/blog/post/1', 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 = {
|
||||
'title': 'Updated Title five',
|
||||
'content': 'Updated content for blog post five.'
|
||||
}
|
||||
response = client.put('/api/blog/post/5', headers=request_headers, json=request_json)
|
||||
assert response.status_code == 404
|
||||
|
||||
# GET /blog/posts
|
||||
|
||||
def test_get_blog_post_200(client):
|
||||
response = client.get('/api/blog/posts')
|
||||
assert response.status_code == 200
|
||||
assert response.json['offset'] == 0
|
||||
assert response.json['limit'] == 25
|
||||
assert response.json['count'] == 4
|
||||
assert response.json['blog_posts'][0]['id'] == 4
|
||||
assert response.json['blog_posts'][0]['title'] == "Title Four"
|
||||
assert response.json['blog_posts'][0]['date'] == "2024-07-06 16:30:04"
|
||||
assert response.json['blog_posts'][0]['author'] == "User Two"
|
||||
assert response.json['blog_posts'][0]['content'] == "Content for blog post four."
|
||||
assert response.json['blog_posts'][1]['id'] == 3
|
||||
assert response.json['blog_posts'][1]['title'] == "Title Three"
|
||||
assert response.json['blog_posts'][1]['date'] == "2024-07-05 15:00:03"
|
||||
assert response.json['blog_posts'][1]['author'] == "User Three"
|
||||
assert response.json['blog_posts'][1]['content'] == "Content for blog post three."
|
||||
assert response.json['blog_posts'][2]['id'] == 2
|
||||
assert response.json['blog_posts'][2]['title'] == "Title Two"
|
||||
assert response.json['blog_posts'][2]['date'] == "2024-07-04 14:30:02"
|
||||
assert response.json['blog_posts'][2]['author'] == "User One"
|
||||
assert response.json['blog_posts'][2]['content'] == "Content for blog post two."
|
||||
assert response.json['blog_posts'][3]['id'] == 1
|
||||
assert response.json['blog_posts'][3]['title'] == "Title One"
|
||||
assert response.json['blog_posts'][3]['date'] == "2024-07-03 13:00:01"
|
||||
assert response.json['blog_posts'][3]['author'] == "User One"
|
||||
assert response.json['blog_posts'][3]['content'] == "Content for blog post one."
|
||||
|
||||
def test_get_blog_post_200_offset(client):
|
||||
response = client.get('/api/blog/posts?offset=2')
|
||||
assert response.status_code == 200
|
||||
assert response.json['offset'] == 2
|
||||
assert response.json['limit'] == 25
|
||||
assert response.json['count'] == 2
|
||||
assert response.json['blog_posts'][0]['id'] == 2
|
||||
assert response.json['blog_posts'][0]['title'] == "Title Two"
|
||||
assert response.json['blog_posts'][0]['date'] == "2024-07-04 14:30:02"
|
||||
assert response.json['blog_posts'][0]['author'] == "User One"
|
||||
assert response.json['blog_posts'][0]['content'] == "Content for blog post two."
|
||||
assert response.json['blog_posts'][1]['id'] == 1
|
||||
assert response.json['blog_posts'][1]['title'] == "Title One"
|
||||
assert response.json['blog_posts'][1]['date'] == "2024-07-03 13:00:01"
|
||||
assert response.json['blog_posts'][1]['author'] == "User One"
|
||||
assert response.json['blog_posts'][1]['content'] == "Content for blog post one."
|
||||
|
||||
|
||||
def test_get_blog_post_200_limit(client):
|
||||
response = client.get('/api/blog/posts?limit=1')
|
||||
assert response.status_code == 200
|
||||
assert response.json['offset'] == 0
|
||||
assert response.json['limit'] == 1
|
||||
assert response.json['count'] == 1
|
||||
assert response.json['blog_posts'][0]['id'] == 4
|
||||
assert response.json['blog_posts'][0]['title'] == "Title Four"
|
||||
assert response.json['blog_posts'][0]['date'] == "2024-07-06 16:30:04"
|
||||
assert response.json['blog_posts'][0]['author'] == "User Two"
|
||||
assert response.json['blog_posts'][0]['content'] == "Content for blog post four."
|
||||
|
||||
def test_get_blog_post_200_offset_and_limit(client):
|
||||
response = client.get('/api/blog/posts?offset=1&limit=1')
|
||||
assert response.status_code == 200
|
||||
assert response.json['offset'] == 1
|
||||
assert response.json['limit'] == 1
|
||||
assert response.json['count'] == 1
|
||||
assert response.json['blog_posts'][0]['id'] == 3
|
||||
assert response.json['blog_posts'][0]['title'] == "Title Three"
|
||||
assert response.json['blog_posts'][0]['date'] == "2024-07-05 15:00:03"
|
||||
assert response.json['blog_posts'][0]['author'] == "User Three"
|
||||
assert response.json['blog_posts'][0]['content'] == "Content for blog post three."
|
||||
|
||||
def test_get_blog_post_400_offset_less_than_0(client):
|
||||
response = client.get('/api/blog/posts?offset=-1')
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_get_blog_post_400_limit_less_than_0(client):
|
||||
response = client.get('/api/blog/posts?limit=-1')
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_get_blog_post_400_limit_greater_than_100(client):
|
||||
response = client.get('/api/blog/posts?limit=101')
|
||||
assert response.status_code == 400
|
||||
|
|
@ -0,0 +1,194 @@
|
|||
from flask_jwt_extended import create_access_token
|
||||
|
||||
# POST /link
|
||||
|
||||
def test_post_link_200(client, app):
|
||||
with app.app_context():
|
||||
access_token = create_access_token(identity="developer")
|
||||
request_headers = {
|
||||
'Authorization': 'Bearer {}'.format(access_token),
|
||||
}
|
||||
request_json = {
|
||||
'name': 'Link Four',
|
||||
'address': 'https://adsbportal.com/four'
|
||||
}
|
||||
response = client.post('/api/link', headers=request_headers, json=request_json)
|
||||
assert response.status_code == 201
|
||||
|
||||
def test_post_link_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 = {
|
||||
'address': 'https://adsbportal.com/four'
|
||||
}
|
||||
response = client.post('/api/link', headers=request_headers, json=request_json)
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_post_link_400_missing_address(client, app):
|
||||
with app.app_context():
|
||||
access_token = create_access_token(identity="developer")
|
||||
request_headers = {
|
||||
'Authorization': 'Bearer {}'.format(access_token),
|
||||
}
|
||||
request_json = {
|
||||
'name': 'Link Four',
|
||||
}
|
||||
response = client.post('/api/link', headers=request_headers, json=request_json)
|
||||
assert response.status_code == 400
|
||||
|
||||
# DELETE /link/{id}
|
||||
|
||||
def test_delete_link_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/link/1', headers=request_headers)
|
||||
assert response.status_code == 204
|
||||
|
||||
def test_delete_link_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/link/4', headers=request_headers)
|
||||
assert response.status_code == 404
|
||||
|
||||
# GET /link/{id}
|
||||
|
||||
def test_get_link_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/link/2', headers=request_headers)
|
||||
assert response.status_code == 200
|
||||
assert response.json['id'] == 2
|
||||
assert response.json['address'] == "https://adsbportal.com/two"
|
||||
|
||||
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/link/4', headers=request_headers)
|
||||
assert response.status_code == 404
|
||||
|
||||
# PUT /link/{id}
|
||||
|
||||
def test_put_link_204(client, app):
|
||||
with app.app_context():
|
||||
access_token = create_access_token(identity="developer")
|
||||
request_headers = {
|
||||
'Authorization': 'Bearer {}'.format(access_token),
|
||||
}
|
||||
request_json = {
|
||||
'name': 'Link Three Updated',
|
||||
'address': 'https://adsbportal.com/three-updated'
|
||||
}
|
||||
response = client.put('/api/link/3', headers=request_headers, json=request_json)
|
||||
assert response.status_code == 204
|
||||
|
||||
def test_put_link_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 = {
|
||||
'address': 'https://adsbportal.com/three-updated'
|
||||
}
|
||||
response = client.put('/api/link/3', headers=request_headers, json=request_json)
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_put_link_400_missing_address(client, app):
|
||||
with app.app_context():
|
||||
access_token = create_access_token(identity="developer")
|
||||
request_headers = {
|
||||
'Authorization': 'Bearer {}'.format(access_token),
|
||||
}
|
||||
request_json = {
|
||||
'name': 'Link Three Updated'
|
||||
}
|
||||
response = client.put('/api/link/3', headers=request_headers, json=request_json)
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_put_link_404(client, app):
|
||||
with app.app_context():
|
||||
access_token = create_access_token(identity="developer")
|
||||
request_headers = {
|
||||
'Authorization': 'Bearer {}'.format(access_token),
|
||||
}
|
||||
request_json = {
|
||||
'name': 'Link Four Updated',
|
||||
'address': 'https://adsbportal.com/four-updated'
|
||||
}
|
||||
response = client.put('/api/link/4', headers=request_headers, json=request_json)
|
||||
assert response.status_code == 404
|
||||
|
||||
# GET /links
|
||||
|
||||
def test_get_links_200(client):
|
||||
response = client.get('/api/links')
|
||||
assert response.status_code == 200
|
||||
assert response.json['offset'] == 0
|
||||
assert response.json['limit'] == 50
|
||||
assert response.json['count'] == 3
|
||||
assert response.json['links'][0]['id'] == 1
|
||||
assert response.json['links'][0]['name'] == "Link One"
|
||||
assert response.json['links'][0]['address'] == "https://adsbportal.com/one"
|
||||
assert response.json['links'][1]['id'] == 3
|
||||
assert response.json['links'][1]['name'] == "Link Three"
|
||||
assert response.json['links'][1]['address'] == "https://adsbportal.com/three"
|
||||
assert response.json['links'][2]['id'] == 2
|
||||
assert response.json['links'][2]['name'] == "Link Two"
|
||||
assert response.json['links'][2]['address'] == "https://adsbportal.com/two"
|
||||
|
||||
def test_get_links_200_offset(client):
|
||||
response = client.get('/api/links?offset=2')
|
||||
assert response.status_code == 200
|
||||
assert response.json['offset'] == 2
|
||||
assert response.json['limit'] == 50
|
||||
assert response.json['count'] == 1
|
||||
assert response.json['links'][0]['id'] == 2
|
||||
assert response.json['links'][0]['name'] == "Link Two"
|
||||
assert response.json['links'][0]['address'] == "https://adsbportal.com/two"
|
||||
|
||||
def test_get_links_200_limit(client):
|
||||
response = client.get('/api/links?limit=1')
|
||||
assert response.status_code == 200
|
||||
assert response.json['offset'] == 0
|
||||
assert response.json['limit'] == 1
|
||||
assert response.json['count'] == 1
|
||||
assert response.json['links'][0]['id'] == 1
|
||||
assert response.json['links'][0]['name'] == "Link One"
|
||||
assert response.json['links'][0]['address'] == "https://adsbportal.com/one"
|
||||
|
||||
def test_get_links_200_offset_and_limit(client):
|
||||
response = client.get('/api/links?offset=1&limit=1')
|
||||
assert response.status_code == 200
|
||||
assert response.json['offset'] == 1
|
||||
assert response.json['limit'] == 1
|
||||
assert response.json['count'] == 1
|
||||
assert response.json['links'][0]['id'] == 3
|
||||
assert response.json['links'][0]['name'] == "Link Three"
|
||||
assert response.json['links'][0]['address'] == "https://adsbportal.com/three"
|
||||
|
||||
def test_get_links_400_offset_less_than_0(client):
|
||||
response = client.get('/api/links?offset=-1')
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_get_links_400_limit_less_than_0(client):
|
||||
response = client.get('/api/links?limit=-1')
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_get_links_400_limit_greater_than_100(client):
|
||||
response = client.get('/api/links?limit=101')
|
||||
assert response.status_code == 400
|
||||
|
|
@ -0,0 +1,86 @@
|
|||
from flask_jwt_extended import create_access_token
|
||||
|
||||
# DELETE /notification/{id}
|
||||
|
||||
def test_delete_notification_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/notification/FLT0013', headers=request_headers)
|
||||
assert response.status_code == 204
|
||||
|
||||
def test_delete_notification_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/notification/FLT0000', headers=request_headers)
|
||||
assert response.status_code == 404
|
||||
|
||||
# POST /notification
|
||||
|
||||
def test_post_notification_200(client, app):
|
||||
with app.app_context():
|
||||
access_token = create_access_token(identity="developer")
|
||||
request_headers = {
|
||||
'Authorization': 'Bearer {}'.format(access_token),
|
||||
}
|
||||
response = client.post('/api/notification/FLT0014', headers=request_headers)
|
||||
assert response.status_code == 201
|
||||
|
||||
# GET /notifications
|
||||
|
||||
def test_get_notifications_200(client):
|
||||
response = client.get('/api/notifications')
|
||||
assert response.status_code == 200
|
||||
assert response.json['offset'] == 0
|
||||
assert response.json['limit'] == 100
|
||||
assert response.json['count'] == 3
|
||||
assert response.json['notifications'][0]['id'] == 1
|
||||
assert response.json['notifications'][0]['flight'] == "FLT0011"
|
||||
assert response.json['notifications'][1]['id'] == 2
|
||||
assert response.json['notifications'][1]['flight'] == "FLT0012"
|
||||
assert response.json['notifications'][2]['id'] == 3
|
||||
assert response.json['notifications'][2]['flight'] == "FLT0013"
|
||||
|
||||
def test_get_notifications_200_offset(client):
|
||||
response = client.get('/api/notifications?offset=2')
|
||||
assert response.status_code == 200
|
||||
assert response.json['offset'] == 2
|
||||
assert response.json['limit'] == 100
|
||||
assert response.json['count'] == 1
|
||||
assert response.json['notifications'][0]['id'] == 3
|
||||
assert response.json['notifications'][0]['flight'] == "FLT0013"
|
||||
|
||||
def test_get_notifications_200_limit(client):
|
||||
response = client.get('/api/notifications?limit=1')
|
||||
assert response.status_code == 200
|
||||
assert response.json['offset'] == 0
|
||||
assert response.json['limit'] == 1
|
||||
assert response.json['count'] == 1
|
||||
assert response.json['notifications'][0]['id'] == 1
|
||||
assert response.json['notifications'][0]['flight'] == "FLT0011"
|
||||
|
||||
def test_get_notifications_200_offset_and_limit(client):
|
||||
response = client.get('/api/notifications?offset=1&limit=1')
|
||||
assert response.status_code == 200
|
||||
assert response.json['offset'] == 1
|
||||
assert response.json['limit'] == 1
|
||||
assert response.json['count'] == 1
|
||||
assert response.json['notifications'][0]['id'] == 2
|
||||
assert response.json['notifications'][0]['flight'] == "FLT0012"
|
||||
|
||||
def test_get_notifications_400_offset_less_than_0(client):
|
||||
response = client.get('/api/notifications?offset=-1')
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_get_notifications_400_limit_less_than_0(client):
|
||||
response = client.get('/api/notifications?limit=-1')
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_get_notifications_400_limit_greater_than_1000(client):
|
||||
response = client.get('/api/notifications?limit=1001')
|
||||
assert response.status_code == 400
|
||||
|
|
@ -0,0 +1,87 @@
|
|||
from flask_jwt_extended import create_access_token
|
||||
|
||||
# PUT /setting/{id}
|
||||
|
||||
def test_put_setting_204(client, app):
|
||||
with app.app_context():
|
||||
access_token = create_access_token(identity="developer")
|
||||
request_headers = {
|
||||
'Authorization': 'Bearer {}'.format(access_token),
|
||||
}
|
||||
request_json = {
|
||||
'name': 'setting_three',
|
||||
'value': 'Updated Setting Three'
|
||||
}
|
||||
response = client.put('/api/setting', headers=request_headers, json=request_json)
|
||||
assert response.status_code == 204
|
||||
|
||||
def test_put_setting_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 = {
|
||||
'value': 'New Value Two'
|
||||
}
|
||||
response = client.put('/api/setting', headers=request_headers, json=request_json)
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_put_setting_400_missing_value(client, app):
|
||||
with app.app_context():
|
||||
access_token = create_access_token(identity="developer")
|
||||
request_headers = {
|
||||
'Authorization': 'Bearer {}'.format(access_token),
|
||||
}
|
||||
request_json = {
|
||||
'name': 'setting_two'
|
||||
}
|
||||
response = client.put('/api/setting', headers=request_headers, json=request_json)
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_put_setting_404(client, app):
|
||||
with app.app_context():
|
||||
access_token = create_access_token(identity="developer")
|
||||
request_headers = {
|
||||
'Authorization': 'Bearer {}'.format(access_token),
|
||||
}
|
||||
request_json = {
|
||||
'name': 'setting_four',
|
||||
'value': 'Updated Setting Four'
|
||||
}
|
||||
response = client.put('/api/setting', headers=request_headers, json=request_json)
|
||||
assert response.status_code == 404
|
||||
|
||||
# GET /setting
|
||||
|
||||
def test_get_links_200(client):
|
||||
response = client.get('/api/setting/setting_three')
|
||||
assert response.status_code == 200
|
||||
assert response.json['id'] == 3
|
||||
assert response.json['name'] == "setting_three"
|
||||
assert response.json['value'] == "Value Three"
|
||||
|
||||
def test_get_link_404(client):
|
||||
response = client.get('/api/link/setting_four')
|
||||
assert response.status_code == 404
|
||||
|
||||
# GET /settings
|
||||
|
||||
def test_get_settings_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/settings', headers=request_headers)
|
||||
assert response.status_code == 200
|
||||
assert response.json[0]['id'] == 1
|
||||
assert response.json[0]['name'] == "setting_one"
|
||||
assert response.json[0]['value'] == "Value One"
|
||||
assert response.json[1]['id'] == 3
|
||||
assert response.json[1]['name'] == "setting_three"
|
||||
assert response.json[1]['value'] == "Value Three"
|
||||
assert response.json[2]['id'] == 2
|
||||
assert response.json[2]['name'] == "setting_two"
|
||||
assert response.json[2]['value'] == "Value Two"
|
||||
Ładowanie…
Reference in New Issue