Coverage for app/backend/src/couchers/helpers/messages.py: 100%

1""" 

2Message serialization shared by the conversations and requests APIs: both surface messages from the 

3same table, so they must render every message type identically. 

4""" 

5 

6from couchers.models import HostRequestStatus, Message, MessageType 

7from couchers.proto import messages_pb2 

8from couchers.utils import Timestamp_from_datetime 

9 

10hostrequeststatus2api = { 

11 HostRequestStatus.pending: messages_pb2.HOST_REQUEST_STATUS_PENDING, 

12 HostRequestStatus.accepted: messages_pb2.HOST_REQUEST_STATUS_ACCEPTED, 

13 HostRequestStatus.rejected: messages_pb2.HOST_REQUEST_STATUS_REJECTED, 

14 HostRequestStatus.confirmed: messages_pb2.HOST_REQUEST_STATUS_CONFIRMED, 

15 HostRequestStatus.cancelled: messages_pb2.HOST_REQUEST_STATUS_CANCELLED, 

16} 

17 

18api2hostrequeststatus = { 

19 messages_pb2.HOST_REQUEST_STATUS_PENDING: HostRequestStatus.pending, 

20 messages_pb2.HOST_REQUEST_STATUS_ACCEPTED: HostRequestStatus.accepted, 

21 messages_pb2.HOST_REQUEST_STATUS_REJECTED: HostRequestStatus.rejected, 

22 messages_pb2.HOST_REQUEST_STATUS_CONFIRMED: HostRequestStatus.confirmed, 

23 messages_pb2.HOST_REQUEST_STATUS_CANCELLED: HostRequestStatus.cancelled, 

24} 

25 

26 

27def message_to_pb(message: Message) -> messages_pb2.Message: 

28 """ 

29 Turns the given message to a protocol buffer 

30 """ 

31 if message.is_normal_message: 

32 return messages_pb2.Message( 

33 message_id=message.id, 

34 author_user_id=message.author_id, 

35 time=Timestamp_from_datetime(message.time), 

36 text=messages_pb2.MessageContentText(text=message.text), 

37 ) 

38 else: 

39 return messages_pb2.Message( 

40 message_id=message.id, 

41 author_user_id=message.author_id, 

42 time=Timestamp_from_datetime(message.time), 

43 chat_created=( 

44 messages_pb2.MessageContentChatCreated() if message.message_type == MessageType.chat_created else None 

45 ), 

46 chat_edited=( 

47 messages_pb2.MessageContentChatEdited() if message.message_type == MessageType.chat_edited else None 

48 ), 

49 user_invited=( 

50 messages_pb2.MessageContentUserInvited(target_user_id=message.target_id) 

51 if message.message_type == MessageType.user_invited 

52 else None 

53 ), 

54 user_left=( 

55 messages_pb2.MessageContentUserLeft() if message.message_type == MessageType.user_left else None 

56 ), 

57 user_made_admin=( 

58 messages_pb2.MessageContentUserMadeAdmin(target_user_id=message.target_id) 

59 if message.message_type == MessageType.user_made_admin 

60 else None 

61 ), 

62 user_removed_admin=( 

63 messages_pb2.MessageContentUserRemovedAdmin(target_user_id=message.target_id) 

64 if message.message_type == MessageType.user_removed_admin 

65 else None 

66 ), 

67 group_chat_user_removed=( 

68 messages_pb2.MessageContentUserRemoved(target_user_id=message.target_id) 

69 if message.message_type == MessageType.user_removed 

70 else None 

71 ), 

72 host_request_status_changed=( 

73 messages_pb2.MessageContentHostRequestStatusChanged( 

74 status=hostrequeststatus2api[message.host_request_status_target] # type: ignore[index] 

75 ) 

76 if message.message_type == MessageType.host_request_status_changed 

77 else None 

78 ), 

79 )