# Coverage for views.py : 43%

## 208 statements 96 run 112 missing 0 excluded 10 partial

Hot-keys on this page

r m x p toggle line displays

j k next/prev highlighted chunk

0 (zero) top of page

1 (one) first highlighted chunk

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 | from flask import Blueprint, request, jsonify, make_response from flask_restful import Api, Resource from http_status import HttpStatus from models import orm, NotificationCategory, NotificationCategorySchema, Notification, NotificationSchema from sqlalchemy.exc import SQLAlchemyError from helpers import PaginationHelper from flask_httpauth import HTTPBasicAuth from flask import g from models import User, UserSchema auth = HTTPBasicAuth() @auth.verify_password def verify_user_password(name, password): user = User.query.filter_by(name=name).first() if not user or not user.verify_password(password): return False g.user = user return True class AuthenticationRequiredResource(Resource): method_decorators = [auth.login_required] user_schema = UserSchema() service_blueprint = Blueprint('service', __name__) notification_category_schema = NotificationCategorySchema() notification_schema = NotificationSchema() service = Api(service_blueprint) class UserResource(AuthenticationRequiredResource): def get(self, id): user = User.query.get_or_404(id) result = user_schema.dump(user).data return result class UserListResource(Resource): @auth.login_required def get(self): pagination_helper = PaginationHelper( request, query=User.query, resource_for_url='service.userlistresource', key_name='results', schema=user_schema) result = pagination_helper.paginate_query() return result def post(self): user_dict = request.get_json() 56 ↛ 57 line 56 didn't jump to line 57, because the condition on line 56 was never true if not user_dict: response = {'user': 'No input data provided'} return response, HttpStatus.bad_request_400.value errors = user_schema.validate(user_dict) 60 ↛ 61 line 60 didn't jump to line 61, because the condition on line 60 was never true if errors: return errors, HttpStatus.bad_request_400.value user_name = user_dict['name'] existing_user = User.query.filter_by(name=user_name).first() 64 ↛ 65 line 64 didn't jump to line 65, because the condition on line 64 was never true if existing_user is not None: response = {'user': 'An user with the name {} already exists'.format(user_name)} return response, HttpStatus.bad_request_400.value try: user = User(name=user_name) error_message, password_ok = \ user.check_password_strength_and_hash_if_ok(user_dict['password']) 71 ↛ 77 line 71 didn't jump to line 77, because the condition on line 71 was never false if password_ok: user.add(user) query = User.query.get(user.id) dump_result = user_schema.dump(query).data return dump_result, HttpStatus.created_201.value else: return {"error": error_message}, HttpStatus.bad_request_400.value except SQLAlchemyError as e: orm.session.rollback() response = {"error": str(e)} return response, HttpStatus.bad_request_400.value class NotificationResource(AuthenticationRequiredResource): def get(self, id): notification = Notification.query.get_or_404(id) dumped_notification = notification_schema.dump(notification).data return dumped_notification def patch(self, id): notification = Notification.query.get_or_404(id) notification_dict = request.get_json(force=True) print(notification_dict) if 'message' in notification_dict and notification_dict['message'] is not None: notification_message = notification_dict['message'] if not Notification.is_message_unique(id=0, message=notification_message): response = {'error': 'A notification with the message {} already exists'.format(notification_message)} return response, HttpStatus.bad_request_400.value notification.message = notification_message if 'ttl' in notification_dict and notification_dict['ttl'] is not None: notification.duration = notification_dict['duration'] if 'displayed_times' in notification_dict and notification_dict['displayed_times'] is not None: notification.displayed_times = notification_dict['displayed_times'] if 'displayed_once' in notification_dict and notification_dict['displayed_once'] is not None: notification.displayed_once = notification_dict['displayed_once'] == 'true' dumped_notification, dump_errors = notification_schema.dump(notification) if dump_errors: return dump_errors, HttpStatus.bad_request_400.value validate_errors = notification_schema.validate(dumped_notification) if validate_errors: return validate_errors, HttpStatus.bad_request_400.value try: notification.update() return self.get(id) except SQLAlchemyError as e: orm.session.rollback() response = {"error": str(e)} return response, HttpStatus.bad_request_400.value def delete(self, id): notification = Notification.query.get_or_404(id) try: delete = notification.delete(notification) response = make_response() return response, HttpStatus.no_content_204.value except SQLAlchemyError as e: orm.session.rollback() response = {"error": str(e)} return response, HttpStatus.unauthorized_401.value class NotificationListResource(AuthenticationRequiredResource): def get(self): pagination_helper = PaginationHelper( request, query=Notification.query, resource_for_url='service.notificationlistresource', key_name='results', schema=notification_schema) pagination_result = pagination_helper.paginate_query() return pagination_result def post(self): notification_category_dict = request.get_json() if not notification_category_dict: response = {'message': 'No input data provided'} return response, HttpStatus.bad_request_400.value errors = notification_schema.validate(notification_category_dict) if errors: return errors, HttpStatus.bad_request_400.value notification_message = notification_category_dict['message'] if not Notification.is_message_unique(id=0, message=notification_message): response = {'error': 'A notification with the message {} already exists'.format(notification_message)} return response, HttpStatus.bad_request_400.value try: notification_category_name = notification_category_dict['notification_category']['name'] notification_category = NotificationCategory.query.filter_by(name=notification_category_name).first() if notification_category is None: # Create a new NotificationCategory notification_category = NotificationCategory(name=notification_category_name) orm.session.add(notification_category) # Now that we are sure we have a notification category, # we can create a new Notification notification = Notification( message=notification_message, ttl=notification_category_dict['ttl'], notification_category=notification_category) notification.add(notification) query = Notification.query.get(notification.id) dump_result = notification_schema.dump(query).data return dump_result, HttpStatus.created_201.value except SQLAlchemyError as e: orm.session.rollback() response = {"error": str(e)} return response, HttpStatus.bad_request_400.value class NotificationCategoryResource(AuthenticationRequiredResource): def get(self, id): notification_category = NotificationCategory.query.get_or_404(id) dump_result = notification_category_schema.dump(notification_category).data return dump_result def patch(self, id): notification_category = NotificationCategory.query.get_or_404(id) notification_category_dict = request.get_json() 187 ↛ 188 line 187 didn't jump to line 188, because the condition on line 187 was never true if not notification_category_dict: response = {'message': 'No input data provided'} return response, HttpStatus.bad_request_400.value errors = notification_category_schema.validate(notification_category_dict) 191 ↛ 192 line 191 didn't jump to line 192, because the condition on line 191 was never true if errors: return errors, HttpStatus.bad_request_400.value try: 194 ↛ 201 line 194 didn't jump to line 201, because the condition on line 194 was never false if 'name' in notification_category_dict and notification_category_dict['name'] is not None: notification_category_name = notification_category_dict['name'] 196 ↛ 199 line 196 didn't jump to line 199, because the condition on line 196 was never false if NotificationCategory.is_name_unique(id=id, name=notification_category_name): notification_category.name = notification_category_name else: response = {'error': 'A category with the name {} already exists'.format(notification_category_name)} return response, HttpStatus.bad_request_400.value notification_category.update() return self.get(id) except SQLAlchemyError as e: orm.session.rollback() response = {"error": str(e)} return response, HttpStatus.bad_request_400.value def delete(self, id): notification_category = NotificationCategory.query.get_or_404(id) try: notification_category.delete(notification_category) response = make_response() return response, HttpStatus.no_content_204.value except SQLAlchemyError as e: orm.session.rollback() response = {"error": str(e)} return response, HttpStatus.unauthorized_401.value class NotificationCategoryListResource(AuthenticationRequiredResource): def get(self): notification_categories = NotificationCategory.query.all() dump_results = notification_category_schema.dump(notification_categories, many=True).data return dump_results def post(self): print("Processing") notification_category_dict = request.get_json() 229 ↛ 230 line 229 didn't jump to line 230, because the condition on line 229 was never true if not notification_category_dict: response = {'message': 'No input data provided'} return response, HttpStatus.bad_request_400.value errors = notification_category_schema.validate(notification_category_dict) 233 ↛ 234 line 233 didn't jump to line 234, because the condition on line 233 was never true if errors: return errors, HttpStatus.bad_request_400.value notification_category_name = notification_category_dict['name'] if not NotificationCategory.is_name_unique(id=0, name=notification_category_name): response = {'error': 'A notification category with the name {} already exists'.format(notification_category_name)} return response, HttpStatus.bad_request_400.value try: notification_category = NotificationCategory(notification_category_name) notification_category.add(notification_category) query = NotificationCategory.query.get(notification_category.id) dump_result = notification_category_schema.dump(query).data return dump_result, HttpStatus.created_201.value except SQLAlchemyError as e: print("Error") print(e) orm.session.rollback() response = {"error": str(e)} return response, HttpStatus.bad_request_400.value service.add_resource(NotificationCategoryListResource, '/notification_categories/') service.add_resource(NotificationCategoryResource, '/notification_categories/<int:id>') service.add_resource(NotificationListResource, '/notifications/') service.add_resource(NotificationResource, '/notifications/<int:id>') service.add_resource(UserListResource, '/users/') service.add_resource(UserResource, '/users/<int:id>')