mirror of
https://github.com/GSA/notifications-api.git
synced 2026-06-17 19:55:05 -04:00
@@ -34,8 +34,4 @@ def get_invited_user_by_token(token):
|
||||
|
||||
invited_user = get_invited_user_by_id(invited_user_id)
|
||||
|
||||
if not invited_user:
|
||||
message = 'Invited user not found with id: {}'.format(invited_user_id)
|
||||
return jsonify(result='error', message=message), 404
|
||||
|
||||
return jsonify(data=invited_user_schema.dump(invited_user).data), 200
|
||||
|
||||
@@ -10,11 +10,11 @@ def save_invited_user(invited_user):
|
||||
|
||||
|
||||
def get_invited_user(service_id, invited_user_id):
|
||||
return InvitedUser.query.filter_by(service_id=service_id, id=invited_user_id).first()
|
||||
return InvitedUser.query.filter_by(service_id=service_id, id=invited_user_id).one()
|
||||
|
||||
|
||||
def get_invited_user_by_id(invited_user_id):
|
||||
return InvitedUser.query.filter_by(id=invited_user_id).first()
|
||||
return InvitedUser.query.filter_by(id=invited_user_id).one()
|
||||
|
||||
|
||||
def get_invited_users_for_service(service_id):
|
||||
|
||||
@@ -3,7 +3,7 @@ from app.models import Job
|
||||
|
||||
|
||||
def dao_get_job_by_service_id_and_job_id(service_id, job_id):
|
||||
return Job.query.filter_by(service_id=service_id, id=job_id).first()
|
||||
return Job.query.filter_by(service_id=service_id, id=job_id).one()
|
||||
|
||||
|
||||
def dao_get_jobs_by_service_id(service_id):
|
||||
@@ -11,7 +11,7 @@ def dao_get_jobs_by_service_id(service_id):
|
||||
|
||||
|
||||
def dao_get_job_by_id(job_id):
|
||||
return Job.query.filter_by(id=job_id).first()
|
||||
return Job.query.filter_by(id=job_id).one()
|
||||
|
||||
|
||||
def dao_create_job(job):
|
||||
|
||||
@@ -8,7 +8,7 @@ def dao_fetch_all_services():
|
||||
|
||||
|
||||
def dao_fetch_service_by_id(service_id):
|
||||
return Service.query.filter_by(id=service_id).first()
|
||||
return Service.query.filter_by(id=service_id).one()
|
||||
|
||||
|
||||
def dao_fetch_all_services_by_user(user_id):
|
||||
@@ -16,7 +16,7 @@ def dao_fetch_all_services_by_user(user_id):
|
||||
|
||||
|
||||
def dao_fetch_service_by_id_and_user(service_id, user_id):
|
||||
return Service.query.filter(Service.users.any(id=user_id)).filter_by(id=service_id).first()
|
||||
return Service.query.filter(Service.users.any(id=user_id)).filter_by(id=service_id).one()
|
||||
|
||||
|
||||
def dao_create_service(service, user):
|
||||
|
||||
@@ -14,11 +14,11 @@ def dao_update_template(template):
|
||||
|
||||
|
||||
def dao_get_template_by_id_and_service_id(template_id, service_id):
|
||||
return Template.query.filter_by(id=template_id, service_id=service_id).first()
|
||||
return Template.query.filter_by(id=template_id, service_id=service_id).one()
|
||||
|
||||
|
||||
def dao_get_template_by_id(template_id):
|
||||
return Template.query.filter_by(id=template_id).first()
|
||||
return Template.query.filter_by(id=template_id).one()
|
||||
|
||||
|
||||
def dao_get_all_templates_for_service(service_id):
|
||||
|
||||
@@ -69,12 +69,12 @@ def delete_model_user(user):
|
||||
|
||||
def get_model_users(user_id=None):
|
||||
if user_id:
|
||||
return User.query.filter_by(id=user_id).first()
|
||||
return User.query.filter_by(id=user_id).one()
|
||||
return User.query.filter_by().all()
|
||||
|
||||
|
||||
def get_user_by_email(email):
|
||||
return User.query.filter_by(email_address=email).first()
|
||||
return User.query.filter_by(email_address=email).one()
|
||||
|
||||
|
||||
def increment_failed_login_count(user):
|
||||
|
||||
@@ -48,15 +48,15 @@ def register_errors(blueprint):
|
||||
|
||||
@blueprint.app_errorhandler(NoResultFound)
|
||||
def no_result_found(e):
|
||||
current_app.logger.error(e)
|
||||
current_app.logger.exception(e)
|
||||
return jsonify(result='error', message="No result found"), 404
|
||||
|
||||
@blueprint.app_errorhandler(DataError)
|
||||
def data_error(e):
|
||||
current_app.logger.error(e)
|
||||
current_app.logger.exception(e)
|
||||
return jsonify(result='error', message="No result found"), 404
|
||||
|
||||
@blueprint.app_errorhandler(SQLAlchemyError)
|
||||
def db_error(e):
|
||||
current_app.logger.error(e)
|
||||
current_app.logger.exception(e)
|
||||
return jsonify(result='error', message=str(e)), 500
|
||||
|
||||
@@ -43,16 +43,13 @@ def get_invited_users_by_service(service_id):
|
||||
@invite.route('/<invited_user_id>', methods=['GET'])
|
||||
def get_invited_user_by_service_and_id(service_id, invited_user_id):
|
||||
invited_user = get_invited_user(service_id=service_id, invited_user_id=invited_user_id)
|
||||
if not invited_user:
|
||||
return _invited_user_not_found(service_id, invited_user_id)
|
||||
|
||||
return jsonify(data=invited_user_schema.dump(invited_user).data), 200
|
||||
|
||||
|
||||
@invite.route('/<invited_user_id>', methods=['POST'])
|
||||
def update_invited_user(service_id, invited_user_id):
|
||||
fetched = get_invited_user(service_id=service_id, invited_user_id=invited_user_id)
|
||||
if not fetched:
|
||||
return _invited_user_not_found(service_id=service_id, invited_user_id=invited_user_id)
|
||||
|
||||
current_data = dict(invited_user_schema.dump(fetched).data.items())
|
||||
current_data.update(request.get_json())
|
||||
@@ -63,12 +60,6 @@ def update_invited_user(service_id, invited_user_id):
|
||||
return jsonify(data=invited_user_schema.dump(fetched).data), 200
|
||||
|
||||
|
||||
def _invited_user_not_found(service_id, invited_user_id):
|
||||
message = 'Invited user not found for service id: {} and invited user id: {}'.format(service_id,
|
||||
invited_user_id)
|
||||
return jsonify(result='error', message=message), 404
|
||||
|
||||
|
||||
def _create_invitation(invited_user):
|
||||
from utils.url_safe_token import generate_token
|
||||
token = generate_token(str(invited_user.id), current_app.config['SECRET_KEY'], current_app.config['DANGEROUS_SALT'])
|
||||
|
||||
@@ -19,7 +19,7 @@ from app.schemas import job_schema
|
||||
|
||||
from app.celery.tasks import process_job
|
||||
|
||||
job = Blueprint('job', __name__, url_prefix='/service/<service_id>/job')
|
||||
job = Blueprint('job', __name__, url_prefix='/service/<uuid:service_id>/job')
|
||||
|
||||
from app.errors import register_errors
|
||||
|
||||
@@ -29,8 +29,6 @@ register_errors(job)
|
||||
@job.route('/<job_id>', methods=['GET'])
|
||||
def get_job_by_service_and_job_id(service_id, job_id):
|
||||
job = dao_get_job_by_service_id_and_job_id(service_id, job_id)
|
||||
if not job:
|
||||
return jsonify(result="error", message="Job {} not found for service {}".format(job_id, service_id)), 404
|
||||
data, errors = job_schema.dump(job)
|
||||
return jsonify(data=data)
|
||||
|
||||
@@ -44,10 +42,7 @@ def get_jobs_by_service(service_id):
|
||||
|
||||
@job.route('', methods=['POST'])
|
||||
def create_job(service_id):
|
||||
|
||||
service = dao_fetch_service_by_id(service_id)
|
||||
if not service:
|
||||
return jsonify(result="error", message="Service {} not found".format(service_id)), 404
|
||||
dao_fetch_service_by_id(service_id)
|
||||
|
||||
data = request.get_json()
|
||||
data.update({
|
||||
@@ -65,9 +60,6 @@ def create_job(service_id):
|
||||
@job.route('/<job_id>', methods=['POST'])
|
||||
def update_job(service_id, job_id):
|
||||
fetched_job = dao_get_job_by_service_id_and_job_id(service_id, job_id)
|
||||
if not fetched_job:
|
||||
return jsonify(result="error", message="Job {} not found for service {}".format(job_id, service_id)), 404
|
||||
|
||||
current_data = dict(job_schema.dump(fetched_job).data.items())
|
||||
current_data.update(request.get_json())
|
||||
|
||||
|
||||
@@ -32,13 +32,10 @@ from app.errors import register_errors
|
||||
register_errors(notifications)
|
||||
|
||||
|
||||
@notifications.route('/notifications/<string:notification_id>', methods=['GET'])
|
||||
@notifications.route('/notifications/<uuid:notification_id>', methods=['GET'])
|
||||
def get_notifications(notification_id):
|
||||
try:
|
||||
notification = notifications_dao.get_notification(api_user['client'], notification_id)
|
||||
return jsonify({'notification': notification_status_schema.dump(notification).data}), 200
|
||||
except NoResultFound:
|
||||
return jsonify(result="error", message="not found"), 404
|
||||
notification = notifications_dao.get_notification(api_user['client'], notification_id)
|
||||
return jsonify({'notification': notification_status_schema.dump(notification).data}), 200
|
||||
|
||||
|
||||
@notifications.route('/notifications', methods=['GET'])
|
||||
@@ -152,13 +149,6 @@ def send_notification(notification_type):
|
||||
template_id=notification['template'],
|
||||
service_id=service_id
|
||||
)
|
||||
if not template:
|
||||
return jsonify(
|
||||
result="error",
|
||||
message={
|
||||
'template': ['Template {} not found for service {}'.format(notification['template'], service_id)]
|
||||
}
|
||||
), 404
|
||||
|
||||
template_object = Template(template.__dict__, notification.get('personalisation', {}))
|
||||
if template_object.missing_data:
|
||||
|
||||
@@ -19,9 +19,7 @@ def get_permissions():
|
||||
|
||||
@permission.route('/<permission_id>', methods=['GET'])
|
||||
def get_permission(permission_id):
|
||||
inst = permission_dao.get_query(filter_by_dict={'id': permission_id}).first()
|
||||
if not inst:
|
||||
abort(404, 'Permission not found for id: {permission_id}'.format(permission_id))
|
||||
inst = permission_dao.get_query(filter_by_dict={'id': permission_id}).one()
|
||||
data, errors = permission_schema.dump(inst)
|
||||
if errors:
|
||||
abort(500, errors)
|
||||
|
||||
@@ -50,18 +50,14 @@ def get_services():
|
||||
return jsonify(data=data)
|
||||
|
||||
|
||||
@service.route('/<service_id>', methods=['GET'])
|
||||
@service.route('/<uuid:service_id>', methods=['GET'])
|
||||
def get_service_by_id(service_id):
|
||||
user_id = request.args.get('user_id', None)
|
||||
if user_id:
|
||||
fetched = dao_fetch_service_by_id_and_user(service_id, user_id)
|
||||
else:
|
||||
fetched = dao_fetch_service_by_id(service_id)
|
||||
if not fetched:
|
||||
message_with_user_id = 'and for user id: {}'.format(user_id) if user_id else ''
|
||||
return jsonify(result="error",
|
||||
message="Service not found for service id: {0} {1}".format(service_id,
|
||||
message_with_user_id)), 404
|
||||
|
||||
data, errors = service_schema.dump(fetched)
|
||||
return jsonify(data=data)
|
||||
|
||||
@@ -73,8 +69,6 @@ def create_service():
|
||||
return jsonify(result="error", message={'user_id': ['Missing data for required field.']}), 400
|
||||
|
||||
user = get_model_users(data['user_id'])
|
||||
if not user:
|
||||
return jsonify(result="error", message={'user_id': ['not found']}), 400
|
||||
|
||||
data.pop('user_id', None)
|
||||
if 'name' in data:
|
||||
@@ -89,11 +83,9 @@ def create_service():
|
||||
return jsonify(data=service_schema.dump(valid_service).data), 201
|
||||
|
||||
|
||||
@service.route('/<service_id>', methods=['POST'])
|
||||
@service.route('/<uuid:service_id>', methods=['POST'])
|
||||
def update_service(service_id):
|
||||
fetched_service = dao_fetch_service_by_id(service_id)
|
||||
if not fetched_service:
|
||||
return _service_not_found(service_id)
|
||||
|
||||
current_data = dict(service_schema.dump(fetched_service).data.items())
|
||||
current_data.update(request.get_json())
|
||||
@@ -104,11 +96,9 @@ def update_service(service_id):
|
||||
return jsonify(data=service_schema.dump(fetched_service).data), 200
|
||||
|
||||
|
||||
@service.route('/<service_id>/api-key', methods=['POST'])
|
||||
@service.route('/<uuid:service_id>/api-key', methods=['POST'])
|
||||
def renew_api_key(service_id=None):
|
||||
fetched_service = dao_fetch_service_by_id(service_id=service_id)
|
||||
if not fetched_service:
|
||||
return _service_not_found(service_id)
|
||||
|
||||
# create a new one
|
||||
# TODO: what validation should be done here?
|
||||
@@ -120,7 +110,7 @@ def renew_api_key(service_id=None):
|
||||
return jsonify(data=unsigned_api_key), 201
|
||||
|
||||
|
||||
@service.route('/<service_id>/api-key/revoke/<int:api_key_id>', methods=['POST'])
|
||||
@service.route('/<uuid:service_id>/api-key/revoke/<int:api_key_id>', methods=['POST'])
|
||||
def revoke_api_key(service_id, api_key_id):
|
||||
service_api_key = get_model_api_keys(service_id=service_id, id=api_key_id)
|
||||
|
||||
@@ -128,12 +118,11 @@ def revoke_api_key(service_id, api_key_id):
|
||||
return jsonify(), 202
|
||||
|
||||
|
||||
@service.route('/<service_id>/api-keys', methods=['GET'])
|
||||
@service.route('/<service_id>/api-keys/<int:key_id>', methods=['GET'])
|
||||
@service.route('/<uuid:service_id>/api-keys', methods=['GET'])
|
||||
@service.route('/<uuid:service_id>/api-keys/<int:key_id>', methods=['GET'])
|
||||
def get_api_keys(service_id, key_id=None):
|
||||
fetched_service = dao_fetch_service_by_id(service_id=service_id)
|
||||
if not fetched_service:
|
||||
return _service_not_found(service_id)
|
||||
dao_fetch_service_by_id(service_id=service_id)
|
||||
|
||||
try:
|
||||
if key_id:
|
||||
api_keys = [get_model_api_keys(service_id=service_id, id=key_id)]
|
||||
@@ -145,30 +134,19 @@ def get_api_keys(service_id, key_id=None):
|
||||
return jsonify(apiKeys=api_key_schema.dump(api_keys, many=True).data), 200
|
||||
|
||||
|
||||
@service.route('/<service_id>/users', methods=['GET'])
|
||||
@service.route('/<uuid:service_id>/users', methods=['GET'])
|
||||
def get_users_for_service(service_id):
|
||||
fetched = dao_fetch_service_by_id(service_id)
|
||||
if not fetched:
|
||||
return _service_not_found(service_id)
|
||||
# TODO why is this code here, the same functionality exists without it?
|
||||
if not fetched.users:
|
||||
return jsonify(data=[])
|
||||
|
||||
result = user_schema.dump(fetched.users, many=True)
|
||||
return jsonify(data=result.data)
|
||||
|
||||
|
||||
@service.route('/<service_id>/users/<user_id>', methods=['POST'])
|
||||
@service.route('/<uuid:service_id>/users/<user_id>', methods=['POST'])
|
||||
def add_user_to_service(service_id, user_id):
|
||||
service = dao_fetch_service_by_id(service_id)
|
||||
if not service:
|
||||
return _service_not_found(service_id)
|
||||
user = get_model_users(user_id=user_id)
|
||||
|
||||
if not user:
|
||||
return jsonify(result='error',
|
||||
message='User not found for id: {}'.format(user_id)), 404
|
||||
|
||||
if user in service.users:
|
||||
return jsonify(result='error',
|
||||
message='User id: {} already part of service id: {}'.format(user_id, service_id)), 400
|
||||
@@ -181,10 +159,6 @@ def add_user_to_service(service_id, user_id):
|
||||
return jsonify(data=data), 201
|
||||
|
||||
|
||||
def _service_not_found(service_id):
|
||||
return jsonify(result='error', message='Service not found for id: {}'.format(service_id)), 404
|
||||
|
||||
|
||||
def _process_permissions(user, service, permission_groups):
|
||||
from app.permissions_utils import get_permissions_by_group
|
||||
from app.dao.permissions_dao import permission_dao
|
||||
|
||||
@@ -18,7 +18,7 @@ from app.dao.services_dao import (
|
||||
)
|
||||
from app.schemas import template_schema
|
||||
|
||||
template = Blueprint('template', __name__, url_prefix='/service/<service_id>/template')
|
||||
template = Blueprint('template', __name__, url_prefix='/service/<uuid:service_id>/template')
|
||||
|
||||
from app.errors import register_errors
|
||||
|
||||
@@ -28,8 +28,6 @@ register_errors(template)
|
||||
@template.route('', methods=['POST'])
|
||||
def create_template(service_id):
|
||||
fetched_service = dao_fetch_service_by_id(service_id=service_id)
|
||||
if not fetched_service:
|
||||
return jsonify(result="error", message="Service not found"), 404
|
||||
|
||||
new_template, errors = template_schema.load(request.get_json())
|
||||
if errors:
|
||||
@@ -52,8 +50,6 @@ def create_template(service_id):
|
||||
@template.route('/<int:template_id>', methods=['POST'])
|
||||
def update_template(service_id, template_id):
|
||||
fetched_template = dao_get_template_by_id_and_service_id(template_id=template_id, service_id=service_id)
|
||||
if not fetched_template:
|
||||
return jsonify(result="error", message="Template not found"), 404
|
||||
|
||||
current_data = dict(template_schema.dump(fetched_template).data.items())
|
||||
current_data.update(request.get_json())
|
||||
@@ -77,11 +73,8 @@ def get_all_templates_for_service(service_id):
|
||||
@template.route('/<int:template_id>', methods=['GET'])
|
||||
def get_template_by_id_and_service_id(service_id, template_id):
|
||||
fetched_template = dao_get_template_by_id_and_service_id(template_id=template_id, service_id=service_id)
|
||||
if fetched_template:
|
||||
data, errors = template_schema.dump(fetched_template)
|
||||
return jsonify(data=data)
|
||||
else:
|
||||
return jsonify(result="error", message="Template not found"), 404
|
||||
data, errors = template_schema.dump(fetched_template)
|
||||
return jsonify(data=data)
|
||||
|
||||
|
||||
def _strip_html(content):
|
||||
|
||||
@@ -33,7 +33,7 @@ register_errors(user)
|
||||
|
||||
@user.route('', methods=['POST'])
|
||||
def create_user():
|
||||
user, errors = user_schema.load(request.get_json())
|
||||
user_to_create, errors = user_schema.load(request.get_json())
|
||||
req_json = request.get_json()
|
||||
# TODO password policy, what is valid password
|
||||
if not req_json.get('password', None):
|
||||
@@ -41,16 +41,13 @@ def create_user():
|
||||
return jsonify(result="error", message=errors), 400
|
||||
if errors:
|
||||
return jsonify(result="error", message=errors), 400
|
||||
save_model_user(user, pwd=req_json.get('password'))
|
||||
return jsonify(data=user_schema.dump(user).data), 201
|
||||
save_model_user(user_to_create, pwd=req_json.get('password'))
|
||||
return jsonify(data=user_schema.dump(user_to_create).data), 201
|
||||
|
||||
|
||||
@user.route('/<int:user_id>', methods=['PUT'])
|
||||
def update_user(user_id):
|
||||
user_to_update = get_model_users(user_id=user_id)
|
||||
if not user_to_update:
|
||||
return _user_not_found(user_id)
|
||||
|
||||
req_json = request.get_json()
|
||||
update_dct, errors = user_schema_load_json.load(req_json)
|
||||
pwd = req_json.get('password', None)
|
||||
@@ -116,10 +113,6 @@ def verify_user_code(user_id):
|
||||
@user.route('/<int:user_id>/sms-code', methods=['POST'])
|
||||
def send_user_sms_code(user_id):
|
||||
user_to_send_to = get_model_users(user_id=user_id)
|
||||
|
||||
if not user_to_send_to:
|
||||
return _user_not_found(user_id)
|
||||
|
||||
verify_code, errors = request_verify_code_schema.load(request.get_json())
|
||||
if errors:
|
||||
return jsonify(result="error", message=errors), 400
|
||||
@@ -139,9 +132,6 @@ def send_user_sms_code(user_id):
|
||||
@user.route('/<int:user_id>/email-code', methods=['POST'])
|
||||
def send_user_email_code(user_id):
|
||||
user_to_send_to = get_model_users(user_id=user_id)
|
||||
if not user_to_send_to:
|
||||
return _user_not_found(user_id)
|
||||
|
||||
verify_code, errors = request_verify_code_schema.load(request.get_json())
|
||||
if errors:
|
||||
return jsonify(result="error", message=errors), 400
|
||||
@@ -162,22 +152,16 @@ def send_user_email_code(user_id):
|
||||
@user.route('', methods=['GET'])
|
||||
def get_user(user_id=None):
|
||||
users = get_model_users(user_id=user_id)
|
||||
if not users:
|
||||
return jsonify(result="error", message="not found"), 404
|
||||
result = user_schema.dump(users, many=True) if isinstance(users, list) else user_schema.dump(users)
|
||||
return jsonify(data=result.data)
|
||||
|
||||
|
||||
@user.route('/<int:user_id>/service/<service_id>/permission', methods=['POST'])
|
||||
@user.route('/<int:user_id>/service/<uuid:service_id>/permission', methods=['POST'])
|
||||
def set_permissions(user_id, service_id):
|
||||
# TODO fix security hole, how do we verify that the user
|
||||
# who is making this request has permission to make the request.
|
||||
user = get_model_users(user_id=user_id)
|
||||
if not user:
|
||||
_user_not_found(user_id)
|
||||
service = dao_fetch_service_by_id(service_id=service_id)
|
||||
if not service:
|
||||
abort(404, 'Service not found for id: {}'.format(service_id))
|
||||
permissions, errors = permission_schema.load(request.get_json(), many=True)
|
||||
if errors:
|
||||
abort(400, errors)
|
||||
@@ -194,8 +178,6 @@ def get_by_email():
|
||||
if not email:
|
||||
return jsonify(result="error", message="invalid request"), 400
|
||||
fetched_user = get_user_by_email(email)
|
||||
if not fetched_user:
|
||||
return _user_not_found_for_email()
|
||||
result = user_schema.dump(fetched_user)
|
||||
|
||||
return jsonify(data=result.data)
|
||||
@@ -208,8 +190,6 @@ def send_user_reset_password():
|
||||
return jsonify(result="error", message=errors), 400
|
||||
|
||||
user_to_send_to = get_user_by_email(email['email'])
|
||||
if not user_to_send_to:
|
||||
return _user_not_found_for_email()
|
||||
|
||||
reset_password_message = {'to': user_to_send_to.email_address,
|
||||
'name': user_to_send_to.name,
|
||||
@@ -220,14 +200,6 @@ def send_user_reset_password():
|
||||
return jsonify({}), 204
|
||||
|
||||
|
||||
def _user_not_found(user_id):
|
||||
return abort(404, 'User not found for id: {}'.format(user_id))
|
||||
|
||||
|
||||
def _user_not_found_for_email():
|
||||
return abort(404, 'User not found for email address')
|
||||
|
||||
|
||||
def _create_reset_password_url(email):
|
||||
from utils.url_safe_token import generate_token
|
||||
import json
|
||||
|
||||
@@ -1,5 +1,9 @@
|
||||
from datetime import datetime, timedelta
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.orm.exc import NoResultFound
|
||||
|
||||
from app import db
|
||||
|
||||
from app.models import InvitedUser
|
||||
@@ -50,8 +54,9 @@ def test_get_invited_user_by_id(notify_db, notify_db_session, sample_invited_use
|
||||
def test_get_unknown_invited_user_returns_none(notify_db, notify_db_session, sample_service):
|
||||
unknown_id = uuid.uuid4()
|
||||
|
||||
unknown = get_invited_user(sample_service.id, unknown_id)
|
||||
assert unknown is None
|
||||
with pytest.raises(NoResultFound) as e:
|
||||
get_invited_user(sample_service.id, unknown_id)
|
||||
assert 'No row was found for one()' in str(e.value)
|
||||
|
||||
|
||||
def test_get_invited_users_for_service(notify_db, notify_db_session, sample_service):
|
||||
|
||||
@@ -11,7 +11,7 @@ from app.dao.services_dao import (
|
||||
)
|
||||
from app.dao.users_dao import save_model_user
|
||||
from app.models import Service, User
|
||||
from sqlalchemy.orm.exc import FlushError
|
||||
from sqlalchemy.orm.exc import FlushError, NoResultFound
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
|
||||
|
||||
@@ -145,7 +145,9 @@ def test_get_all_user_services_should_return_empty_list_if_no_services_for_user(
|
||||
|
||||
|
||||
def test_get_service_by_id_returns_none_if_no_service(notify_db):
|
||||
assert not dao_fetch_service_by_id(str(uuid.uuid4()))
|
||||
with pytest.raises(NoResultFound) as e:
|
||||
dao_fetch_service_by_id(str(uuid.uuid4()))
|
||||
assert 'No row was found for one()' in str(e)
|
||||
|
||||
|
||||
def test_get_service_by_id_returns_service(service_factory):
|
||||
@@ -169,4 +171,6 @@ def test_cannot_get_service_by_id_and_owned_by_different_user(service_factory, s
|
||||
save_model_user(new_user)
|
||||
service2 = service_factory.get('service 2', new_user)
|
||||
assert dao_fetch_service_by_id_and_user(service1.id, sample_user.id).name == 'service 1'
|
||||
assert not dao_fetch_service_by_id_and_user(service2.id, sample_user.id)
|
||||
with pytest.raises(NoResultFound) as e:
|
||||
dao_fetch_service_by_id_and_user(service2.id, sample_user.id)
|
||||
assert 'No row was found for one()' in str(e)
|
||||
|
||||
@@ -155,4 +155,6 @@ def test_get_template_by_id_and_service(notify_db, notify_db_session, sample_ser
|
||||
|
||||
|
||||
def test_get_template_by_id_and_service_returns_none_if_no_template(sample_service):
|
||||
assert not dao_get_template_by_id_and_service_id(template_id=999, service_id=sample_service.id)
|
||||
with pytest.raises(NoResultFound) as e:
|
||||
dao_get_template_by_id_and_service_id(template_id=999, service_id=sample_service.id)
|
||||
assert 'No row was found for one' in str(e.value)
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
from datetime import datetime, timedelta
|
||||
from sqlalchemy.exc import DataError
|
||||
from sqlalchemy.orm.exc import NoResultFound
|
||||
|
||||
from app import db
|
||||
import pytest
|
||||
|
||||
@@ -55,7 +57,7 @@ def test_get_user_not_exists(notify_api, notify_db, notify_db_session):
|
||||
try:
|
||||
get_model_users(user_id="12345")
|
||||
pytest.fail("NoResultFound exception not thrown.")
|
||||
except:
|
||||
except NoResultFound as e:
|
||||
pass
|
||||
|
||||
|
||||
|
||||
@@ -233,8 +233,7 @@ def test_update_invited_user_for_wrong_service_returns_404(notify_api, sample_in
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
assert response.status_code == 404
|
||||
json_response = json.loads(response.get_data(as_text=True))['message']
|
||||
assert json_response == 'Invited user not found for service id: {} and invited user id: {}'\
|
||||
.format(bad_service_id, sample_invited_user.id)
|
||||
assert json_response == 'No result found'
|
||||
|
||||
|
||||
def test_update_invited_user_for_invalid_data_returns_400(notify_api, sample_invited_user):
|
||||
|
||||
@@ -68,7 +68,7 @@ def test_get_job_with_unknown_id_returns404(notify_api, sample_template):
|
||||
assert response.status_code == 404
|
||||
resp_json = json.loads(response.get_data(as_text=True))
|
||||
assert resp_json == {
|
||||
'message': 'Job {} not found for service {}'.format(random_id, service_id),
|
||||
'message': 'No result found',
|
||||
'result': 'error'
|
||||
}
|
||||
|
||||
@@ -182,7 +182,7 @@ def test_create_job_returns_404_if_missing_service(notify_api, sample_template,
|
||||
|
||||
app.celery.tasks.process_job.apply_async.assert_not_called()
|
||||
assert resp_json['result'] == 'error'
|
||||
assert resp_json['message'] == 'Service {} not found'.format(random_id)
|
||||
assert resp_json['message'] == 'No result found'
|
||||
|
||||
|
||||
def test_get_update_job(notify_api, sample_job):
|
||||
|
||||
@@ -45,7 +45,7 @@ def test_get_notifications_empty_result(notify_api, sample_api_key):
|
||||
|
||||
notification = json.loads(response.get_data(as_text=True))
|
||||
assert notification['result'] == "error"
|
||||
assert notification['message'] == "not found"
|
||||
assert notification['message'] == "No result found"
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
@@ -329,9 +329,8 @@ def test_send_notification_invalid_template_id(notify_api, sample_template, mock
|
||||
app.celery.tasks.send_sms.apply_async.assert_not_called()
|
||||
|
||||
assert response.status_code == 404
|
||||
assert len(json_resp['message'].keys()) == 1
|
||||
test_string = 'Template {} not found for service {}'.format(9999, sample_template.service.id)
|
||||
assert test_string in json_resp['message']['template']
|
||||
test_string = 'No result found'
|
||||
assert test_string in json_resp['message']
|
||||
|
||||
|
||||
@freeze_time("2016-01-01 11:09:00.061258")
|
||||
@@ -495,8 +494,8 @@ def test_should_not_allow_template_from_another_service(notify_api, service_fact
|
||||
app.celery.tasks.send_sms.apply_async.assert_not_called()
|
||||
|
||||
assert response.status_code == 404
|
||||
test_string = 'Template {} not found for service {}'.format(service_2_templates[0].id, service_1.id)
|
||||
assert test_string in json_resp['message']['template']
|
||||
test_string = 'No result found'
|
||||
assert test_string in json_resp['message']
|
||||
|
||||
|
||||
@freeze_time("2016-01-01 11:09:00.061258")
|
||||
@@ -612,11 +611,8 @@ def test_should_reject_email_notification_with_template_id_that_cant_be_found(
|
||||
app.celery.tasks.send_email.apply_async.assert_not_called()
|
||||
assert response.status_code == 404
|
||||
assert data['result'] == 'error'
|
||||
test_string = 'Template {} not found for service {}'.format(
|
||||
1234,
|
||||
sample_email_template.service.id
|
||||
)
|
||||
assert test_string in data['message']['template']
|
||||
test_string = 'No result found'
|
||||
assert test_string in data['message']
|
||||
|
||||
|
||||
def test_should_not_allow_email_template_from_another_service(notify_api, service_factory, sample_user, mocker):
|
||||
@@ -649,8 +645,8 @@ def test_should_not_allow_email_template_from_another_service(notify_api, servic
|
||||
app.celery.tasks.send_email.apply_async.assert_not_called()
|
||||
|
||||
assert response.status_code == 404
|
||||
test_string = 'Template {} not found for service {}'.format(service_2_templates[0].id, service_1.id)
|
||||
assert test_string in json_resp['message']['template']
|
||||
test_string = 'No result found'
|
||||
assert test_string in json_resp['message']
|
||||
|
||||
|
||||
def test_should_not_send_email_if_restricted_and_not_a_service_user(notify_api, sample_email_template, mocker):
|
||||
|
||||
@@ -130,7 +130,7 @@ def test_get_service_by_id_should_404_if_no_service(notify_api, notify_db):
|
||||
assert resp.status_code == 404
|
||||
json_resp = json.loads(resp.get_data(as_text=True))
|
||||
assert json_resp['result'] == 'error'
|
||||
assert json_resp['message'] == 'Service not found for service id: {} '.format(service_id)
|
||||
assert json_resp['message'] == 'No result found'
|
||||
|
||||
|
||||
def test_get_service_by_id_and_user(notify_api, service_factory, sample_user):
|
||||
@@ -166,8 +166,7 @@ def test_get_service_by_id_should_404_if_no_service_for_user(notify_api, sample_
|
||||
assert resp.status_code == 404
|
||||
json_resp = json.loads(resp.get_data(as_text=True))
|
||||
assert json_resp['result'] == 'error'
|
||||
assert json_resp['message'] == \
|
||||
'Service not found for service id: {0} and for user id: {1}'.format(service_id, sample_user.id)
|
||||
assert json_resp['message'] == 'No result found'
|
||||
|
||||
|
||||
def test_create_service(notify_api, sample_user):
|
||||
@@ -257,9 +256,9 @@ def test_should_not_create_service_with_missing_if_user_id_is_not_in_database(no
|
||||
data=json.dumps(data),
|
||||
headers=headers)
|
||||
json_resp = json.loads(resp.get_data(as_text=True))
|
||||
assert resp.status_code == 400
|
||||
assert resp.status_code == 404
|
||||
assert json_resp['result'] == 'error'
|
||||
assert 'not found' in json_resp['message']['user_id']
|
||||
assert 'No result found' == json_resp['message']
|
||||
|
||||
|
||||
def test_should_not_create_service_if_missing_data(notify_api, sample_user):
|
||||
@@ -467,7 +466,7 @@ def test_get_users_for_service_returns_404_when_service_does_not_exist(notify_ap
|
||||
assert response.status_code == 404
|
||||
result = json.loads(response.get_data(as_text=True))
|
||||
assert result['result'] == 'error'
|
||||
assert result['message'] == 'Service not found for id: {}'.format(service_id)
|
||||
assert result['message'] == 'No result found'
|
||||
|
||||
|
||||
def test_default_permissions_are_added_for_user_service(notify_api,
|
||||
@@ -779,7 +778,7 @@ def test_add_existing_user_to_non_existing_service_returns404(notify_api,
|
||||
)
|
||||
|
||||
result = json.loads(resp.get_data(as_text=True))
|
||||
expected_message = 'Service not found for id: {}'.format(incorrect_id)
|
||||
expected_message = 'No result found'
|
||||
|
||||
assert resp.status_code == 404
|
||||
assert result['result'] == 'error'
|
||||
@@ -833,7 +832,7 @@ def test_add_unknown_user_to_service_returns404(notify_api, notify_db, notify_db
|
||||
)
|
||||
|
||||
result = json.loads(resp.get_data(as_text=True))
|
||||
expected_message = 'User not found for id: {}'.format(incorrect_id)
|
||||
expected_message = 'No result found'
|
||||
|
||||
assert resp.status_code == 404
|
||||
assert result['result'] == 'error'
|
||||
|
||||
@@ -92,7 +92,7 @@ def test_should_be_error_if_service_does_not_exist_on_create(notify_api):
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
assert response.status_code == 404
|
||||
assert json_resp['result'] == 'error'
|
||||
assert json_resp['message'] == 'Service not found'
|
||||
assert json_resp['message'] == 'No result found'
|
||||
|
||||
|
||||
def test_should_be_error_if_service_does_not_exist_on_update(notify_api):
|
||||
@@ -117,7 +117,7 @@ def test_should_be_error_if_service_does_not_exist_on_update(notify_api):
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
assert response.status_code == 404
|
||||
assert json_resp['result'] == 'error'
|
||||
assert json_resp['message'] == 'Template not found'
|
||||
assert json_resp['message'] == 'No result found'
|
||||
|
||||
|
||||
def test_must_have_a_subject_on_an_email_template(notify_api, sample_service):
|
||||
@@ -397,17 +397,18 @@ def test_should_return_404_if_no_templates_for_service_with_id(notify_api, sampl
|
||||
with notify_api.test_request_context():
|
||||
with notify_api.test_client() as client:
|
||||
|
||||
uuid_ = uuid.uuid4()
|
||||
auth_header = create_authorization_header(
|
||||
path='/service/{}/template/{}'.format(sample_service.id, 111),
|
||||
path='/service/{}/template/{}'.format(sample_service.id, 9999),
|
||||
method='GET'
|
||||
)
|
||||
|
||||
response = client.get(
|
||||
'/service/{}/template/{}'.format(sample_service.id, 111),
|
||||
'/service/{}/template/{}'.format(sample_service.id, 9999),
|
||||
headers=[auth_header]
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
json_resp = json.loads(response.get_data(as_text=True))
|
||||
assert json_resp['result'] == 'error'
|
||||
assert json_resp['message'] == 'Template not found'
|
||||
assert json_resp['message'] == 'No result found'
|
||||
|
||||
@@ -258,7 +258,7 @@ def test_put_user_not_exists(notify_api, notify_db, notify_db_session, sample_us
|
||||
user = User.query.filter_by(id=sample_user.id).first()
|
||||
json_resp = json.loads(resp.get_data(as_text=True))
|
||||
assert json_resp['result'] == "error"
|
||||
assert json_resp['message'] == "User not found for id: {}".format("9999")
|
||||
assert json_resp['message'] == 'No result found'
|
||||
|
||||
assert user == sample_user
|
||||
assert user.email_address != new_email
|
||||
@@ -299,7 +299,7 @@ def test_get_user_by_email_not_found_returns_404(notify_api,
|
||||
assert resp.status_code == 404
|
||||
json_resp = json.loads(resp.get_data(as_text=True))
|
||||
assert json_resp['result'] == 'error'
|
||||
assert json_resp['message'] == 'User not found for email address'
|
||||
assert json_resp['message'] == 'No result found'
|
||||
|
||||
|
||||
def test_get_user_by_email_bad_url_returns_404(notify_api,
|
||||
@@ -469,7 +469,7 @@ def test_send_user_reset_password_should_return_400_when_user_doesnot_exist(noti
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
|
||||
assert resp.status_code == 404
|
||||
assert json.loads(resp.get_data(as_text=True))['message'] == 'User not found for email address'
|
||||
assert json.loads(resp.get_data(as_text=True))['message'] == 'No result found'
|
||||
|
||||
|
||||
def test_send_user_reset_password_should_return_400_when_data_is_not_email_address(notify_api, mocker):
|
||||
|
||||
@@ -298,7 +298,7 @@ def test_send_sms_code_returns_404_for_bad_input_data(notify_api, notify_db, not
|
||||
data=data,
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
assert resp.status_code == 404
|
||||
assert json.loads(resp.get_data(as_text=True))['message'] == 'User not found for id: {}'.format(int(uuid_))
|
||||
assert json.loads(resp.get_data(as_text=True))['message'] == 'No result found'
|
||||
|
||||
|
||||
def test_send_user_email_code(notify_api,
|
||||
@@ -340,4 +340,4 @@ def test_send_user_email_code_returns_404_for_when_user_does_not_exist(notify_ap
|
||||
data=data,
|
||||
headers=[('Content-Type', 'application/json'), auth_header])
|
||||
assert resp.status_code == 404
|
||||
assert json.loads(resp.get_data(as_text=True))['message'] == 'User not found for id: {}'.format(1)
|
||||
assert json.loads(resp.get_data(as_text=True))['message'] == 'No result found'
|
||||
|
||||
Reference in New Issue
Block a user