Initial work to remove role column from UserModel (#358)

* Initial work to remove role column from UserModel

We don't really have different roles so they are being removed. We have
different types of users which have been implemented using STI.

This is the first phase to remove roles from the backend. We will still
have a role per se as far as the frontend code is concerned.

The next phase will be to update the frontend to move off of integers as
their Role, and use the typed strings. Along with changing any hard
coded dependencies on Roles. The final phase will be back in the backend
to remove the final remnants of Roles.

Doing it in a 3 phase step as to not break anything for other
developers :-)

* Refactor UserLogin

* We still need to assign a role on update

Still need to assign a role until the frontend is updated not to use
roles.

* User singular name for user_type class

Author: NickSchimek

app.py

@@ -32,16 +32,27 @@ def create_app(env):
     # initialize Mail
     app.mail = Mail(app)
 
-    # check the user role in the JSON Web Token (JWT)
+    @app.jwt.user_identity_loader
+    def user_identity_lookup(user):
+        return user.id
+
+    @app.jwt.user_lookup_loader
+    def user_lookup_callback(_jwt_header, jwt_data):
+        user = UserModel.query.filter_by(
+            id=jwt_data["sub"], archived=False
+        ).one_or_none()
+        if user and user.type == "user":
+            return None
+        else:
+            return user
+
     @app.jwt.additional_claims_loader
-    def role_loader(identity):
-        user = UserModel.find(identity)
+    def role_loader(user):
         return {
             "email": user.email,
             "phone": user.phone,
             "firstName": user.firstName,
             "lastName": user.lastName,
-            "role": user.role.value,
         }
 
     # checking if the token's jti (jwt id) is in the set of revoked tokens

models/user.py

@@ -33,6 +33,20 @@ def has_role(cls, role):
         return role in role_values
 
 
+class UserType(Enum):
+    ADMIN = "admin"
+    STAFF = "staff"
+    PROPERTY_MANAGER = "property_manager"
+
+    @classmethod
+    def get(cls, value, default=None):
+        return cls._values().get(value, default)
+
+    @classmethod
+    def _values(cls):
+        return {member.value: member.value for member in cls.__members__.values()}
+
+
 class UserModel(BaseModel):
     __tablename__ = "users"
 
@@ -117,6 +131,18 @@ def json(self, refresh_token=False):
     def serialize(self):
         return {}
 
+    @staticmethod
+    def authenticate(user, password):
+        if user and (user.archived or user.type == "user" or user.role is None):
+            return {"message": "Invalid user"}, 403
+        elif user and user.check_pw(password):
+            access_token = create_access_token(identity=user, fresh=True)
+            refresh_token = create_refresh_token(user)
+            user.update_last_active()
+            return {"access_token": access_token, "refresh_token": refresh_token}, 200
+        else:
+            return {"message": "Invalid credentials"}, 401
+
     @classmethod
     def find_by_email(cls, email):
         return cls.query.filter_by(email=email).first()
@@ -149,11 +175,17 @@ def full_name(self):
     def is_admin(self):
         return False
 
+    def has_staff_privs(self):
+        return False
+
+    def has_pm_privs(self):
+        return False
+
     def _make_token(self, refresh):
         if refresh:
             return {
-                "access_token": create_access_token(identity=self.id, fresh=True),
-                "refresh_token": create_refresh_token(self.id),
+                "access_token": create_access_token(identity=self, fresh=True),
+                "refresh_token": create_refresh_token(self),
             }
         else:
             return {}

models/users/admin.py

@@ -6,3 +6,9 @@ class Admin(UserModel):
 
     def is_admin(self):
         return True
+
+    def has_staff_privs(self):
+        return True
+
+    def has_pm_privs(self):
+        return True

models/users/property_manager.py

@@ -14,5 +14,8 @@ class PropertyManager(UserModel):
         viewonly=True,
     )
 
+    def has_pm_privs(self):
+        return True
+
     def serialize(self):
         return {"properties": self.properties.json(include_managers=False)}

models/users/staff.py

@@ -13,3 +13,9 @@ class Staff(UserModel):
         collection_class=NobiruList,
         viewonly=True,
     )
+
+    def has_staff_privs(self):
+        return True
+
+    def has_pm_privs(self):
+        return True

resources/user.py

@@ -2,12 +2,11 @@
 from flask import request
 from schemas import UserRegisterSchema, UserSchema
 from utils.authorizations import admin_required, pm_level_required
-from models.user import UserModel, RoleEnum
+from models.user import UserModel, RoleEnum, UserType
 from flask_jwt_extended import (
     create_access_token,
-    create_refresh_token,
-    get_jwt_identity,
     jwt_required,
+    current_user,
 )
 
 
@@ -25,18 +24,24 @@ def get(self, id):
 
     @pm_level_required
     def patch(self, id):
-        current_user = UserModel.find(get_jwt_identity())
         user = UserModel.find(id)
         role = request.json.get("role")
+        user_type = UserType.get(request.json.get("type"))
         password = request.json.get("new_password")
+        role_change = role or user_type
 
-        if not self._authorized(current_user=current_user, user=user, role_change=role):
+        if not self._authorized(
+            current_user=current_user, user=user, role_change=role_change
+        ):
             return {"message": "Not Authorized"}, 403
 
         if role:
             user.role = RoleEnum(role)
             user.type = RoleEnum(role).name.lower()
 
+        if user_type:
+            user.type = user_type
+
         if password:
             if password != request.json.get("confirm_password"):
                 return {"message": "Unconfirmed password"}, 422
@@ -59,7 +64,7 @@ def _authorized(self, current_user, user, role_change):
     def delete(self, id):
         user = UserModel.find(id)
 
-        if user.id == get_jwt_identity():
+        if user == current_user:
             return {"message": "Cannot delete self"}, 400
 
         user.delete_from_db()
@@ -69,7 +74,7 @@ def delete(self, id):
 class ArchiveUser(Resource):
     @admin_required
     def post(self, user_id):
-        if user_id == get_jwt_identity():
+        if user_id == current_user.id:
             return {"message": "Cannot archive self"}, 400
 
         user = UserModel.find(user_id)
@@ -83,18 +88,10 @@ def post(self, user_id):
 
 class UserLogin(Resource):
     def post(self):
-        user = UserModel.find_by_email(request.json.get("email", ""))
-
-        if user and (user.archived or user.role is None):
-            return {"message": "Invalid user"}, 403
-
-        if user and user.check_pw(request.json.get("password", "")):
-            access_token = create_access_token(identity=user.id, fresh=True)
-            refresh_token = create_refresh_token(user.id)
-            user.update_last_active()
-            return {"access_token": access_token, "refresh_token": refresh_token}, 200
-
-        return {"message": "Invalid credentials"}, 401
+        return UserModel.authenticate(
+            UserModel.find_by_email(request.json.get("email", "")),
+            request.json.get("password", ""),
+        )
 
 
 # This endpoint allows the app to use a refresh token to get a new access token
@@ -107,19 +104,22 @@ class UserAccessRefresh(Resource):
     # to make a new access token for this identity.
     @jwt_required(refresh=True)
     def post(self):
-        current_user = get_jwt_identity()
         ret = {"access_token": create_access_token(identity=current_user)}
         return ret, 200
 
 
 class Users(Resource):
     @admin_required
     def get(self):
-        if RoleEnum.has_role(int(request.args["r"])):
+        user_type = UserType.get(request.args.get("type"))
+        # TODO: This is temporary. Switching to type from role lookup.
+        if RoleEnum.has_role(int(request.args.get("r", 99))):
             return {
                 "users": UserModel.query.filter_by(
                     role=RoleEnum(int(request.args["r"]))
                 ).json()
             }
-        else:
-            return {"message": "Invalid role"}, 400
+        elif user_type:
+            return {"users": UserModel.query.filter_by(type=user_type).json()}
+
+        return {"message": "Invalid role"}, 400

schemas/property_assignment.py

@@ -1,6 +1,7 @@
 from ma import ma
 from models.property_assignment import PropertyAssignment
-from models.user import UserModel, RoleEnum
+from models.user import UserModel
+from models.users.property_manager import PropertyManager
 from models.property import PropertyModel
 from marshmallow import fields, validates, ValidationError
 
@@ -26,6 +27,5 @@ def validate_is_valid_property(self, value):
 
     @validates("manager_id")
     def validate_role_property_manager(self, value):
-        user = UserModel.query.get(value)
-        if user.role != RoleEnum.PROPERTY_MANAGER:
+        if not PropertyManager.query.get(value):
             raise ValidationError("User is not a property manager")

schemas/staff_tenants.py

@@ -1,5 +1,5 @@
 from models.tenant import TenantModel
-from models.user import UserModel, RoleEnum
+from models.users.staff import Staff
 from marshmallow import Schema, fields, validates, ValidationError
 
 
@@ -19,9 +19,5 @@ def validate_tenant(self, value):
 
     @validates("staff")
     def validate_staff(self, value):
-        staff_query = UserModel.query.filter(
-            UserModel.id.in_(value), UserModel.role == RoleEnum.STAFF
-        )
-
-        if not len(staff_query.all()) == len(value):
+        if not len(Staff.query.filter(Staff.id.in_(value)).all()) == len(value):
             raise ValidationError(f"{value} contains invalid staff IDs")

schemas/user.py

@@ -33,3 +33,7 @@ class UserRegisterSchema(UserSchema):
     @validates("role")
     def user_cannot_register_a_role(self, _):
         raise ValidationError("Role is not allowed")
+
+    @validates("type")
+    def user_cannot_register_a_type(self, _):
+        raise ValidationError("Type is not allowed")

shelltools.py

@@ -1,5 +1,5 @@
 from db import db
-from models.user import UserModel, RoleEnum
+from models.user import UserModel, UserType
 from models.users.admin import Admin
 from models.users.property_manager import PropertyManager
 from models.users.staff import Staff

tests/authorizations/test_user_authentication.py

@@ -188,8 +188,8 @@ def test_jwt_refresh(self, header, create_user):
         """
 
         user = create_user()
-        original_access_token = create_access_token(identity=user.id, fresh=True)
-        original_refresh_token = create_refresh_token(user.id)
+        original_access_token = create_access_token(identity=user, fresh=True)
+        original_refresh_token = create_refresh_token(user)
 
         newPhone = "555-555-5555"
 

tests/integration/test_users.py

@@ -89,6 +89,19 @@ def test_get(self, create_admin_user, header):
         )
         assert is_valid(unknown_user_response, 400)
 
+    def test_get_with_type(self, create_join_staff, valid_header):
+        staff = create_join_staff()
+        response = self.client.get("/api/user?type=staff", headers=valid_header)
+
+        assert response.status_code == 200
+        assert response.json == {"users": [staff.json()]}
+
+    def test_get_with_invalid_type(self, create_join_staff, valid_header):
+        response = self.client.get("/api/user?type=big_honcho", headers=valid_header)
+
+        assert response.status_code == 400
+        assert response.json == {"message": "Invalid role"}
+
 
 @pytest.mark.usefixtures("client_class", "empty_test_db")
 class TestArchiveUser:

tests/schemas/test_user_schema.py

@@ -1,5 +1,5 @@
 from schemas.user import UserSchema, UserRegisterSchema
-from models.user import RoleEnum
+from models.user import RoleEnum, UserType
 
 
 def user_register_valid_payload(user):
@@ -74,6 +74,14 @@ def test_presence_of_role_key_is_not_valid(self):
         assert "role" in validation_errors
         assert validation_errors["role"] == ["Role is not allowed"]
 
+    def test_presence_of_type_is_not_valid(self):
+        payload = {"type": UserType.ADMIN.value}
+
+        validation_errors = UserRegisterSchema().validate(payload)
+
+        assert "type" in validation_errors
+        assert validation_errors["type"] == ["Type is not allowed"]
+
     def test_password_is_required(self):
         validation_errors = UserRegisterSchema().validate({})
 

tests/unit/test_user.py

@@ -2,7 +2,7 @@
 import jwt
 import time
 from unittest.mock import patch
-from models.user import UserModel, RoleEnum
+from models.user import UserModel
 from schemas.user import UserSchema
 from freezegun import freeze_time
 from tests.unit.base_interface_test import BaseInterfaceTest
@@ -73,7 +73,7 @@ class TestFixtures:
     def test_create_admin_user(self, create_admin_user):
         admin = create_admin_user()
         assert admin
-        assert admin.role == RoleEnum.ADMIN
+        assert admin.type == "admin"
 
         def test_multiple_admins_can_be_created():
             return create_admin_user()
@@ -83,7 +83,7 @@ def test_multiple_admins_can_be_created():
     def test_create_join_staff(self, create_join_staff):
         staff = create_join_staff()
         assert staff
-        assert staff.role == RoleEnum.STAFF
+        assert staff.type == "staff"
 
         def test_multiple_join_staff_can_be_created():
             return create_join_staff()
@@ -93,7 +93,7 @@ def test_multiple_join_staff_can_be_created():
     def test_create_property_manager(self, create_property_manager):
         pm = create_property_manager()
         assert pm
-        assert pm.role == RoleEnum.PROPERTY_MANAGER
+        assert pm.type == "property_manager"
 
         def test_multiple_pms_can_be_created():
             return create_property_manager()
@@ -102,7 +102,7 @@ def test_multiple_pms_can_be_created():
 
     def test_create_unauthorized_user(self, create_unauthorized_user):
         user = create_unauthorized_user()
-        assert user.role is None
+        assert user.type == "user"
 
         def test_multiple_users_can_be_created():
             return create_unauthorized_user()