Coverage for databases/tests/types/test_datetime.py: 100%

82 statements  

« prev     ^ index     » next       coverage.py v7.2.7, created at 2024-08-27 18:25 +0000

1import datetime 

2 

3import pytest 

4from dirty_equals import IsPartialDict 

5 

6from prisma import Prisma 

7from lib.testing import assert_similar_time 

8from prisma.models import Types 

9from prisma._compat import PYDANTIC_V2, model_json_schema 

10 

11 

12@pytest.mark.asyncio 

13async def test_filtering(client: Prisma) -> None: 

14 """Finding records by a DateTime value""" 

15 now = datetime.datetime.now(datetime.timezone.utc) 

16 async with client.batch_() as batcher: 

17 for i in range(10): 

18 batcher.types.create({'datetime_': now + datetime.timedelta(hours=i)}) 

19 

20 total = await client.types.count(where={'datetime_': {'gte': now + datetime.timedelta(hours=5)}}) 

21 assert total == 5 

22 

23 found = await client.types.find_first( 

24 where={ 

25 'datetime_': { 

26 'equals': now, 

27 }, 

28 }, 

29 ) 

30 assert found is not None 

31 assert_similar_time(found.datetime_, now) 

32 

33 results = await client.types.find_many( 

34 where={ 

35 'datetime_': { 

36 'in': [ 

37 now + datetime.timedelta(hours=1), 

38 now + datetime.timedelta(hours=4), 

39 now + datetime.timedelta(hours=10), 

40 ], 

41 }, 

42 }, 

43 order={ 

44 'datetime_': 'asc', 

45 }, 

46 ) 

47 assert len(results) == 2 

48 assert_similar_time(results[0].datetime_, now + datetime.timedelta(hours=1)) 

49 assert_similar_time(results[1].datetime_, now + datetime.timedelta(hours=4)) 

50 

51 found = await client.types.find_first( 

52 where={ 

53 'datetime_': { 

54 'not_in': [ 

55 now, 

56 now + datetime.timedelta(hours=1), 

57 now + datetime.timedelta(hours=2), 

58 ], 

59 }, 

60 }, 

61 order={ 

62 'datetime_': 'asc', 

63 }, 

64 ) 

65 assert found is not None 

66 assert_similar_time(found.datetime_, now + datetime.timedelta(hours=3)) 

67 

68 found = await client.types.find_first( 

69 where={ 

70 'datetime_': { 

71 'lt': now + datetime.timedelta(hours=1), 

72 }, 

73 }, 

74 order={ 

75 'datetime_': 'desc', 

76 }, 

77 ) 

78 assert found is not None 

79 assert_similar_time(found.datetime_, now) 

80 

81 found = await client.types.find_first( 

82 where={ 

83 'datetime_': { 

84 'lte': now + datetime.timedelta(hours=1), 

85 }, 

86 }, 

87 order={ 

88 'datetime_': 'desc', 

89 }, 

90 ) 

91 assert found is not None 

92 assert_similar_time(found.datetime_, now + datetime.timedelta(hours=1)) 

93 

94 found = await client.types.find_first( 

95 where={ 

96 'datetime_': { 

97 'gt': now, 

98 }, 

99 }, 

100 order={ 

101 'datetime_': 'asc', 

102 }, 

103 ) 

104 assert found is not None 

105 assert_similar_time(found.datetime_, now + datetime.timedelta(hours=1)) 

106 

107 found = await client.types.find_first( 

108 where={ 

109 'datetime_': { 

110 'gte': now, 

111 }, 

112 }, 

113 order={ 

114 'datetime_': 'asc', 

115 }, 

116 ) 

117 assert found is not None 

118 assert_similar_time(found.datetime_, now) 

119 

120 found = await client.types.find_first( 

121 where={ 

122 'datetime_': { 

123 'not': now, 

124 }, 

125 }, 

126 order={ 

127 'datetime_': 'asc', 

128 }, 

129 ) 

130 assert found is not None 

131 assert_similar_time(found.datetime_, now + datetime.timedelta(hours=1)) 

132 

133 found = await client.types.find_first( 

134 where={ 

135 'datetime_': { 

136 'not': { 

137 'equals': now, 

138 }, 

139 }, 

140 }, 

141 order={ 

142 'datetime_': 'asc', 

143 }, 

144 ) 

145 assert found is not None 

146 assert_similar_time(now + datetime.timedelta(hours=1), found.datetime_) 

147 

148 

149@pytest.mark.asyncio 

150async def test_finds(client: Prisma) -> None: 

151 """Adding 1 second timedelta finds the record""" 

152 record = await client.types.create(data={}) 

153 found = await client.types.find_first( 

154 where={ 

155 'datetime_': { 

156 'lt': record.datetime_ + datetime.timedelta(seconds=1), 

157 }, 

158 }, 

159 ) 

160 assert found is not None 

161 assert found.id == record.id 

162 

163 

164@pytest.mark.asyncio 

165async def test_tz_aware(client: Prisma) -> None: 

166 """Modifying timezone still finds the record""" 

167 record = await client.types.create(data={}) 

168 found = await client.types.find_first( 

169 where={'datetime_': {'lt': (record.datetime_ + datetime.timedelta(hours=1)).astimezone(datetime.timezone.max)}} 

170 ) 

171 assert found is not None 

172 assert found.id == record.id 

173 

174 

175@pytest.mark.asyncio 

176async def test_filtering_nulls(client: Prisma) -> None: 

177 """None is a valid filter for nullable DateTime fields""" 

178 now = datetime.datetime.now(datetime.timezone.utc) 

179 await client.types.create( 

180 { 

181 'string': 'a', 

182 'optional_datetime': None, 

183 }, 

184 ) 

185 await client.types.create( 

186 { 

187 'string': 'b', 

188 'optional_datetime': now, 

189 }, 

190 ) 

191 await client.types.create( 

192 { 

193 'string': 'c', 

194 'optional_datetime': now + datetime.timedelta(days=1), 

195 }, 

196 ) 

197 

198 found = await client.types.find_first( 

199 where={ 

200 'NOT': [ 

201 { 

202 'optional_datetime': None, 

203 }, 

204 ], 

205 }, 

206 order={ 

207 'string': 'asc', 

208 }, 

209 ) 

210 assert found is not None 

211 assert found.string == 'b' 

212 assert found.optional_datetime is not None 

213 assert_similar_time(now, found.optional_datetime) 

214 

215 count = await client.types.count( 

216 where={ 

217 'optional_datetime': None, 

218 }, 

219 ) 

220 assert count == 1 

221 

222 count = await client.types.count( 

223 where={ 

224 'NOT': [ 

225 { 

226 'optional_datetime': None, 

227 }, 

228 ], 

229 }, 

230 ) 

231 assert count == 2 

232 

233 

234@pytest.mark.asyncio 

235async def test_precision_loss(client: Prisma) -> None: 

236 """https://github.com/RobertCraigie/prisma-client-py/issues/129""" 

237 date = datetime.datetime.now(datetime.timezone.utc) 

238 post = await client.post.create( 

239 data={ 

240 'title': 'My first post', 

241 'published': False, 

242 'created_at': date, 

243 }, 

244 ) 

245 found = await client.post.find_first( 

246 where={ 

247 'created_at': date, 

248 }, 

249 ) 

250 assert found is not None 

251 

252 found = await client.post.find_first( 

253 where={ 

254 'created_at': post.created_at, 

255 }, 

256 ) 

257 assert found is not None 

258 

259 

260def test_json_schema() -> None: 

261 """Ensure a JSON Schema definition can be created""" 

262 if PYDANTIC_V2: 

263 assert model_json_schema(Types) == IsPartialDict( 

264 properties=IsPartialDict( 

265 { 

266 'datetime_': { 

267 'title': 'Datetime ', 

268 'type': 'string', 

269 'format': 'date-time', 

270 }, 

271 'optional_datetime': { 

272 'title': 'Optional Datetime', 

273 'anyOf': [ 

274 {'format': 'date-time', 'type': 'string'}, 

275 {'type': 'null'}, 

276 ], 

277 'default': None, 

278 }, 

279 } 

280 ) 

281 ) 

282 else: 

283 assert model_json_schema(Types) == IsPartialDict( 

284 properties=IsPartialDict( 

285 { 

286 'datetime_': { 

287 'title': 'Datetime ', 

288 'type': 'string', 

289 'format': 'date-time', 

290 }, 

291 'optional_datetime': { 

292 'title': 'Optional Datetime', 

293 'type': 'string', 

294 'format': 'date-time', 

295 }, 

296 } 

297 ) 

298 )