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

1import pytest 

2 

3from prisma import Prisma, errors 

4from prisma.models import Post, User 

5from prisma.partials import PostOnlyPublished 

6 

7from ..utils import RawQueries, DatabaseConfig 

8 

9 

10@pytest.mark.asyncio 

11async def test_query_raw( 

12 client: Prisma, 

13 raw_queries: RawQueries, 

14 config: DatabaseConfig, 

15) -> None: 

16 """Standard usage, erroneous query and correct queries""" 

17 with pytest.raises(errors.RawQueryError): 

18 await client.query_raw(raw_queries.select_unknown_table) 

19 

20 post = await client.post.create( 

21 { 

22 'title': 'My post title!', 

23 'published': False, 

24 } 

25 ) 

26 

27 results = await client.query_raw(raw_queries.count_posts) 

28 assert len(results) == 1 

29 assert isinstance(results[0]['count'], int) 

30 

31 results = await client.query_raw(raw_queries.find_post_by_id, post.id) 

32 assert len(results) == 1 

33 assert results[0]['id'] == post.id 

34 if config.bools_are_ints: 

35 assert results[0]['published'] == 0 

36 else: 

37 assert results[0]['published'] is False 

38 

39 

40@pytest.mark.asyncio 

41async def test_query_raw_model( 

42 client: Prisma, 

43 raw_queries: RawQueries, 

44) -> None: 

45 """Transforms resuls to a BaseModel when given""" 

46 post = await client.post.create( 

47 { 

48 'title': 'My post title!', 

49 'published': False, 

50 } 

51 ) 

52 

53 posts = await client.query_raw(raw_queries.find_post_by_id, post.id, model=Post) 

54 assert len(posts) == 1 

55 

56 found = posts[0] 

57 assert isinstance(found, Post) 

58 assert found == post 

59 assert found.id == post.id 

60 

61 

62@pytest.mark.asyncio 

63async def test_query_raw_partial_model( 

64 client: Prisma, 

65 raw_queries: RawQueries, 

66) -> None: 

67 """Transforms results to a partial model""" 

68 posts = [ 

69 await client.post.create({'title': 'foo', 'published': False}), 

70 await client.post.create({'title': 'foo', 'published': True}), 

71 await client.post.create({'title': 'foo', 'published': True}), 

72 await client.post.create({'title': 'foo', 'published': False}), 

73 ] 

74 results = await client.query_raw( 

75 raw_queries.find_posts_not_published, 

76 model=PostOnlyPublished, 

77 ) 

78 assert len(results) == 2 

79 assert {p.id for p in results} == {p.id for p in posts if p.published is False} 

80 assert not hasattr(results[0], 'title') 

81 assert results[0].published is False 

82 assert results[1].published is False 

83 

84 

85@pytest.mark.asyncio 

86async def test_query_raw_no_result( 

87 client: Prisma, 

88 raw_queries: RawQueries, 

89) -> None: 

90 """No result returns empty list""" 

91 results = await client.query_raw(raw_queries.test_query_raw_no_result) 

92 assert len(results) == 0 

93 

94 results = await client.query_raw( 

95 raw_queries.test_query_raw_no_result, 

96 model=Post, 

97 ) 

98 assert len(results) == 0 

99 

100 

101@pytest.mark.asyncio 

102@pytest.mark.skip(reason='Disabled as this test broke with prisma v5.15.0 - pending resolution with their team') 

103async def test_query_raw_incorrect_params( 

104 client: Prisma, 

105 raw_queries: RawQueries, 

106 config: DatabaseConfig, 

107) -> None: 

108 """Passings too many parameters raises an error""" 

109 if config.id == 'mysql': 

110 pytest.skip( 

111 'Passing the incorrect number of query parameters breaks subsequent queries', 

112 ) 

113 

114 results = await client.query_raw(raw_queries.count_posts) 

115 assert len(results) == 1 

116 assert results[0]['count'] == 0 

117 

118 # SQLite raises RawQueryError 

119 # PostgreSQL raises DataError 

120 with pytest.raises((errors.RawQueryError, errors.DataError)): 

121 await client.query_raw(raw_queries.count_posts, 1) 

122 

123 # subsequent queries can still be made 

124 results = await client.query_raw(raw_queries.count_posts) 

125 assert len(results) == 1 

126 assert results[0]['count'] == 0 

127 

128 

129@pytest.mark.asyncio 

130async def test_execute_raw(client: Prisma, raw_queries: RawQueries) -> None: 

131 """Basic usage""" 

132 post = await client.post.create( 

133 { 

134 'title': 'My post title.', 

135 'published': False, 

136 } 

137 ) 

138 assert isinstance(post.id, str) 

139 

140 count = await client.execute_raw( 

141 raw_queries.update_unique_post_title, 

142 post.id, 

143 ) 

144 assert count == 1 

145 

146 found = await client.post.find_unique(where={'id': post.id}) 

147 assert found is not None 

148 assert found.id == post.id 

149 assert found.title == 'My edited title' 

150 

151 

152@pytest.mark.asyncio 

153async def test_execute_raw_no_result( 

154 client: Prisma, 

155 raw_queries: RawQueries, 

156) -> None: 

157 """No result returns 0""" 

158 count = await client.execute_raw(raw_queries.test_execute_raw_no_result) 

159 assert count == 0 

160 

161 

162@pytest.mark.asyncio 

163async def test_query_first( 

164 client: Prisma, 

165 raw_queries: RawQueries, 

166) -> None: 

167 """Standard usage""" 

168 user = await client.user.create({'name': 'Robert'}) 

169 

170 found = await client.query_first(raw_queries.find_user_by_id, user.id) 

171 assert found['id'] == user.id 

172 assert found['name'] == 'Robert' 

173 

174 

175@pytest.mark.asyncio 

176async def test_query_first_model( 

177 client: Prisma, 

178 raw_queries: RawQueries, 

179) -> None: 

180 """Transforms result to a BaseModel if given""" 

181 user = await client.user.create({'name': 'Robert'}) 

182 

183 found = await client.query_first( 

184 raw_queries.find_user_by_id, 

185 user.id, 

186 model=User, 

187 ) 

188 assert found is not None 

189 assert found.id == user.id 

190 assert found.name == 'Robert'