Coverage for app / backend / src / tests / test_sanitized_bytes.py: 100%

1from google.protobuf import empty_pb2 

2 

3from couchers.interceptors import _sanitized_bytes 

4from couchers.proto import api_pb2, auth_pb2, conversations_pb2 

5 

6 

7class TestSanitizedBytes: 

8 """Test suite for _sanitized_bytes function.""" 

9 

10 def test_none_input(self): 

11 """Test that None input returns None.""" 

12 result = _sanitized_bytes(None) 

13 assert result is None 

14 

15 def test_empty_message(self): 

16 """Test that an empty message is serialized correctly.""" 

17 proto = empty_pb2.Empty() 

18 result = _sanitized_bytes(proto) 

19 

20 assert isinstance(result, bytes) 

21 # Verify we can deserialize it back 

22 deserialized = empty_pb2.Empty.FromString(result) 

23 assert deserialized == proto 

24 

25 def test_message_with_no_sensitive_fields(self): 

26 """Test that messages without sensitive fields are not modified.""" 

27 # AuthRes has no sensitive fields 

28 proto = auth_pb2.AuthRes(user_id=12345, jailed=False) 

29 result = _sanitized_bytes(proto) 

30 

31 deserialized = auth_pb2.AuthRes.FromString(result) 

32 assert deserialized.user_id == 12345 

33 assert deserialized.jailed is False 

34 

35 def test_message_with_sensitive_field(self): 

36 """Test that sensitive fields are cleared from messages.""" 

37 # AuthReq has a sensitive password field 

38 proto = auth_pb2.AuthReq(user="testuser", password="supersecret123", remember_device=True) 

39 result = _sanitized_bytes(proto) 

40 

41 deserialized = auth_pb2.AuthReq.FromString(result) 

42 

43 # Non-sensitive fields should remain 

44 assert deserialized.user == "testuser" 

45 assert deserialized.remember_device is True 

46 

47 # Sensitive field should be cleared 

48 assert deserialized.password == "" 

49 

50 def test_message_with_nested_message_non_repeated(self): 

51 """Test that nested messages are recursively sanitized (non-repeated case).""" 

52 # SignupFlowReq contains a nested SignupAccount message 

53 proto = auth_pb2.SignupFlowReq( 

54 flow_token="token123", 

55 account=auth_pb2.SignupAccount(username="nesteduser", password="nestedsecret", city="Boston"), 

56 ) 

57 result = _sanitized_bytes(proto) 

58 

59 deserialized = auth_pb2.SignupFlowReq.FromString(result) 

60 

61 # Top-level fields should remain 

62 assert deserialized.flow_token == "token123" 

63 

64 # Nested message non-sensitive fields should remain 

65 assert deserialized.account.username == "nesteduser" 

66 assert deserialized.account.city == "Boston" 

67 

68 # Nested message sensitive field should be cleared 

69 assert deserialized.account.password == "" 

70 

71 def test_message_with_empty_nested_message(self): 

72 """Test that empty nested message fields don't cause errors (continue branch).""" 

73 # Create a message where nested message field is default/empty 

74 # SignupFlowRes with auth_res not set (optional field) 

75 proto = auth_pb2.SignupFlowRes( 

76 flow_token="token456", 

77 need_basic=True, 

78 need_account=False, 

79 # auth_res is not set 

80 ) 

81 result = _sanitized_bytes(proto) 

82 

83 deserialized = auth_pb2.SignupFlowRes.FromString(result) 

84 assert deserialized.flow_token == "token456" 

85 assert deserialized.need_basic is True 

86 assert deserialized.need_account is False 

87 # auth_res should still be empty/default 

88 assert not deserialized.HasField("auth_res") 

89 

90 def test_message_with_empty_repeated_field(self): 

91 """ 

92 Test that empty repeated message fields trigger the continue branch. 

93 

94 This covers line 171-172 where submessage evaluates to False (empty list). 

95 """ 

96 # Create a message with a repeated field but don't add any items 

97 proto = conversations_pb2.GetGroupChatMessagesRes( 

98 last_message_id=999, 

99 no_more=True, 

100 # messages field is empty (default empty repeated field) 

101 ) 

102 result = _sanitized_bytes(proto) 

103 

104 deserialized = conversations_pb2.GetGroupChatMessagesRes.FromString(result) 

105 assert deserialized.last_message_id == 999 

106 assert deserialized.no_more is True 

107 # Repeated field should be empty 

108 assert len(deserialized.messages) == 0 

109 

110 def test_message_with_repeated_messages(self): 

111 """Test that repeated message fields are recursively sanitized.""" 

112 # Create a message with repeated nested messages 

113 # Using GetGroupChatMessagesRes which has repeated Message fields 

114 proto = conversations_pb2.GetGroupChatMessagesRes(last_message_id=100, no_more=False) 

115 

116 # Add messages to the repeated field 

117 msg1 = proto.messages.add() 

118 msg1.message_id = 1 

119 msg1.author_user_id = 123 

120 msg1.text.text = "Hello" 

121 

122 msg2 = proto.messages.add() 

123 msg2.message_id = 2 

124 msg2.author_user_id = 456 

125 msg2.text.text = "World" 

126 

127 result = _sanitized_bytes(proto) 

128 

129 deserialized = conversations_pb2.GetGroupChatMessagesRes.FromString(result) 

130 

131 # Check that repeated messages are preserved 

132 assert len(deserialized.messages) == 2 

133 assert deserialized.messages[0].message_id == 1 

134 assert deserialized.messages[0].text.text == "Hello" 

135 assert deserialized.messages[0].author_user_id == 123 

136 assert deserialized.messages[1].message_id == 2 

137 assert deserialized.messages[1].text.text == "World" 

138 assert deserialized.messages[1].author_user_id == 456 

139 

140 def test_deeply_nested_messages(self): 

141 """Test that deeply nested messages are recursively sanitized.""" 

142 # SignupFlowRes contains AuthRes which is nested 

143 proto = auth_pb2.SignupFlowRes( 

144 flow_token="deep_token", 

145 auth_res=auth_pb2.AuthRes(user_id=789, jailed=False), 

146 need_basic=False, 

147 need_account=True, 

148 ) 

149 result = _sanitized_bytes(proto) 

150 

151 deserialized = auth_pb2.SignupFlowRes.FromString(result) 

152 

153 # All nested data should be preserved (no sensitive fields in this structure) 

154 assert deserialized.flow_token == "deep_token" 

155 assert deserialized.auth_res.user_id == 789 

156 assert deserialized.auth_res.jailed is False 

157 assert deserialized.need_basic is False 

158 assert deserialized.need_account is True 

159 

160 def test_multiple_nested_messages_with_sensitive_fields(self): 

161 """ 

162 Test sanitization when a message has multiple nested messages with sensitive fields. 

163 

164 This ensures the function properly handles multiple different nested message fields. 

165 """ 

166 # Create a SignupFlowReq with multiple nested messages 

167 proto = auth_pb2.SignupFlowReq( 

168 flow_token="repeat_token", 

169 basic=auth_pb2.SignupBasic(name="Test User", email="test@example.com"), 

170 account=auth_pb2.SignupAccount(username="testuser", password="shouldberemoved"), 

171 ) 

172 result = _sanitized_bytes(proto) 

173 

174 deserialized = auth_pb2.SignupFlowReq.FromString(result) 

175 

176 # Check basic nested message 

177 assert deserialized.basic.name == "Test User" 

178 assert deserialized.basic.email == "test@example.com" 

179 

180 # Check account nested message with sensitive field 

181 assert deserialized.account.username == "testuser" 

182 assert deserialized.account.password == "" 

183 

184 def test_message_preserves_original(self): 

185 """Test that the original message is not modified (deepcopy is used).""" 

186 original = auth_pb2.AuthReq(user="original_user", password="original_password") 

187 

188 # Store original values 

189 original_user = original.user 

190 original_password = original.password 

191 

192 # Call _sanitized_bytes 

193 result = _sanitized_bytes(original) 

194 

195 # Original should not be modified 

196 assert original.user == original_user 

197 assert original.password == original_password 

198 

199 # But the result should have password cleared 

200 deserialized = auth_pb2.AuthReq.FromString(result) 

201 assert deserialized.user == original_user 

202 assert deserialized.password == "" 

203 

204 def test_message_with_non_message_type_fields(self): 

205 """Test messages with primitive fields (no message_type).""" 

206 # Create a message with only primitive types 

207 proto = auth_pb2.AuthRes(user_id=12345, jailed=True) 

208 result = _sanitized_bytes(proto) 

209 

210 deserialized = auth_pb2.AuthRes.FromString(result) 

211 assert deserialized.user_id == 12345 

212 assert deserialized.jailed is True 

213 

214 def test_complex_nested_structure(self): 

215 """Test a complex nested structure to ensure all branches are covered.""" 

216 # Create a complex nested message 

217 proto = auth_pb2.SignupFlowReq( 

218 flow_token="complex_token", 

219 basic=auth_pb2.SignupBasic(name="Complex User", email="complex@example.com", invite_code="INVITE123"), 

220 account=auth_pb2.SignupAccount( 

221 username="complexuser", 

222 password="complexsecret", 

223 city="Seattle", 

224 lat=47.6062, 

225 lng=-122.3321, 

226 birthdate="1985-05-15", 

227 gender="other", 

228 hosting_status=api_pb2.HOSTING_STATUS_CAN_HOST, 

229 accept_tos=True, 

230 ), 

231 feedback=auth_pb2.ContributorForm(ideas="Great platform!", contribute=auth_pb2.CONTRIBUTE_OPTION_YES), 

232 email_token="email_verification_token", 

233 ) 

234 

235 result = _sanitized_bytes(proto) 

236 

237 deserialized = auth_pb2.SignupFlowReq.FromString(result) 

238 

239 # Verify all non-sensitive fields are preserved 

240 assert deserialized.flow_token == "complex_token" 

241 assert deserialized.basic.name == "Complex User" 

242 assert deserialized.basic.email == "complex@example.com" 

243 assert deserialized.basic.invite_code == "INVITE123" 

244 assert deserialized.account.username == "complexuser" 

245 assert deserialized.account.city == "Seattle" 

246 assert deserialized.account.lat == 47.6062 

247 assert deserialized.account.birthdate == "1985-05-15" 

248 assert deserialized.feedback.ideas == "Great platform!" 

249 assert deserialized.email_token == "email_verification_token" 

250 

251 # Verify sensitive field is cleared 

252 assert deserialized.account.password == ""