Coverage for app/backend/src/tests/fixtures/api/requests.py: 100%

1from contextlib import contextmanager 

2from datetime import date, timedelta 

3from typing import cast 

4 

5import grpc 

6 

7from couchers.constants import HOST_REQUEST_MIN_LENGTH_UTF16 

8from couchers.proto import messages_pb2, requests_pb2, requests_pb2_grpc 

9from couchers.servicers.requests import Requests 

10from couchers.utils import date_to_api, today 

11from tests.fixtures.misc import Moderator 

12from tests.fixtures.sessions import FakeChannel 

13 

14 

15def _pad_request_text_to_valid_length(text: str) -> str: 

16 # Request lengths are measured in utf-16 code units to match the frontend. 

17 while len(text.encode("utf-16-le")) // 2 < HOST_REQUEST_MIN_LENGTH_UTF16: 

18 text += " padding" 

19 return text 

20 

21 

22# Neutral default text of valid length 

23REQUEST_TEXT_DEFAULT = _pad_request_text_to_valid_length("Dummy request text.") 

24 

25 

26@contextmanager 

27def _requests_session(token: str): 

28 channel = FakeChannel(token) 

29 requests_pb2_grpc.add_RequestsServicer_to_server(Requests(), cast(grpc.Server, channel)) 

30 yield requests_pb2_grpc.RequestsStub(cast(grpc.Channel, channel)) 

31 

32 

33def create_host_request( 

34 *, 

35 token: str, 

36 host_user_id: int, 

37 # Strings allow testing malformed dates. 

38 from_date: date | str | None = None, 

39 to_date: date | str | None = None, 

40 text: str | None = REQUEST_TEXT_DEFAULT, 

41 pad_text_to_valid_length: bool = False, 

42 public_trip_id: int | None = None, 

43 approve_by: Moderator | None = None, 

44) -> int: 

45 if text is not None and pad_text_to_valid_length: 

46 text = _pad_request_text_to_valid_length(text) 

47 if from_date is None: 

48 # Two days later is in the future no matter the host's timezone. 

49 from_date = today() + timedelta(days=2) 

50 if to_date is None: 

51 to_date = (date.fromisoformat(from_date) if isinstance(from_date, str) else from_date) + timedelta(days=1) 

52 

53 with _requests_session(token) as session: 

54 res: requests_pb2.CreateHostRequestRes = session.CreateHostRequest( 

55 requests_pb2.CreateHostRequestReq( 

56 host_user_id=host_user_id, 

57 from_date=from_date if isinstance(from_date, str) else date_to_api(from_date), 

58 to_date=to_date if isinstance(to_date, str) else date_to_api(to_date), 

59 text=text, 

60 public_trip_id=public_trip_id, 

61 ) 

62 ) 

63 

64 if approve_by is not None: 

65 approve_by.approve_host_request(res.host_request_id) 

66 

67 return res.host_request_id 

68 

69 

70def get_host_request(*, token: str, host_request_id: int) -> requests_pb2.HostRequest: 

71 with _requests_session(token) as session: 

72 host_request: requests_pb2.HostRequest = session.GetHostRequest( 

73 requests_pb2.GetHostRequestReq(host_request_id=host_request_id) 

74 ) 

75 return host_request 

76 

77 

78def respond_host_request( 

79 *, 

80 token: str, 

81 host_request_id: int, 

82 status: messages_pb2.HostRequestStatus.ValueType, 

83 text: str | None = None, 

84) -> None: 

85 with _requests_session(token) as session: 

86 session.RespondHostRequest( 

87 requests_pb2.RespondHostRequestReq(host_request_id=host_request_id, status=status, text=text) 

88 ) 

89 

90 

91def set_host_request_archive_status( 

92 *, token: str, host_request_id: int, is_archived: bool 

93) -> requests_pb2.SetHostRequestArchiveStatusRes: 

94 with _requests_session(token) as session: 

95 res: requests_pb2.SetHostRequestArchiveStatusRes = session.SetHostRequestArchiveStatus( 

96 requests_pb2.SetHostRequestArchiveStatusReq(host_request_id=host_request_id, is_archived=is_archived) 

97 ) 

98 return res 

99 

100 

101def list_host_requests( 

102 *, 

103 token: str, 

104 number: int | None = None, 

105 only_active: bool = False, 

106 only_sent: bool = False, 

107 only_received: bool = False, 

108 only_archived: bool | None = None, 

109 status_in: list[messages_pb2.HostRequestStatus.ValueType] | None = None, 

110 sort_by: requests_pb2.HostRequestSortBy.ValueType | None = None, 

111 page_token: str | None = None, 

112) -> requests_pb2.ListHostRequestsRes: 

113 with _requests_session(token) as session: 

114 res: requests_pb2.ListHostRequestsRes = session.ListHostRequests( 

115 requests_pb2.ListHostRequestsReq( 

116 number=number, 

117 only_active=only_active, 

118 only_sent=only_sent, 

119 only_received=only_received, 

120 only_archived=only_archived, 

121 status_in=status_in, 

122 sort_by=sort_by, 

123 page_token=page_token, 

124 ) 

125 ) 

126 return res 

127 

128 

129def list_stays( 

130 *, 

131 token: str, 

132 role: requests_pb2.StayRole.ValueType, 

133 page_size: int | None = None, 

134 page_token: str | None = None, 

135) -> requests_pb2.ListStaysRes: 

136 with _requests_session(token) as session: 

137 res: requests_pb2.ListStaysRes = session.ListStays( 

138 requests_pb2.ListStaysReq(role=role, page_size=page_size, page_token=page_token) 

139 ) 

140 return res 

141 

142 

143def get_host_request_messages( 

144 *, token: str, host_request_id: int, last_message_id: int | None = None, number: int | None = None 

145) -> requests_pb2.GetHostRequestMessagesRes: 

146 with _requests_session(token) as session: 

147 res: requests_pb2.GetHostRequestMessagesRes = session.GetHostRequestMessages( 

148 requests_pb2.GetHostRequestMessagesReq( 

149 host_request_id=host_request_id, last_message_id=last_message_id, number=number 

150 ) 

151 ) 

152 return res 

153 

154 

155def send_host_request_message(*, token: str, host_request_id: int, text: str = "Dummy message.") -> None: 

156 with _requests_session(token) as session: 

157 session.SendHostRequestMessage( 

158 requests_pb2.SendHostRequestMessageReq(host_request_id=host_request_id, text=text) 

159 ) 

160 

161 

162def get_host_request_updates( 

163 *, 

164 token: str, 

165 newest_message_id: int, 

166 number: int | None = None, 

167 only_sent: bool = False, 

168 only_received: bool = False, 

169) -> requests_pb2.GetHostRequestUpdatesRes: 

170 with _requests_session(token) as session: 

171 res: requests_pb2.GetHostRequestUpdatesRes = session.GetHostRequestUpdates( 

172 requests_pb2.GetHostRequestUpdatesReq( 

173 newest_message_id=newest_message_id, 

174 number=number, 

175 only_sent=only_sent, 

176 only_received=only_received, 

177 ) 

178 ) 

179 return res 

180 

181 

182def mark_last_seen_host_request(*, token: str, host_request_id: int, last_seen_message_id: int) -> None: 

183 with _requests_session(token) as session: 

184 session.MarkLastSeenHostRequest( 

185 requests_pb2.MarkLastSeenHostRequestReq( 

186 host_request_id=host_request_id, last_seen_message_id=last_seen_message_id 

187 ) 

188 ) 

189 

190 

191def get_response_rate(*, token: str, user_id: int) -> requests_pb2.GetResponseRateRes: 

192 with _requests_session(token) as session: 

193 res: requests_pb2.GetResponseRateRes = session.GetResponseRate(requests_pb2.GetResponseRateReq(user_id=user_id)) 

194 return res 

195 

196 

197def send_host_request_feedback( 

198 *, 

199 token: str, 

200 host_request_id: int, 

201 host_request_quality: requests_pb2.HostRequestQuality.ValueType = requests_pb2.HOST_REQUEST_QUALITY_UNSPECIFIED, 

202 decline_reason: str | None = None, 

203) -> None: 

204 with _requests_session(token) as session: 

205 session.SendHostRequestFeedback( 

206 requests_pb2.SendHostRequestFeedbackReq( 

207 host_request_id=host_request_id, 

208 host_request_quality=host_request_quality, 

209 decline_reason=decline_reason, 

210 ) 

211 )