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

73 statements  

« prev     ^ index     » next       coverage.py v7.2.7, created at 2024-04-28 15:17 +0000

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 

10def test_query_raw( 

11 client: Prisma, 

12 raw_queries: RawQueries, 

13 config: DatabaseConfig, 

14) -> None: 

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

16 with pytest.raises(errors.RawQueryError): 

17 client.query_raw(raw_queries.select_unknown_table) 

18 

19 post = client.post.create( 

20 { 

21 'title': 'My post title!', 

22 'published': False, 

23 } 

24 ) 

25 

26 results = client.query_raw(raw_queries.count_posts) 

27 assert len(results) == 1 

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

29 

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

31 assert len(results) == 1 

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

33 if config.bools_are_ints: 

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

35 else: 

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

37 

38 

39def test_query_raw_model( 

40 client: Prisma, 

41 raw_queries: RawQueries, 

42) -> None: 

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

44 post = client.post.create( 

45 { 

46 'title': 'My post title!', 

47 'published': False, 

48 } 

49 ) 

50 

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

52 assert len(posts) == 1 

53 

54 found = posts[0] 

55 assert isinstance(found, Post) 

56 assert found == post 

57 assert found.id == post.id 

58 

59 

60def test_query_raw_partial_model( 

61 client: Prisma, 

62 raw_queries: RawQueries, 

63) -> None: 

64 """Transforms results to a partial model""" 

65 posts = [ 

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

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

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

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

70 ] 

71 results = client.query_raw( 

72 raw_queries.find_posts_not_published, 

73 model=PostOnlyPublished, 

74 ) 

75 assert len(results) == 2 

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

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

78 assert results[0].published is False 

79 assert results[1].published is False 

80 

81 

82def test_query_raw_no_result( 

83 client: Prisma, 

84 raw_queries: RawQueries, 

85) -> None: 

86 """No result returns empty list""" 

87 results = client.query_raw(raw_queries.test_query_raw_no_result) 

88 assert len(results) == 0 

89 

90 results = client.query_raw( 

91 raw_queries.test_query_raw_no_result, 

92 model=Post, 

93 ) 

94 assert len(results) == 0 

95 

96 

97def test_query_raw_incorrect_params( 

98 client: Prisma, 

99 raw_queries: RawQueries, 

100 config: DatabaseConfig, 

101) -> None: 

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

103 if config.id == 'mysql': 

104 pytest.skip( 

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

106 ) 

107 

108 results = client.query_raw(raw_queries.count_posts) 

109 assert len(results) == 1 

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

111 

112 # SQLite raises RawQueryError 

113 # PostgreSQL raises DataError 

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

115 client.query_raw(raw_queries.count_posts, 1) 

116 

117 # subsequent queries can still be made 

118 results = client.query_raw(raw_queries.count_posts) 

119 assert len(results) == 1 

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

121 

122 

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

124 """Basic usage""" 

125 post = client.post.create( 

126 { 

127 'title': 'My post title.', 

128 'published': False, 

129 } 

130 ) 

131 assert isinstance(post.id, str) 

132 

133 count = client.execute_raw( 

134 raw_queries.update_unique_post_title, 

135 post.id, 

136 ) 

137 assert count == 1 

138 

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

140 assert found is not None 

141 assert found.id == post.id 

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

143 

144 

145def test_execute_raw_no_result( 

146 client: Prisma, 

147 raw_queries: RawQueries, 

148) -> None: 

149 """No result returns 0""" 

150 count = client.execute_raw(raw_queries.test_execute_raw_no_result) 

151 assert count == 0 

152 

153 

154def test_query_first( 

155 client: Prisma, 

156 raw_queries: RawQueries, 

157) -> None: 

158 """Standard usage""" 

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

160 

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

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

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

164 

165 

166def test_query_first_model( 

167 client: Prisma, 

168 raw_queries: RawQueries, 

169) -> None: 

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

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

172 

173 found = client.query_first( 

174 raw_queries.find_user_by_id, 

175 user.id, 

176 model=User, 

177 ) 

178 assert found is not None 

179 assert found.id == user.id 

180 assert found.name == 'Robert'