Coverage for databases/tests/types/test_bytes.py: 100%

1from typing import List 

2 

3import pytest 

4from pydantic import BaseModel 

5from dirty_equals import IsPartialDict 

6 

7from prisma import Prisma 

8from prisma.fields import Base64 

9from prisma.models import Types 

10from prisma._compat import ( 

11 PYDANTIC_V2, 

12 model_dict, 

13 model_json, 

14 model_parse, 

15 model_parse_json, 

16 model_json_schema, 

17) 

18 

19 

20@pytest.mark.asyncio 

21async def test_filtering(client: Prisma) -> None: 

22 """Finding records by a Bytes value""" 

23 async with client.batch_() as batcher: 

24 batcher.types.create({'bytes': Base64.encode(b'a')}) 

25 batcher.types.create({'bytes': Base64.encode(b'b')}) 

26 batcher.types.create({'bytes': Base64.encode(b'foo bar')}) 

27 

28 total = await client.types.count( 

29 where={ 

30 'bytes': { 

31 'equals': Base64.encode(b'a'), 

32 }, 

33 }, 

34 ) 

35 assert total == 1 

36 

37 found = await client.types.find_first( 

38 where={ 

39 'bytes': { 

40 'equals': Base64.encode(b'foo bar'), 

41 }, 

42 }, 

43 ) 

44 assert found is not None 

45 assert found.bytes.decode() == b'foo bar' 

46 assert found.bytes.decode_str() == 'foo bar' 

47 

48 found = await client.types.find_first( 

49 where={ 

50 'bytes': { 

51 'not': Base64.encode(b'a'), 

52 }, 

53 }, 

54 ) 

55 assert found is not None 

56 assert found.bytes.decode() == b'b' 

57 

58 found = await client.types.find_first( 

59 where={ 

60 'bytes': Base64.encode(b'a'), 

61 }, 

62 ) 

63 assert found is not None 

64 assert found.bytes.decode() == b'a' 

65 

66 found = await client.types.find_first( 

67 where={ 

68 'bytes': { 

69 'in': [Base64.encode(b'a'), Base64.encode(b'c')], 

70 } 

71 }, 

72 ) 

73 assert found is not None 

74 assert found.bytes.decode() == b'a' 

75 

76 found = await client.types.find_first( 

77 where={ 

78 'bytes': { 

79 'in': [Base64.encode(b'c')], 

80 }, 

81 }, 

82 ) 

83 assert found is None 

84 

85 found_list = await client.types.find_many( 

86 where={ 

87 'bytes': { 

88 'not_in': [Base64.encode(b'a'), Base64.encode(b'c')], 

89 } 

90 }, 

91 ) 

92 found_values = {record.bytes.decode() for record in found_list} 

93 assert found_values == {b'b', b'foo bar'} 

94 

95 

96@pytest.mark.asyncio 

97async def test_json(client: Prisma) -> None: 

98 """Base64 fields can be serialised to and from JSON""" 

99 record = await client.types.create( 

100 data={ 

101 'bytes': Base64.encode(b'foo'), 

102 }, 

103 ) 

104 model = model_parse_json(Types, model_json(record, exclude={'json_obj'})) 

105 assert isinstance(model.bytes, Base64) 

106 assert model.bytes.decode() == b'foo' 

107 

108 

109@pytest.mark.asyncio 

110async def test_constructing(client: Prisma) -> None: 

111 """Base64 fields can be passed to the model constructor""" 

112 record = await client.types.create({}) 

113 model = model_parse( 

114 Types, 

115 { 

116 **model_dict(record, exclude={'json_obj'}), 

117 'bytes': Base64.encode(b'foo'), 

118 }, 

119 ) 

120 assert model.bytes == Base64.encode(b'foo') 

121 

122 

123@pytest.mark.asyncio 

124async def test_filtering_nulls(client: Prisma) -> None: 

125 """None is a valid filter for nullable Bytes fields""" 

126 await client.types.create( 

127 { 

128 'string': 'a', 

129 'optional_bytes': None, 

130 }, 

131 ) 

132 await client.types.create( 

133 { 

134 'string': 'b', 

135 'optional_bytes': Base64.encode(b'foo'), 

136 }, 

137 ) 

138 await client.types.create( 

139 { 

140 'string': 'c', 

141 'optional_bytes': Base64.encode(b'bar'), 

142 }, 

143 ) 

144 

145 found = await client.types.find_first( 

146 where={ 

147 'NOT': [ 

148 { 

149 'optional_bytes': None, 

150 }, 

151 ], 

152 }, 

153 order={ 

154 'string': 'asc', 

155 }, 

156 ) 

157 assert found is not None 

158 assert found.string == 'b' 

159 assert found.optional_bytes == Base64.encode(b'foo') 

160 

161 count = await client.types.count( 

162 where={ 

163 'optional_bytes': None, 

164 }, 

165 ) 

166 assert count == 1 

167 

168 count = await client.types.count( 

169 where={ 

170 'NOT': [ 

171 { 

172 'optional_bytes': None, 

173 }, 

174 ], 

175 }, 

176 ) 

177 assert count == 2 

178 

179 

180class Base64Model(BaseModel): 

181 value: Base64 

182 array: List[Base64] 

183 

184 

185def test_pydantic_conversion() -> None: 

186 """Raw inputs are converted to Base64 objects at the Pydantic level""" 

187 record = model_parse(Base64Model, {'value': 'foo', 'array': []}) 

188 assert isinstance(record.value, Base64) 

189 assert record.value._raw == b'foo' 

190 assert record.array == [] 

191 

192 record = model_parse( 

193 Base64Model, 

194 { 

195 'value': Base64.encode(b'foo'), 

196 'array': ['foo', b'bar', Base64.encode(b'baz')], 

197 }, 

198 ) 

199 assert isinstance(record.value, Base64) 

200 assert record.value.decode() == b'foo' 

201 assert len(record.array) == 3 

202 assert record.array[0]._raw == b'foo' 

203 assert record.array[1]._raw == b'bar' 

204 assert record.array[2].decode_str() == 'baz' 

205 

206 

207def test_json_schema() -> None: 

208 """Ensure a JSON Schema definition can be created""" 

209 if PYDANTIC_V2: 

210 assert model_json_schema(Types) == IsPartialDict( 

211 properties=IsPartialDict( 

212 { 

213 'bytes': { 

214 'format': 'byte', 

215 'title': 'Bytes', 

216 'type': 'string', 

217 }, 

218 'optional_bytes': { 

219 'anyOf': [ 

220 {'format': 'byte', 'type': 'string'}, 

221 {'type': 'null'}, 

222 ], 

223 'default': None, 

224 'title': 'Optional Bytes', 

225 }, 

226 } 

227 ) 

228 ) 

229 else: 

230 assert model_json_schema(Types) == IsPartialDict( 

231 properties=IsPartialDict( 

232 { 

233 'bytes': { 

234 'title': 'Bytes', 

235 'type': 'string', 

236 'format': 'byte', 

237 }, 

238 'optional_bytes': { 

239 'title': 'Optional Bytes', 

240 'type': 'string', 

241 'format': 'byte', 

242 }, 

243 } 

244 ) 

245 )