Coverage for src/tests/test_verification.py: 100%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

156 statements  

1from unittest.mock import Mock, patch 

2 

3import grpc 

4import pytest 

5from google.protobuf import empty_pb2 

6 

7import couchers.phone.sms 

8from couchers.config import config 

9from couchers.crypto import random_hex 

10from couchers.db import session_scope 

11from couchers.models import SMS, User 

12from couchers.sql import couchers_select as select 

13from couchers.utils import now 

14from proto import account_pb2, api_pb2 

15from tests.test_fixtures import account_session, api_session, db, generate_user, testconfig # noqa 

16 

17 

18@pytest.fixture(autouse=True) 

19def _(testconfig): 

20 pass 

21 

22 

23def test_ChangePhone(db, monkeypatch): 

24 user, token = generate_user() 

25 user_id = user.id 

26 

27 with account_session(token) as account: 

28 res = account.GetAccountInfo(empty_pb2.Empty()) 

29 assert res.phone == "" 

30 

31 monkeypatch.setattr(couchers.phone.sms, "send_sms", pytest.fail) 

32 

33 # Try with a too long number 

34 with pytest.raises(grpc.RpcError) as e: 

35 account.ChangePhone(account_pb2.ChangePhoneReq(phone="+4670174060666666")) 

36 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

37 

38 # try to see if one digit too much is caught before attempting to send sms 

39 with pytest.raises(grpc.RpcError) as e: 

40 account.ChangePhone(account_pb2.ChangePhoneReq(phone="+467017406066")) 

41 assert e.value.code() == grpc.StatusCode.UNIMPLEMENTED 

42 

43 # Test with operator not supported by SMS backend 

44 def deny_operator(phone, message): 

45 assert phone == "+46701740605" 

46 return "unsupported operator" 

47 

48 monkeypatch.setattr(couchers.phone.sms, "send_sms", deny_operator) 

49 

50 with pytest.raises(grpc.RpcError) as e: 

51 account.ChangePhone(account_pb2.ChangePhoneReq(phone="+46701740605")) 

52 assert e.value.code() == grpc.StatusCode.UNIMPLEMENTED 

53 

54 # Test with successfully sent SMS 

55 def succeed(phone, message): 

56 assert phone == "+46701740605" 

57 return "success" 

58 

59 monkeypatch.setattr(couchers.phone.sms, "send_sms", succeed) 

60 

61 account.ChangePhone(account_pb2.ChangePhoneReq(phone="+46701740605")) 

62 

63 with session_scope() as session: 

64 user = session.execute(select(User).where(User.id == user_id)).scalar_one() 

65 assert user.phone == "+46701740605" 

66 assert len(user.phone_verification_token) == 6 

67 

68 # Phone number should show up but not be verified in your profile settings 

69 res = account.GetAccountInfo(empty_pb2.Empty()) 

70 assert res.phone == "+46701740605" 

71 assert not res.phone_verified 

72 

73 # Remove phone number 

74 account.ChangePhone(account_pb2.ChangePhoneReq(phone="")) 

75 

76 with session_scope() as session: 

77 user = session.execute(select(User).where(User.id == user_id)).scalar_one() 

78 assert user.phone is None 

79 assert user.phone_verification_token is None 

80 

81 

82def test_ChangePhone_ratelimit(db, monkeypatch): 

83 user, token = generate_user() 

84 user_id = user.id 

85 with account_session(token) as account: 

86 

87 def succeed(phone, message): 

88 return "success" 

89 

90 monkeypatch.setattr(couchers.phone.sms, "send_sms", succeed) 

91 

92 account.ChangePhone(account_pb2.ChangePhoneReq(phone="+46701740605")) 

93 

94 with pytest.raises(grpc.RpcError) as e: 

95 account.ChangePhone(account_pb2.ChangePhoneReq(phone="+46701740606")) 

96 assert e.value.code() == grpc.StatusCode.RESOURCE_EXHAUSTED 

97 

98 # Check that an earlier phone number/verification status is still saved 

99 with session_scope() as session: 

100 user = session.execute(select(User).where(User.id == user_id)).scalar_one() 

101 assert user.phone == "+46701740605" 

102 assert len(user.phone_verification_token) == 6 

103 

104 

105def test_VerifyPhone(): 

106 user, token = generate_user() 

107 user_id = user.id 

108 with account_session(token) as account, api_session(token) as api: 

109 

110 with pytest.raises(grpc.RpcError) as e: 

111 account.VerifyPhone(account_pb2.VerifyPhoneReq(token="123455")) 

112 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

113 

114 res = api.GetUser(api_pb2.GetUserReq(user=str(user_id))) 

115 assert res.verification == 0.0 

116 

117 with session_scope() as session: 

118 user = session.execute(select(User).where(User.id == user_id)).scalar_one() 

119 user.phone = "+46701740605" 

120 user.phone_verification_token = "111112" 

121 user.phone_verification_sent = now() 

122 

123 account.VerifyPhone(account_pb2.VerifyPhoneReq(token="111112")) 

124 

125 res = api.GetUser(api_pb2.GetUserReq(user=str(user_id))) 

126 assert res.verification == 1.0 

127 

128 # Phone number should finally show up on in your profile settings 

129 res = account.GetAccountInfo(empty_pb2.Empty()) 

130 assert res.phone == "+46701740605" 

131 

132 

133def test_VerifyPhone_antibrute(): 

134 user, token = generate_user() 

135 user_id = user.id 

136 with account_session(token) as account, api_session(token) as api: 

137 

138 with session_scope() as session: 

139 user = session.execute(select(User).where(User.id == user_id)).scalar_one() 

140 user.phone_verification_token = "111112" 

141 user.phone_verification_sent = now() 

142 user.phone = "+46701740605" 

143 

144 for i in range(10): 

145 with pytest.raises(grpc.RpcError) as e: 

146 account.VerifyPhone(account_pb2.VerifyPhoneReq(token="123455")) 

147 if e.value.code() != grpc.StatusCode.NOT_FOUND: 

148 break 

149 assert e.value.code() == grpc.StatusCode.RESOURCE_EXHAUSTED 

150 

151 

152def test_phone_uniqueness(monkeypatch): 

153 user1, token1 = generate_user() 

154 user2, token2 = generate_user() 

155 with account_session(token1) as account1, account_session(token2) as account2: 

156 

157 def succeed(phone, message): 

158 return "success" 

159 

160 monkeypatch.setattr(couchers.phone.sms, "send_sms", succeed) 

161 

162 account1.ChangePhone(account_pb2.ChangePhoneReq(phone="+46701740605")) 

163 with session_scope() as session: 

164 user = session.execute(select(User).where(User.id == user1.id)).scalar_one() 

165 token = user.phone_verification_token 

166 account1.VerifyPhone(account_pb2.VerifyPhoneReq(token=token)) 

167 res = account1.GetAccountInfo(empty_pb2.Empty()) 

168 assert res.phone == "+46701740605" 

169 assert res.phone_verified 

170 

171 # Let user2 steal user1:s phone number 

172 

173 account2.ChangePhone(account_pb2.ChangePhoneReq(phone="+46701740605")) 

174 

175 res = account1.GetAccountInfo(empty_pb2.Empty()) 

176 assert res.phone == "+46701740605" 

177 assert res.phone_verified 

178 

179 res = account2.GetAccountInfo(empty_pb2.Empty()) 

180 assert res.phone == "+46701740605" 

181 assert not res.phone_verified 

182 

183 with session_scope() as session: 

184 user = session.execute(select(User).where(User.id == user2.id)).scalar_one() 

185 token = user.phone_verification_token 

186 account2.VerifyPhone(account_pb2.VerifyPhoneReq(token=token)) 

187 

188 # number gets wiped when it's stolen 

189 res = account1.GetAccountInfo(empty_pb2.Empty()) 

190 assert not res.phone 

191 assert not res.phone_verified 

192 

193 res = account2.GetAccountInfo(empty_pb2.Empty()) 

194 assert res.phone == "+46701740605" 

195 assert res.phone_verified 

196 

197 

198def test_send_sms(db, monkeypatch): 

199 new_config = config.copy() 

200 new_config["ENABLE_SMS"] = True 

201 new_config["SMS_SENDER_ID"] = "CouchersOrg" 

202 monkeypatch.setattr(couchers.phone.sms, "config", new_config) 

203 

204 msg_id = random_hex() 

205 

206 with patch("couchers.phone.sms.boto3") as mock: 

207 sns = Mock() 

208 sns.publish.return_value = {"MessageId": msg_id} 

209 mock.client.return_value = sns 

210 

211 assert couchers.phone.sms.send_sms("+46701740605", "Testing SMS message") == "success" 

212 

213 mock.client.assert_called_once_with("sns") 

214 sns.publish.assert_called_once_with( 

215 PhoneNumber="+46701740605", 

216 Message="Testing SMS message", 

217 MessageAttributes={ 

218 "AWS.SNS.SMS.SMSType": {"DataType": "String", "StringValue": "Transactional"}, 

219 "AWS.SNS.SMS.SenderID": {"DataType": "String", "StringValue": "CouchersOrg"}, 

220 }, 

221 ) 

222 

223 with session_scope() as session: 

224 sms = session.execute(select(SMS)).scalar_one() 

225 assert sms.message_id == msg_id 

226 assert sms.sms_sender_id == "CouchersOrg" 

227 assert sms.number == "+46701740605" 

228 assert sms.message == "Testing SMS message" 

229 

230 

231def test_send_sms_disabled(db): 

232 assert not config["ENABLE_SMS"] 

233 assert couchers.phone.sms.send_sms("+46701740605", "Testing SMS message") == "SMS not enabled."