Coverage for databases/tests/arrays/test_datetime.py: 100%

1from typing import List 

2from datetime import datetime, timezone, timedelta 

3 

4import pytest 

5 

6from prisma import Prisma 

7 

8 

9def _utcnow() -> datetime: 

10 # workaround for https://github.com/RobertCraigie/prisma-client-py/issues/129 

11 now = datetime.now(timezone.utc) 

12 return now.replace(microsecond=int(now.microsecond / 1000) * 1000) 

13 

14 

15def _assert_datelist_equal(actual: List[datetime], expected: List[datetime]) -> None: 

16 actual = [dt.replace(tzinfo=None) for dt in actual] 

17 expected = [dt.replace(tzinfo=None) for dt in expected] 

18 assert actual == expected 

19 

20 

21@pytest.mark.asyncio 

22async def test_updating_datetime(client: Prisma) -> None: 

23 """Updating a DateTime[] value""" 

24 now = _utcnow() 

25 models = [ 

26 await client.lists.create({}), 

27 await client.lists.create( 

28 data={ 

29 'dates': [now, now + timedelta(hours=3)], 

30 }, 

31 ), 

32 ] 

33 

34 model = await client.lists.update( 

35 where={ 

36 'id': models[1].id, 

37 }, 

38 data={ 

39 'dates': { 

40 'set': [now + timedelta(days=999)], 

41 }, 

42 }, 

43 ) 

44 assert model is not None 

45 _assert_datelist_equal(model.dates, [now + timedelta(days=999)]) 

46 

47 model = await client.lists.update( 

48 where={ 

49 'id': models[1].id, 

50 }, 

51 data={ 

52 'dates': [now + timedelta(hours=999)], 

53 }, 

54 ) 

55 assert model is not None 

56 _assert_datelist_equal(model.dates, [now + timedelta(hours=999)]) 

57 

58 

59@pytest.mark.asyncio 

60async def test_filtering_datetime(client: Prisma) -> None: 

61 """Searching for records by a DateTime[] value""" 

62 now = _utcnow() 

63 expected_objects = [now, now + timedelta(hours=1)] 

64 async with client.batch_() as batcher: 

65 batcher.lists.create({}) 

66 batcher.lists.create( 

67 data={ 

68 'dates': [], 

69 }, 

70 ) 

71 batcher.lists.create( 

72 data={ 

73 'dates': expected_objects, 

74 }, 

75 ) 

76 

77 model = await client.lists.find_first( 

78 where={ 

79 'dates': { 

80 'equals': None, 

81 }, 

82 }, 

83 ) 

84 assert model is not None 

85 assert model.dates == [] 

86 

87 model = await client.lists.find_first( 

88 where={ 

89 'dates': { 

90 'equals': expected_objects, 

91 }, 

92 }, 

93 ) 

94 assert model is not None 

95 _assert_datelist_equal(model.dates, expected_objects) 

96 

97 model = await client.lists.find_first( 

98 where={ 

99 'dates': { 

100 'has': now, 

101 }, 

102 }, 

103 ) 

104 assert model is not None 

105 _assert_datelist_equal(model.dates, expected_objects) 

106 

107 model = await client.lists.find_first( 

108 where={ 

109 'dates': { 

110 'has': now + timedelta(seconds=5), 

111 }, 

112 }, 

113 ) 

114 assert model is None 

115 

116 model = await client.lists.find_first( 

117 where={ 

118 'dates': { 

119 'has_some': [*expected_objects, now + timedelta(seconds=5)], 

120 }, 

121 }, 

122 ) 

123 assert model is not None 

124 _assert_datelist_equal(model.dates, expected_objects) 

125 

126 model = await client.lists.find_first( 

127 where={ 

128 'dates': { 

129 'has_every': [*expected_objects, now + timedelta(minutes=10)], 

130 }, 

131 }, 

132 ) 

133 assert model is None 

134 

135 model = await client.lists.find_first( 

136 where={ 

137 'dates': { 

138 'has_every': [*expected_objects[:2]], 

139 }, 

140 }, 

141 ) 

142 assert model is not None 

143 _assert_datelist_equal(model.dates, expected_objects) 

144 

145 count = await client.lists.count( 

146 where={ 

147 'dates': { 

148 'is_empty': True, 

149 }, 

150 }, 

151 ) 

152 assert count == 1