Bug fix and added user model test

Author: kartikbehl99

.isort.cfg

@@ -2,4 +2,4 @@
 line_length = 88
 multi_line_output = 3
 include_trailing_comma = True
-known_third_party = cryptography,flask,sqlalchemy
+known_third_party = cryptography,flask,pytest,sqlalchemy

ssh_manager_backend/app/models/user.py

@@ -94,6 +94,7 @@ def create(
             self.session.add(user)
             self.session.commit()
         except SQLAlchemyError:
+            self.session.rollback()
             return False
 
         return True
@@ -124,3 +125,23 @@ def get_user(self, username: str) -> Union[None, User]:
         """
 
         return self.session.query(User).filter(User.username == username).first()
+
+    def destroy_user(self, username: str) -> bool:
+        """
+        Deletes a user.
+
+        :param username:
+        :return: Boolean value indicating success/failure.
+        """
+
+        try:
+            user: User = self.session.query(User).filter(
+                User.username == username
+            ).first()
+            self.session.delete(user)
+            self.session.commit()
+        except SQLAlchemyError:
+            self.session.rollback()
+            return False
+
+        return True

ssh_manager_backend/db/database.py

@@ -3,7 +3,7 @@
 from sqlalchemy.orm import scoped_session, sessionmaker
 
 DB_URI = "postgresql+psycopg2://postgres:ssh_manager@localhost/ssh_manager_dev"
-engine = create_engine(DB_URI, convert_unicode=True, echo=True)
+engine = create_engine(DB_URI, echo=True)
 db_session = scoped_session(
     sessionmaker(autocommit=False, autoflush=False, bind=engine)
 )

ssh_manager_backend/db/schema.py

@@ -1,4 +1,4 @@
-from sqlalchemy import ARRAY, Binary, Boolean, Column, ForeignKey, Integer, String
+from sqlalchemy import ARRAY, Boolean, Column, ForeignKey, Integer, LargeBinary, String
 from sqlalchemy.orm import relationship
 
 from ssh_manager_backend.db.database import Base
@@ -12,12 +12,12 @@ class User(Base):
     username = Column(String, unique=True)
     password = Column(String)
     admin = Column(Boolean)
-    encrypted_dek = Column(Binary, unique=True)
-    iv_for_dek = Column(Binary, unique=True)
-    salt_for_dek = Column(Binary, unique=True)
-    iv_for_kek = Column(Binary, unique=True)
-    salt_for_kek = Column(Binary, unique=True)
-    salt_for_password = Column(Binary, unique=True)
+    encrypted_dek = Column(LargeBinary, unique=True)
+    iv_for_dek = Column(LargeBinary, unique=True)
+    salt_for_dek = Column(LargeBinary, unique=True)
+    iv_for_kek = Column(LargeBinary, unique=True)
+    salt_for_kek = Column(LargeBinary, unique=True)
+    salt_for_password = Column(LargeBinary, unique=True)
     keys = relationship("Key", backref="users")
     access_control = relationship("AccessControl", backref="users")
 
@@ -34,7 +34,7 @@ class Key(Base):
 
     id = Column(Integer, primary_key=True)
     name = Column(String, unique=True)
-    encrypted_key = Column(Binary, unique=True)
+    encrypted_key = Column(LargeBinary, unique=True)
     key_hash = Column(String, unique=True)
     user_id = Column(Integer, ForeignKey("users.id"))
     user = relationship("User")

tests/models/acl_model_test.py

@@ -0,0 +1,5 @@
+import pytest
+
+
+class TestAccessControlModel:
+    pass

tests/models/user_model_test.py

@@ -0,0 +1,79 @@
+from ssh_manager_backend.app.models.user import UserModel
+from ssh_manager_backend.db import User
+
+
+class TestUserModel:
+    def test_create(self):
+        user = UserModel()
+        name: str = "test_user"
+        username: str = "test_username"
+        password: str = "test_password"
+        admin: bool = False
+        encrypted_dek: bytes = b"test_encrypted_dek"
+        iv_for_dek: bytes = b"test_iv_for_dek"
+        salt_for_dek: bytes = b"test_salt_for_dek"
+        iv_for_kek: bytes = b"test_iv_for_kek"
+        salt_for_kek: bytes = b"test_salt_for_kek"
+        salt_for_password: bytes = b"test_salt_for_password"
+
+        assert (
+            user.create(
+                name=name,
+                username=username,
+                password=password,
+                admin=admin,
+                encrypted_dek=encrypted_dek,
+                iv_for_dek=iv_for_dek,
+                salt_for_dek=salt_for_dek,
+                iv_for_kek=iv_for_kek,
+                salt_for_kek=salt_for_kek,
+                salt_for_password=salt_for_password,
+            )
+            is True
+        )
+
+        assert (
+            user.create(
+                name=name,
+                username=username,
+                password=password,
+                admin=admin,
+                encrypted_dek=encrypted_dek,
+                iv_for_dek=iv_for_dek,
+                salt_for_dek=salt_for_dek,
+                iv_for_kek=iv_for_kek,
+                salt_for_kek=salt_for_kek,
+                salt_for_password=salt_for_password,
+            )
+            is False
+        )
+
+    def test_admin_exist(self):
+        user = UserModel()
+        assert user.admin_exists() is False
+
+    def test_password_hash(self):
+        user = UserModel()
+        username: str = "test_username"
+        password: str = "test_password"
+        assert user.password_hash("non_existent_username") is None
+        assert user.password_hash(username) == password
+
+    def test_exists(self):
+        user = UserModel()
+        username: str = "test_username"
+        assert user.exists("non_existent_username") is False
+        assert user.exists(username) is True
+
+    def test_get_user(self):
+        user = UserModel()
+        username: str = "test_username"
+        assert user.get_user("non_existent_username") is None
+        assert isinstance(user.get_user(username), User) is True
+        print(User)
+
+    def test_destroy_user(self):
+        user = UserModel()
+        username: str = "test_username"
+        assert user.destroy_user("non_existent_username") is False
+        assert user.destroy_user(username) is True