Coverage for databases/tests/test_models.py: 100%

1import pytest 

2 

3from prisma.models import User 

4 

5from ..utils import RawQueries 

6 

7 

8async def create_user() -> User: 

9 user = await User.prisma().create( 

10 data={ 

11 'name': 'Robert', 

12 }, 

13 ) 

14 assert isinstance(user, User) 

15 return user 

16 

17 

18@pytest.mark.prisma 

19@pytest.mark.asyncio 

20async def test_create() -> None: 

21 """Creating a record using model-based access""" 

22 user = await User.prisma().create({'name': 'Robert'}) 

23 assert isinstance(user, User) 

24 assert user.name == 'Robert' 

25 

26 user = await User.prisma().create( 

27 data={ 

28 'name': 'Tegan', 

29 'posts': { 

30 'create': { 

31 'title': 'My first post!', 

32 'published': True, 

33 }, 

34 }, 

35 }, 

36 include={ 

37 'posts': True, 

38 }, 

39 ) 

40 assert user.name == 'Tegan' 

41 assert user.posts is not None 

42 assert len(user.posts) == 1 

43 assert user.posts[0].title == 'My first post!' 

44 

45 

46@pytest.mark.prisma 

47@pytest.mark.asyncio 

48async def test_delete() -> None: 

49 """Deleting a record using model-based access""" 

50 deleted = await User.prisma().delete( 

51 where={ 

52 'id': 'abc', 

53 }, 

54 ) 

55 assert deleted is None 

56 

57 user = await User.prisma().create({'name': 'Robert'}) 

58 assert isinstance(user, User) 

59 

60 deleted2 = await User.prisma().delete( 

61 where={ 

62 'id': user.id, 

63 }, 

64 ) 

65 assert deleted2 is not None 

66 assert deleted2.name == 'Robert' 

67 

68 

69@pytest.mark.prisma 

70@pytest.mark.asyncio 

71async def test_find_unique() -> None: 

72 """Finding a unique record using model-based access""" 

73 found = await User.prisma().find_unique( 

74 where={ 

75 'id': 'abc', 

76 }, 

77 ) 

78 assert found is None 

79 

80 user = await create_user() 

81 found = await User.prisma().find_unique( 

82 where={ 

83 'id': user.id, 

84 }, 

85 ) 

86 assert found is not None 

87 assert found.name == 'Robert' 

88 

89 

90@pytest.mark.prisma 

91@pytest.mark.asyncio 

92async def test_find_many() -> None: 

93 """Finding many records using model-based access""" 

94 users = [ 

95 await User.prisma().create({'name': 'Robert'}), 

96 await User.prisma().create({'name': 'Robert 2'}), 

97 await User.prisma().create({'name': 'Tegan'}), 

98 ] 

99 

100 found = await User.prisma().find_many( 

101 where={ 

102 'name': { 

103 'startswith': 'Robert', 

104 }, 

105 }, 

106 order={ 

107 'id': 'asc', 

108 }, 

109 ) 

110 assert len(found) == 2 

111 assert found[0].id == users[0].id 

112 assert found[1].id == users[1].id 

113 

114 

115@pytest.mark.prisma 

116@pytest.mark.asyncio 

117async def test_find_first() -> None: 

118 """Finding a record using model-based access""" 

119 found = await User.prisma().find_first(where={'name': 'Robert'}) 

120 assert found is None 

121 

122 await create_user() 

123 

124 found = await User.prisma().find_first(where={'name': 'Robert'}) 

125 assert found is not None 

126 assert found.name == 'Robert' 

127 

128 

129@pytest.mark.prisma 

130@pytest.mark.asyncio 

131async def test_update() -> None: 

132 """Updating a record using model-based access""" 

133 user = await create_user() 

134 assert user.name == 'Robert' 

135 

136 updated = await User.prisma().update( 

137 data={ 

138 'name': 'Tegan', 

139 }, 

140 where={ 

141 'id': user.id, 

142 }, 

143 ) 

144 assert updated is not None 

145 assert updated.id == user.id 

146 assert updated.name == 'Tegan' 

147 

148 

149@pytest.mark.prisma 

150@pytest.mark.asyncio 

151async def test_upsert() -> None: 

152 """Upserting a record using model-based access""" 

153 user = await create_user() 

154 new = await User.prisma().upsert( 

155 where={ 

156 'id': user.id, 

157 }, 

158 data={ 

159 'create': { 

160 'name': 'Robert', 

161 }, 

162 'update': { 

163 'name': 'Robert 2', 

164 }, 

165 }, 

166 ) 

167 assert new.name == 'Robert 2' 

168 

169 

170@pytest.mark.prisma 

171@pytest.mark.asyncio 

172async def test_update_many() -> None: 

173 """Updating many records using model-based access""" 

174 users = [ 

175 await User.prisma().create({'name': 'Robert'}), 

176 await User.prisma().create({'name': 'Robert 2'}), 

177 await User.prisma().create({'name': 'Tegan'}), 

178 ] 

179 total = await User.prisma().update_many( 

180 where={ 

181 'name': { 

182 'startswith': 'Robert', 

183 }, 

184 }, 

185 data={ 

186 'name': 'Robert', 

187 }, 

188 ) 

189 assert total == 2 

190 

191 for id_ in [users[0].id, users[1].id]: 

192 user = await User.prisma().find_unique( 

193 where={ 

194 'id': id_, 

195 }, 

196 ) 

197 assert user is not None 

198 assert user.name == 'Robert' 

199 

200 user = await User.prisma().find_unique( 

201 where={ 

202 'id': users[2].id, 

203 }, 

204 ) 

205 assert user is not None 

206 assert user.name == 'Tegan' 

207 

208 

209@pytest.mark.prisma 

210@pytest.mark.asyncio 

211async def test_count() -> None: 

212 """Counting records using model-based access""" 

213 assert await User.prisma().count() == 0 

214 await create_user() 

215 assert await User.prisma().count() == 1 

216 

217 total = await User.prisma().count(where={'NOT': [{'name': 'Robert'}]}) 

218 assert total == 0 

219 

220 

221@pytest.mark.prisma 

222@pytest.mark.asyncio 

223async def test_delete_many() -> None: 

224 """Deleting many records using model-based access""" 

225 _ = [ 

226 await User.prisma().create({'name': 'Robert'}), 

227 await User.prisma().create({'name': 'Robert 2'}), 

228 await User.prisma().create({'name': 'Tegan'}), 

229 ] 

230 total = await User.prisma().delete_many( 

231 where={ 

232 'name': 'Tegan', 

233 }, 

234 ) 

235 assert total == 1 

236 assert await User.prisma().count() == 2 

237 

238 

239@pytest.mark.prisma 

240@pytest.mark.asyncio 

241async def test_query_raw(raw_queries: RawQueries) -> None: 

242 """Ensure results are transformed to the expected BaseModel""" 

243 users = [ 

244 await User.prisma().create({'name': 'Robert'}), 

245 await User.prisma().create({'name': 'Tegan'}), 

246 ] 

247 results = await User.prisma().query_raw( 

248 raw_queries.find_user_by_id, 

249 users[1].id, 

250 ) 

251 assert len(results) == 1 

252 assert results[0].name == 'Tegan' 

253 

254 

255@pytest.mark.prisma 

256@pytest.mark.asyncio 

257async def test_query_first(raw_queries: RawQueries) -> None: 

258 """Ensure results are transformed to the expected BaseModel""" 

259 users = [ 

260 await User.prisma().create({'name': 'Robert'}), 

261 await User.prisma().create({'name': 'Tegan'}), 

262 ] 

263 user = await User.prisma().query_first(raw_queries.find_user_by_id_limit_1, users[1].id) 

264 assert user is not None 

265 assert user.name == 'Tegan' 

266 

267 

268class MyUser(User): 

269 @property 

270 def info(self) -> str: 

271 return f'{self.id}: {self.name}' 

272 

273 

274@pytest.mark.prisma 

275@pytest.mark.asyncio 

276async def test_custom_model() -> None: 

277 """Subclassed prisma model is returned by actions""" 

278 user = await MyUser.prisma().create( 

279 data={ 

280 'name': 'Robert', 

281 }, 

282 ) 

283 assert user.info == f'{user.id}: Robert' 

284 assert isinstance(user, MyUser)