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

1from typing import List 

2 

3from pydantic import BaseModel 

4 

5from prisma import Prisma 

6from prisma.fields import Base64 

7from prisma.models import Types 

8from prisma._compat import ( 

9 model_dict, 

10 model_json, 

11 model_parse, 

12 model_parse_json, 

13) 

14 

15 

16def test_filtering(client: Prisma) -> None: 

17 """Finding records by a Bytes value""" 

18 with client.batch_() as batcher: 

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

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

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

22 

23 total = client.types.count( 

24 where={ 

25 'bytes': { 

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

27 }, 

28 }, 

29 ) 

30 assert total == 1 

31 

32 found = client.types.find_first( 

33 where={ 

34 'bytes': { 

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

36 }, 

37 }, 

38 ) 

39 assert found is not None 

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

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

42 

43 found = client.types.find_first( 

44 where={ 

45 'bytes': { 

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

47 }, 

48 }, 

49 ) 

50 assert found is not None 

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

52 

53 found = client.types.find_first( 

54 where={ 

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

56 }, 

57 ) 

58 assert found is not None 

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

60 

61 found = client.types.find_first( 

62 where={ 

63 'bytes': { 

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

65 } 

66 }, 

67 ) 

68 assert found is not None 

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

70 

71 found = client.types.find_first( 

72 where={ 

73 'bytes': { 

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

75 }, 

76 }, 

77 ) 

78 assert found is None 

79 

80 found_list = client.types.find_many( 

81 where={ 

82 'bytes': { 

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

84 } 

85 }, 

86 ) 

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

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

89 

90 

91def test_json(client: Prisma) -> None: 

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

93 record = client.types.create( 

94 data={ 

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

96 }, 

97 ) 

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

99 assert isinstance(model.bytes, Base64) 

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

101 

102 

103def test_constructing(client: Prisma) -> None: 

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

105 record = client.types.create({}) 

106 model = model_parse( 

107 Types, 

108 { 

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

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

111 }, 

112 ) 

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

114 

115 

116def test_filtering_nulls(client: Prisma) -> None: 

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

118 client.types.create( 

119 { 

120 'string': 'a', 

121 'optional_bytes': None, 

122 }, 

123 ) 

124 client.types.create( 

125 { 

126 'string': 'b', 

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

128 }, 

129 ) 

130 client.types.create( 

131 { 

132 'string': 'c', 

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

134 }, 

135 ) 

136 

137 found = client.types.find_first( 

138 where={ 

139 'NOT': [ 

140 { 

141 'optional_bytes': None, 

142 }, 

143 ], 

144 }, 

145 order={ 

146 'string': 'asc', 

147 }, 

148 ) 

149 assert found is not None 

150 assert found.string == 'b' 

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

152 

153 count = client.types.count( 

154 where={ 

155 'optional_bytes': None, 

156 }, 

157 ) 

158 assert count == 1 

159 

160 count = client.types.count( 

161 where={ 

162 'NOT': [ 

163 { 

164 'optional_bytes': None, 

165 }, 

166 ], 

167 }, 

168 ) 

169 assert count == 2 

170 

171 

172class Base64Model(BaseModel): 

173 value: Base64 

174 array: List[Base64] 

175 

176 

177def test_pydantic_conversion() -> None: 

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

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

180 assert isinstance(record.value, Base64) 

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

182 assert record.array == [] 

183 

184 record = model_parse( 

185 Base64Model, 

186 { 

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

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

189 }, 

190 ) 

191 assert isinstance(record.value, Base64) 

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

193 assert len(record.array) == 3 

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

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

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