Coverage for src/couchers/helpers/badges.py: 100%

1from sqlalchemy.orm import Session 

2from sqlalchemy.sql import delete 

3 

4from couchers.context import make_background_user_context 

5from couchers.models import UserBadge 

6from couchers.notifications.notify import notify 

7from couchers.proto import notification_data_pb2 

8from couchers.resources import get_badge_dict 

9 

10 

11def user_add_badge(session: Session, user_id: int, badge_id: str, do_notify: bool = True) -> None: 

12 badge = get_badge_dict()[badge_id] 

13 session.add(UserBadge(user_id=user_id, badge_id=badge_id)) 

14 session.flush() 

15 if do_notify: 

16 context = make_background_user_context(user_id=user_id) 

17 notify( 

18 session, 

19 user_id=user_id, 

20 topic_action="badge:add", 

21 data=notification_data_pb2.BadgeAdd( 

22 badge_id=badge["id"], 

23 badge_name=context.get_localized_string("badges", f"{badge_id}_name"), 

24 badge_description=context.get_localized_string("badges", f"{badge_id}_description"), 

25 ), 

26 ) 

27 session.commit() 

28 

29 

30def user_remove_badge(session: Session, user_id: int, badge_id: str) -> None: 

31 badge = get_badge_dict()[badge_id] 

32 session.execute(delete(UserBadge).where(UserBadge.user_id == user_id, UserBadge.badge_id == badge_id)) 

33 session.flush() 

34 context = make_background_user_context(user_id=user_id) 

35 notify( 

36 session, 

37 user_id=user_id, 

38 topic_action="badge:remove", 

39 data=notification_data_pb2.BadgeRemove( 

40 badge_id=badge["id"], 

41 badge_name=context.get_localized_string("badges", f"{badge_id}_name"), 

42 badge_description=context.get_localized_string("badges", f"{badge_id}_description"), 

43 ), 

44 ) 

45 session.commit()