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 

102async def test_query_raw_incorrect_params( 

103 client: Prisma, 

104 raw_queries: RawQueries, 

105 config: DatabaseConfig, 

106) -> None: 

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

108 if config.id == 'mysql': 

109 pytest.skip( 

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

111 ) 

112 

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

114 assert len(results) == 1 

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

116 

117 # SQLite raises RawQueryError 

118 # PostgreSQL raises DataError 

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

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

121 

122 # subsequent queries can still be made 

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

124 assert len(results) == 1 

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

126 

127 

128@pytest.mark.asyncio 

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

130 """Basic usage""" 

131 post = await client.post.create( 

132 { 

133 'title': 'My post title.', 

134 'published': False, 

135 } 

136 ) 

137 assert isinstance(post.id, str) 

138 

139 count = await client.execute_raw( 

140 raw_queries.update_unique_post_title, 

141 post.id, 

142 ) 

143 assert count == 1 

144 

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

146 assert found is not None 

147 assert found.id == post.id 

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

149 

150 

151@pytest.mark.asyncio 

152async def test_execute_raw_no_result( 

153 client: Prisma, 

154 raw_queries: RawQueries, 

155) -> None: 

156 """No result returns 0""" 

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

158 assert count == 0 

159 

160 

161@pytest.mark.asyncio 

162async def test_query_first( 

163 client: Prisma, 

164 raw_queries: RawQueries, 

165) -> None: 

166 """Standard usage""" 

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

168 

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

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

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

172 

173 

174@pytest.mark.asyncio 

175async def test_query_first_model( 

176 client: Prisma, 

177 raw_queries: RawQueries, 

178) -> None: 

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

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

181 

182 found = await client.query_first( 

183 raw_queries.find_user_by_id, 

184 user.id, 

185 model=User, 

186 ) 

187 assert found is not None 

188 assert found.id == user.id 

189 assert found.name == 'Robert'