Coverage for databases/sync_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 

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 

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

98def test_query_raw_incorrect_params( 

99 client: Prisma, 

100 raw_queries: RawQueries, 

101 config: DatabaseConfig, 

102) -> None: 

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

104 if config.id == 'mysql': 

105 pytest.skip( 

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

107 ) 

108 

109 results = client.query_raw(raw_queries.count_posts) 

110 assert len(results) == 1 

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

112 

113 # SQLite raises RawQueryError 

114 # PostgreSQL raises DataError 

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

116 client.query_raw(raw_queries.count_posts, 1) 

117 

118 # subsequent queries can still be made 

119 results = client.query_raw(raw_queries.count_posts) 

120 assert len(results) == 1 

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

122 

123 

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

125 """Basic usage""" 

126 post = client.post.create( 

127 { 

128 'title': 'My post title.', 

129 'published': False, 

130 } 

131 ) 

132 assert isinstance(post.id, str) 

133 

134 count = client.execute_raw( 

135 raw_queries.update_unique_post_title, 

136 post.id, 

137 ) 

138 assert count == 1 

139 

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

141 assert found is not None 

142 assert found.id == post.id 

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

144 

145 

146def test_execute_raw_no_result( 

147 client: Prisma, 

148 raw_queries: RawQueries, 

149) -> None: 

150 """No result returns 0""" 

151 count = client.execute_raw(raw_queries.test_execute_raw_no_result) 

152 assert count == 0 

153 

154 

155def test_query_first( 

156 client: Prisma, 

157 raw_queries: RawQueries, 

158) -> None: 

159 """Standard usage""" 

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

161 

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

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

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

165 

166 

167def test_query_first_model( 

168 client: Prisma, 

169 raw_queries: RawQueries, 

170) -> None: 

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

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

173 

174 found = client.query_first( 

175 raw_queries.find_user_by_id, 

176 user.id, 

177 model=User, 

178 ) 

179 assert found is not None 

180 assert found.id == user.id 

181 assert found.name == 'Robert'