darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-02-04 02:11:11 -05:00
Code Issues Packages Projects Releases Wiki Activity

Changed the token endpoints so that you can renew a token and revoke a token.

Browse Source 
Updated create service so the token is created at that time
This commit is contained in:
Rebecca Law
2016-01-13 16:04:56 +00:00
parent a6dda26ba1
commit 255dddaa27
4 changed files with 85 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
app/dao/tokens_dao.py
View File

@@ -11,12 +11,11 @@ def save_token_model(token, update_dict={}):
db.session.commit()
def get_model_tokens(service_id=None):
def get_model_tokens(service_id=None, raise_=True):
if service_id:
return Token.query.filter_by(service_id=service_id).one()
# If expiry date is None the token is active
if raise_:
return Token.query.filter_by(service_id=service_id, expiry_date=None).one()
else:
return Token.query.filter_by(service_id=service_id, expiry_date=None).first()
return Token.query.filter_by().all()
def delete_model_token(token):
db.session.delete(token)
db.session.commit()

42
app/service/views/rest.py
View File

@@ -5,7 +5,7 @@ from sqlalchemy.exc import DataError
from sqlalchemy.orm.exc import NoResultFound
from app.dao.services_dao import (
save_model_service, get_model_services, delete_model_service)
from app.dao.tokens_dao import (save_token_model, get_model_tokens, delete_model_token)
from app.dao.tokens_dao import (save_token_model, get_model_tokens)
from app.dao.users_dao import get_model_users
from app.dao.templates_dao import (
save_model_template, get_model_templates)
@@ -30,9 +30,11 @@ def create_service():
# db.session.commit
try:
save_model_service(service)
token = str(uuid.uuid4())
save_token_model(Token(service_id=service.id, token=_generate_token(token)))
except DAOException as e:
return jsonify(result="error", message=str(e)), 400
return jsonify(data=service_schema.dump(service).data), 201
return jsonify(data=service_schema.dump(service).data, token=token), 201
# TODO auth to be added
@@ -79,8 +81,8 @@ def get_service(service_id=None):
# TODO auth to be added
@service.route('/<int:service_id>/token', methods=['POST'])
def create_token(service_id=None):
@service.route('/<int:service_id>/token/renew', methods=['POST'])
def renew_token(service_id=None):
try:
service = get_model_services(service_id=service_id)
except DataError:
@@ -90,27 +92,31 @@ def create_token(service_id=None):
token = _generate_token()
try:
try:
service_token = get_model_tokens(service_id=service_id)
save_token_model(service_token, update_dict={'id': service_token.id,
'token': service_token.token,
'expiry_date': datetime.now()})
except NoResultFound:
pass
service_token = get_model_tokens(service_id=service_id, raise_=False)
if service_token:
# expire existing token
save_token_model(service_token, update_dict={'id': service_token.id, 'expiry_date': datetime.now()})
# create a new one
save_token_model(Token(service_id=service_id, token=token))
except DAOException as e:
return jsonify(result='error', message=str(e)), 400
return jsonify(token=str(token)), 201
unsigned_token = str(_get_token(token))
return jsonify(data=unsigned_token), 201
@service.route('/<int:service_id>/token', methods=['DELETE'])
def delete_token(service_id):
@service.route('/<int:service_id>/token/revoke', methods=['POST'])
def revoke_token(service_id):
try:
token = get_model_tokens(service_id=service_id)
delete_model_token(token)
return jsonify(data=token_schema.dump(token).data), 202
service = get_model_services(service_id=service_id)
except DataError:
return jsonify(result="error", message="Invalid service id"), 400
except NoResultFound:
return jsonify(result="error", message="Token not found"), 404
return jsonify(result="error", message="Service not found"), 404
service_token = get_model_tokens(service_id=service_id, raise_=False)
if service_token:
save_token_model(service_token, update_dict={'id': service_token.id, 'expiry_date': datetime.now()})
return jsonify(data=token_schema.dump(service_token)), 202
def _generate_token(token=None):

39
tests/app/dao/test_tokens_dao.py
View File

@@ -1,10 +1,13 @@
import uuid
from app.dao import tokens_dao
from datetime import datetime
from app.models import Token
from pytest import fail
from sqlalchemy.orm.exc import NoResultFound
def test_should_create_token(notify_api, notify_db, notify_db_session, sample_service):
def test_save_token_should_create_new_token(notify_api, notify_db, notify_db_session, sample_service):
token = uuid.uuid4()
api_token = Token(**{'token': token, 'service_id': sample_service.id})
@@ -15,16 +18,29 @@ def test_should_create_token(notify_api, notify_db, notify_db_session, sample_se
assert all_tokens[0].token == str(token)
def test_should_delete_api_token(notify_api, notify_db, notify_db_session, sample_service):
token = uuid.uuid4()
api_token = Token(**{'token': token, 'service_id': sample_service.id})
def test_save_token_should_update_the_token(notify_api, notify_db, notify_db_session, sample_service):
api_token = Token(**{'token': uuid.uuid4(), 'service_id': sample_service.id})
tokens_dao.save_token_model(api_token)
now = datetime.utcnow()
saved_token = tokens_dao.get_model_tokens(sample_service.id)
tokens_dao.save_token_model(saved_token, update_dict={'id': saved_token.id, 'expiry_date': now})
all_tokens = tokens_dao.get_model_tokens()
assert len(all_tokens) == 1
assert all_tokens[0].expiry_date == now
tokens_dao.delete_model_token(all_tokens[0])
empty_token_list = tokens_dao.get_model_tokens()
assert len(empty_token_list) == 0
def test_get_token_should_raise_exception_when_service_does_not_exist(notify_api, notify_db, notify_db_session,
sample_service):
try:
tokens_dao.get_model_tokens(sample_service.id)
fail()
except NoResultFound:
pass
def test_get_token_should_return_none_when_service_does_not_exist(notify_api, notify_db, notify_db_session,
sample_service):
assert tokens_dao.get_model_tokens(service_id=sample_service.id, raise_=False) is None
def test_should_return_token_for_service(notify_api, notify_db, notify_db_session, sample_service):
@@ -34,12 +50,3 @@ def test_should_return_token_for_service(notify_api, notify_db, notify_db_sessio
token = tokens_dao.get_model_tokens(sample_service.id)
assert token.service_id == sample_service.id
assert token.token == str(the_token)
def test_delete_model_token_should_remove_token(notify_api, notify_db, notify_db_session, sample_service):
api_token = Token(**{'token': str(uuid.uuid4()), 'service_id': sample_service.id})
tokens_dao.save_token_model(api_token)
all_tokens = tokens_dao.get_model_tokens()
assert len(all_tokens) == 1
tokens_dao.delete_model_token(all_tokens[0])
assert len(tokens_dao.get_model_tokens()) == 0

43
tests/app/service/views/test_rest.py
View File

@@ -57,6 +57,7 @@ def test_post_service(notify_api, notify_db, notify_db_session, sample_user):
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['name'] == service.name
assert json_resp['data']['limit'] == service.limit
assert json_resp['token'] is not None
def test_post_service_multiple_users(notify_api, notify_db, notify_db_session, sample_user):
@@ -263,10 +264,11 @@ def test_delete_service_not_exists(notify_api, notify_db, notify_db_session, sam
assert Service.query.count() == 1
def test_create_token_should_return_token_when_successful(notify_api, notify_db, notify_db_session, sample_service):
def test_renew_token_should_return_token_when_service_does_not_have_a_valid_token(notify_api, notify_db,
notify_db_session, sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
response = client.post(url_for('service.create_token', service_id=sample_service.id),
response = client.post(url_for('service.renew_token', service_id=sample_service.id),
headers=[('Content-Type', 'application/json')])
assert response.status_code == 201
assert response.get_data is not None
@@ -274,17 +276,17 @@ def test_create_token_should_return_token_when_successful(notify_api, notify_db,
assert saved_token.service_id == sample_service.id
def test_create_token_should_expire_the_old_token_and_create_a_new_token(notify_api, notify_db, notify_db_session,
sample_service):
def test_renew_token_should_expire_the_old_token_and_create_a_new_token(notify_api, notify_db, notify_db_session,
sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
response = client.post(url_for('service.create_token', service_id=sample_service.id),
response = client.post(url_for('service.renew_token', service_id=sample_service.id),
headers=[('Content-Type', 'application/json')])
assert response.status_code == 201
assert len(Token.query.all()) == 1
saved_token = Token.query.first()
response = client.post(url_for('service.create_token', service_id=sample_service.id),
response = client.post(url_for('service.renew_token', service_id=sample_service.id),
headers=[('Content-Type', 'application/json')])
assert response.status_code == 201
all_tokens = Token.query.all()
@@ -301,21 +303,40 @@ def test_create_token_should_return_error_when_service_does_not_exist(notify_api
sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
response = client.post(url_for('service.create_token', service_id=123),
response = client.post(url_for('service.renew_token', service_id=123),
headers=[('Content-Type', 'application/json')])
assert response.status_code == 404
def test_delete_token(notify_api, notify_db, notify_db_session, sample_service):
def test_revoke_token_should_expire_token_for_service(notify_api, notify_db, notify_db_session, sample_service):
with notify_api.test_request_context():
with notify_api.test_client() as client:
client.post(url_for('service.create_token', service_id=sample_service.id),
client.post(url_for('service.renew_token', service_id=sample_service.id),
headers=[('Content-Type', 'application/json')])
assert len(Token.query.all()) == 1
response = client.post(url_for('service.revoke_token', service_id=sample_service.id))
assert response.status_code == 202
all_tokens = Token.query.all()
assert len(all_tokens) == 1
response = client.delete(url_for('service.delete_token', service_id=sample_service.id))
assert response.status_code == 202
assert all_tokens[0].expiry_date is not None
def test_create_service_should_create_new_token_for_service(notify_api, notify_db, notify_db_session, sample_user):
with notify_api.test_request_context():
with notify_api.test_client() as client:
data = {
'name': 'created service',
'users': [sample_user.id],
'limit': 1000,
'restricted': False,
'active': False}
headers = [('Content-Type', 'application/json')]
assert len(Token.query.all()) == 0
resp = client.post(url_for('service.create_service'),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 201
assert len(Token.query.all()) == 1
def test_token_generated_can_be_read_again(notify_api):