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

1import grpc 

2import pytest 

3from google.protobuf import empty_pb2 

4 

5from couchers import errors, models 

6from couchers.constants import TOS_VERSION 

7from couchers.crypto import random_hex 

8from couchers.servicers import jail as servicers_jail 

9from proto import api_pb2, jail_pb2 

10from tests.test_fixtures import ( # noqa 

11 db, 

12 fast_passwords, 

13 generate_user, 

14 real_api_session, 

15 real_jail_session, 

16 testconfig, 

17) 

18 

19 

20@pytest.fixture(autouse=True) 

21def _(testconfig): 

22 pass 

23 

24 

25def test_jail_basic(db): 

26 user1, token1 = generate_user() 

27 

28 with real_api_session(token1) as api: 

29 res = api.Ping(api_pb2.PingReq()) 

30 

31 with real_jail_session(token1) as jail: 

32 res = jail.JailInfo(empty_pb2.Empty()) 

33 # check every field is false 

34 for field in res.DESCRIPTOR.fields: 

35 assert not getattr(res, field.name) 

36 

37 assert not res.jailed 

38 

39 # make the user jailed 

40 user2, token2 = generate_user(accepted_tos=0) 

41 

42 with real_api_session(token2) as api, pytest.raises(grpc.RpcError) as e: 

43 res = api.Ping(api_pb2.PingReq()) 

44 assert e.value.code() == grpc.StatusCode.UNAUTHENTICATED 

45 

46 with real_jail_session(token2) as jail: 

47 res = jail.JailInfo(empty_pb2.Empty()) 

48 

49 assert res.jailed 

50 

51 reason_count = 0 

52 

53 # check at least one field is true 

54 for field in res.DESCRIPTOR.fields: 

55 reason_count += getattr(res, field.name) == True 

56 

57 assert reason_count > 0 

58 

59 

60def test_JailInfo(db): 

61 user1, token1 = generate_user(accepted_tos=0) 

62 

63 with real_jail_session(token1) as jail: 

64 res = jail.JailInfo(empty_pb2.Empty()) 

65 assert res.jailed 

66 assert res.has_not_accepted_tos 

67 

68 with real_api_session(token1) as api, pytest.raises(grpc.RpcError) as e: 

69 res = api.Ping(api_pb2.PingReq()) 

70 assert e.value.code() == grpc.StatusCode.UNAUTHENTICATED 

71 

72 # make the user not jailed 

73 user2, token2 = generate_user() 

74 

75 with real_jail_session(token2) as jail: 

76 res = jail.JailInfo(empty_pb2.Empty()) 

77 assert not res.jailed 

78 assert not res.has_not_accepted_tos 

79 

80 with real_api_session(token2) as api: 

81 res = api.Ping(api_pb2.PingReq()) 

82 

83 

84def test_AcceptTOS(db): 

85 # make them have not accepted TOS 

86 user1, token1 = generate_user(accepted_tos=0) 

87 

88 with real_jail_session(token1) as jail: 

89 res = jail.JailInfo(empty_pb2.Empty()) 

90 assert res.jailed 

91 assert res.has_not_accepted_tos 

92 

93 # make sure we can't unaccept 

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

95 res = jail.AcceptTOS(jail_pb2.AcceptTOSReq(accept=False)) 

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

97 assert e.value.details() == errors.CANT_UNACCEPT_TOS 

98 

99 res = jail.JailInfo(empty_pb2.Empty()) 

100 assert res.jailed 

101 assert res.has_not_accepted_tos 

102 

103 # now accept 

104 res = jail.AcceptTOS(jail_pb2.AcceptTOSReq(accept=True)) 

105 

106 res = jail.JailInfo(empty_pb2.Empty()) 

107 assert not res.jailed 

108 assert not res.has_not_accepted_tos 

109 

110 # make sure we can't unaccept 

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

112 res = jail.AcceptTOS(jail_pb2.AcceptTOSReq(accept=False)) 

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

114 assert e.value.details() == errors.CANT_UNACCEPT_TOS 

115 

116 # make them have accepted TOS 

117 user2, token2 = generate_user() 

118 

119 with real_jail_session(token2) as jail: 

120 res = jail.JailInfo(empty_pb2.Empty()) 

121 assert not res.jailed 

122 assert not res.has_not_accepted_tos 

123 

124 # make sure we can't unaccept 

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

126 res = jail.AcceptTOS(jail_pb2.AcceptTOSReq(accept=False)) 

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

128 assert e.value.details() == errors.CANT_UNACCEPT_TOS 

129 

130 # accepting again doesn't do anything 

131 res = jail.AcceptTOS(jail_pb2.AcceptTOSReq(accept=True)) 

132 

133 res = jail.JailInfo(empty_pb2.Empty()) 

134 assert not res.jailed 

135 assert not res.has_not_accepted_tos 

136 

137 

138def test_TOS_increase(db, monkeypatch): 

139 # test if the TOS version is updated 

140 

141 # not jailed yet 

142 user, token = generate_user() 

143 

144 with real_jail_session(token) as jail: 

145 res = jail.JailInfo(empty_pb2.Empty()) 

146 assert not res.jailed 

147 assert not res.has_not_accepted_tos 

148 

149 with real_api_session(token) as api: 

150 res = api.Ping(api_pb2.PingReq()) 

151 

152 # now we pretend to update the TOS version 

153 new_TOS_VERSION = TOS_VERSION + 1 

154 

155 monkeypatch.setattr(models, "TOS_VERSION", new_TOS_VERSION) 

156 monkeypatch.setattr(servicers_jail, "TOS_VERSION", new_TOS_VERSION) 

157 

158 # make sure we're jailed 

159 with real_api_session(token) as api, pytest.raises(grpc.RpcError) as e: 

160 res = api.Ping(api_pb2.PingReq()) 

161 assert e.value.code() == grpc.StatusCode.UNAUTHENTICATED 

162 

163 with real_jail_session(token) as jail: 

164 res = jail.JailInfo(empty_pb2.Empty()) 

165 assert res.jailed 

166 assert res.has_not_accepted_tos 

167 

168 # now accept 

169 res = jail.AcceptTOS(jail_pb2.AcceptTOSReq(accept=True)) 

170 

171 res = jail.JailInfo(empty_pb2.Empty()) 

172 assert not res.jailed 

173 assert not res.has_not_accepted_tos 

174 

175 

176def test_SetLocation(db): 

177 # make them have not added a location 

178 user1, token1 = generate_user(geom=None, geom_radius=None) 

179 

180 with real_jail_session(token1) as jail: 

181 res = jail.JailInfo(empty_pb2.Empty()) 

182 assert res.jailed 

183 assert res.has_not_added_location 

184 

185 res = jail.SetLocation( 

186 jail_pb2.SetLocationReq( 

187 city="New York City", 

188 lat=40.7812, 

189 lng=-73.9647, 

190 radius=250, 

191 ) 

192 ) 

193 

194 assert not res.jailed 

195 assert not res.has_not_added_location 

196 

197 res = jail.JailInfo(empty_pb2.Empty()) 

198 assert not res.jailed 

199 assert not res.has_not_added_location 

200 

201 

202def test_AcceptCommunityGuidelines(db): 

203 # make them have not accepted GC 

204 user1, token1 = generate_user(accepted_community_guidelines=0) 

205 

206 with real_jail_session(token1) as jail: 

207 res = jail.JailInfo(empty_pb2.Empty()) 

208 assert res.jailed 

209 assert res.has_not_accepted_community_guidelines 

210 

211 # make sure we can't unaccept 

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

213 res = jail.AcceptCommunityGuidelines(jail_pb2.AcceptCommunityGuidelinesReq(accept=False)) 

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

215 assert e.value.details() == errors.CANT_UNACCEPT_COMMUNITY_GUIDELINES 

216 

217 res = jail.JailInfo(empty_pb2.Empty()) 

218 assert res.jailed 

219 assert res.has_not_accepted_community_guidelines 

220 

221 # now accept 

222 res = jail.AcceptCommunityGuidelines(jail_pb2.AcceptCommunityGuidelinesReq(accept=True)) 

223 

224 res = jail.JailInfo(empty_pb2.Empty()) 

225 assert not res.jailed 

226 assert not res.has_not_accepted_community_guidelines 

227 

228 # make sure we can't unaccept 

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

230 res = jail.AcceptCommunityGuidelines(jail_pb2.AcceptCommunityGuidelinesReq(accept=False)) 

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

232 assert e.value.details() == errors.CANT_UNACCEPT_COMMUNITY_GUIDELINES 

233 

234 # make them have accepted GC 

235 user2, token2 = generate_user() 

236 

237 with real_jail_session(token2) as jail: 

238 res = jail.JailInfo(empty_pb2.Empty()) 

239 assert not res.jailed 

240 assert not res.has_not_accepted_community_guidelines 

241 

242 # make sure we can't unaccept 

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

244 res = jail.AcceptCommunityGuidelines(jail_pb2.AcceptCommunityGuidelinesReq(accept=False)) 

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

246 assert e.value.details() == errors.CANT_UNACCEPT_COMMUNITY_GUIDELINES 

247 

248 # accepting again doesn't do anything 

249 res = jail.AcceptCommunityGuidelines(jail_pb2.AcceptCommunityGuidelinesReq(accept=True)) 

250 

251 res = jail.JailInfo(empty_pb2.Empty()) 

252 assert not res.jailed 

253 assert not res.has_not_accepted_community_guidelines 

254 

255 

256def test_SetPassword(db, fast_passwords): 

257 # make them not have a password 

258 user1, token1 = generate_user(hashed_password=None) 

259 

260 with real_jail_session(token1) as jail: 

261 res = jail.JailInfo(empty_pb2.Empty()) 

262 assert res.jailed 

263 assert res.has_not_set_password 

264 

265 # make sure bad password are caught 

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

267 jail.SetPassword(jail_pb2.SetPasswordReq(new_password="password")) 

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

269 assert e.value.details() == errors.INSECURE_PASSWORD 

270 

271 res = jail.JailInfo(empty_pb2.Empty()) 

272 assert res.jailed 

273 assert res.has_not_set_password 

274 

275 # make sure we can set a good password 

276 pwd = random_hex() 

277 res = jail.SetPassword(jail_pb2.SetPasswordReq(new_password=pwd)) 

278 

279 assert not res.jailed 

280 assert not res.has_not_set_password 

281 

282 # make sure we can't set a password again 

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

284 jail.SetPassword(jail_pb2.SetPasswordReq(new_password=random_hex())) 

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

286 assert e.value.details() == errors.ALREADY_HAS_PASSWORD 

287 

288 res = jail.JailInfo(empty_pb2.Empty()) 

289 assert not res.jailed 

290 assert not res.has_not_set_password