Coverage for /usr/lib/python3.10/site-packages/hyd/backend/user/service.py: 65%

1from sqlalchemy.exc import IntegrityError 

2from sqlalchemy.orm import Session 

3 

4from hyd.backend.exc import NameStrError, UnknownUserError 

5from hyd.backend.security import hash_password 

6from hyd.backend.user.models import UserEntry 

7from hyd.backend.util.models import PrimaryKey 

8 

9 

10def create_user(*, username: str, password: str, is_admin: bool, db: Session) -> UserEntry: 

11 hashed_password: bytes = hash_password(password=password) 

12 

13 user_entry = UserEntry(username=username, hashed_password=hashed_password, is_admin=is_admin) 

14 db.add(user_entry) 

15 

16 try: 

17 db.commit() 

18 except IntegrityError: 

19 raise NameStrError 

20 

21 return user_entry 

22 

23 

24def read_user(*, user_id: PrimaryKey, db: Session) -> UserEntry: 

25 user_entry: UserEntry | None = db.query(UserEntry).get(user_id) 

26 

27 if user_entry is None: 

28 raise UnknownUserError 

29 

30 return user_entry 

31 

32 

33def read_users_by_username(*, username: str, db: Session) -> UserEntry: 

34 try: 

35 user_entry: UserEntry = db.query(UserEntry).filter(UserEntry.username == username).all()[0] 

36 except IndexError: 

37 raise UnknownUserError 

38 

39 return user_entry 

40 

41 

42def read_users(*, db: Session) -> list[UserEntry]: 

43 return db.query(UserEntry).all() 

44 

45 

46def enable_user_by_ref(*, user_entry: UserEntry, db: Session) -> None: 

47 user_entry.is_disabled = False 

48 db.commit() 

49 

50 

51def disable_user_by_ref(*, user_entry: UserEntry, db: Session) -> None: 

52 user_entry.is_disabled = True 

53 db.commit() 

54 

55 

56def update_user_pw_by_ref(*, user_entry: UserEntry, new_password: str, db: Session) -> None: 

57 user_entry.hashed_password = hash_password(password=new_password) 

58 db.commit()