FazBrowse GitHub Viewer | Trending | | Home

Tools: [Download Repo ZIP] [View Raw Code] [Original HTTPS Page]

msgpack-python/test/test_timestamp.py at main · DiamondLightSource/msgpack-python · GitHub

218 lines (172 loc) · 7.8 KB

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

import datetime

import pytest

import msgpack

from msgpack.ext import Timestamp

def test_timestamp():

# timestamp32

ts = Timestamp(2**32 - 1)

assert ts.to_bytes() == b"\xff\xff\xff\xff"

packed = msgpack.packb(ts)

assert packed == b"\xd6\xff" + ts.to_bytes()

unpacked = msgpack.unpackb(packed)

assert ts == unpacked

assert ts.seconds == 2**32 - 1 and ts.nanoseconds == 0

# timestamp64

ts = Timestamp(2**34 - 1, 999999999)

assert ts.to_bytes() == b"\xee\x6b\x27\xff\xff\xff\xff\xff"

packed = msgpack.packb(ts)

assert packed == b"\xd7\xff" + ts.to_bytes()

unpacked = msgpack.unpackb(packed)

assert ts == unpacked

assert ts.seconds == 2**34 - 1 and ts.nanoseconds == 999999999

# timestamp96

ts = Timestamp(2**63 - 1, 999999999)

assert ts.to_bytes() == b"\x3b\x9a\xc9\xff\x7f\xff\xff\xff\xff\xff\xff\xff"

packed = msgpack.packb(ts)

assert packed == b"\xc7\x0c\xff" + ts.to_bytes()

unpacked = msgpack.unpackb(packed)

assert ts == unpacked

assert ts.seconds == 2**63 - 1 and ts.nanoseconds == 999999999

# negative fractional

ts = Timestamp.from_unix(-2.3) # s: -3, ns: 700000000

assert ts.seconds == -3 and ts.nanoseconds == 700000000

assert ts.to_bytes() == b"\x29\xb9\x27\x00\xff\xff\xff\xff\xff\xff\xff\xfd"

packed = msgpack.packb(ts)

assert packed == b"\xc7\x0c\xff" + ts.to_bytes()

unpacked = msgpack.unpackb(packed)

assert ts == unpacked

def test_unpack_timestamp():

# timestamp 32

assert msgpack.unpackb(b"\xd6\xff\x00\x00\x00\x00") == Timestamp(0)

# timestamp 64

assert msgpack.unpackb(b"\xd7\xff" + b"\x00" * 8) == Timestamp(0)

with pytest.raises(ValueError):

msgpack.unpackb(b"\xd7\xff" + b"\xff" * 8)

# timestamp 96

assert msgpack.unpackb(b"\xc7\x0c\xff" + b"\x00" * 12) == Timestamp(0)

with pytest.raises(ValueError):

msgpack.unpackb(b"\xc7\x0c\xff" + b"\xff" * 12) == Timestamp(0)

# Undefined

with pytest.raises(ValueError):

msgpack.unpackb(b"\xd4\xff\x00") # fixext 1

with pytest.raises(ValueError):

msgpack.unpackb(b"\xd5\xff\x00\x00") # fixext 2

with pytest.raises(ValueError):

msgpack.unpackb(b"\xc7\x00\xff") # ext8 (len=0)

with pytest.raises(ValueError):

msgpack.unpackb(b"\xc7\x03\xff\0\0\0") # ext8 (len=3)

with pytest.raises(ValueError):

msgpack.unpackb(b"\xc7\x05\xff\0\0\0\0\0") # ext8 (len=5)

def test_timestamp_from():

t = Timestamp(42, 14000)

assert Timestamp.from_unix(42.000014) == t

assert Timestamp.from_unix_nano(42000014000) == t

def test_timestamp_to():

t = Timestamp(42, 14000)

assert t.to_unix() == 42.000014

assert t.to_unix_nano() == 42000014000

def test_timestamp_datetime():

t = Timestamp(42, 14)

utc = datetime.timezone.utc

assert t.to_datetime() == datetime.datetime(1970, 1, 1, 0, 0, 42, 0, tzinfo=utc)

ts = datetime.datetime(2024, 4, 16, 8, 43, 9, 420317, tzinfo=utc)

ts2 = datetime.datetime(2024, 4, 16, 8, 43, 9, 420318, tzinfo=utc)

assert (

Timestamp.from_datetime(ts2).nanoseconds - Timestamp.from_datetime(ts).nanoseconds == 1000

)

ts3 = datetime.datetime(2024, 4, 16, 8, 43, 9, 4256)

ts4 = datetime.datetime(2024, 4, 16, 8, 43, 9, 4257)

assert (

Timestamp.from_datetime(ts4).nanoseconds - Timestamp.from_datetime(ts3).nanoseconds == 1000

)

assert Timestamp.from_datetime(ts).to_datetime() == ts

# Regression test: pre-epoch fractional seconds must floor toward -inf.

pre_epoch = datetime.datetime(1969, 12, 31, 23, 59, 59, 500000, tzinfo=utc)

ts_pre_epoch = Timestamp.from_datetime(pre_epoch)

assert ts_pre_epoch.seconds == -1

assert ts_pre_epoch.nanoseconds == 500000000

assert ts_pre_epoch.to_datetime() == pre_epoch

def test_from_datetime_far_future_precision():

# Regression: Timestamp.from_datetime used the float datetime.timestamp(),

# which cannot hold microsecond precision far from the epoch. It rounded the

# whole-second part up by one while still taking the exact microsecond for the

# nanoseconds, yielding a Timestamp one second in the future -- and raised

# OverflowError near datetime.max. It must instead match the integer

# arithmetic used by the Cython packer.

utc = datetime.timezone.utc

epoch = datetime.datetime(1970, 1, 1, tzinfo=utc)

for dt in [

datetime.datetime(2515, 1, 1, 0, 0, 0, 999999, tzinfo=utc),

datetime.datetime(3000, 1, 1, 0, 0, 0, 999999, tzinfo=utc),

datetime.datetime(5000, 6, 15, 12, 30, 45, 123456, tzinfo=utc),

# Near datetime.max: previously raised OverflowError.

datetime.datetime(9999, 12, 31, 23, 59, 59, 999999, tzinfo=utc),

]:

ts = Timestamp.from_datetime(dt)

# Round-trips exactly (was one second in the future).

assert ts.to_datetime() == dt

# Whole-second part is exact, computed independently of the implementation.

assert ts.seconds == (dt - epoch) // datetime.timedelta(seconds=1)

assert ts.nanoseconds == dt.microsecond * 1000

# The pure-Python path agrees with packing the datetime directly (the

# reference path), i.e. no off-by-one-second divergence.

assert msgpack.packb(ts) == msgpack.packb(dt, datetime=True)

# Explicit value: 3000-01-01T00:00:00.999999Z is 32503680000 s after the

# epoch; the float path produced 32503680001 (one second in the future).

ts = Timestamp.from_datetime(datetime.datetime(3000, 1, 1, 0, 0, 0, 999999, tzinfo=utc))

assert ts.seconds == 32503680000

assert ts.nanoseconds == 999999000

def test_unpack_datetime():

t = Timestamp(42, 14)

utc = datetime.timezone.utc

packed = msgpack.packb(t)

unpacked = msgpack.unpackb(packed, timestamp=3)

assert unpacked == datetime.datetime(1970, 1, 1, 0, 0, 42, 0, tzinfo=utc)

def test_pack_unpack_before_epoch():

utc = datetime.timezone.utc

t_in = datetime.datetime(1960, 1, 1, tzinfo=utc)

packed = msgpack.packb(t_in, datetime=True)

unpacked = msgpack.unpackb(packed, timestamp=3)

assert unpacked == t_in

def test_pack_datetime():

t = Timestamp(42, 14000)

dt = t.to_datetime()

utc = datetime.timezone.utc

assert dt == datetime.datetime(1970, 1, 1, 0, 0, 42, 14, tzinfo=utc)

packed = msgpack.packb(dt, datetime=True)

packed2 = msgpack.packb(t)

assert packed == packed2

unpacked = msgpack.unpackb(packed)

print(packed, unpacked)

assert unpacked == t

unpacked = msgpack.unpackb(packed, timestamp=3)

assert unpacked == dt

x = []

packed = msgpack.packb(dt, datetime=False, default=x.append)

assert x

assert x[0] == dt

assert msgpack.unpackb(packed) is None

def test_issue451():

# https://github.com/msgpack/msgpack-python/issues/451

utc = datetime.timezone.utc

dt = datetime.datetime(2100, 1, 1, 1, 1, tzinfo=utc)

packed = msgpack.packb(dt, datetime=True)

assert packed == b"\xd6\xff\xf4\x86eL"

unpacked = msgpack.unpackb(packed, timestamp=3)

assert dt == unpacked

def test_pack_datetime_without_tzinfo():

dt = datetime.datetime(1970, 1, 1, 0, 0, 42, 14)

with pytest.raises(ValueError, match="where tzinfo=None"):

packed = msgpack.packb(dt, datetime=True)

dt = datetime.datetime(1970, 1, 1, 0, 0, 42, 14)

packed = msgpack.packb(dt, datetime=True, default=lambda x: None)

assert packed == msgpack.packb(None)

utc = datetime.timezone.utc

dt = datetime.datetime(1970, 1, 1, 0, 0, 42, 14, tzinfo=utc)

packed = msgpack.packb(dt, datetime=True)

unpacked = msgpack.unpackb(packed, timestamp=3)

assert unpacked == dt

def test_too_large_timestamp():

# When timestamp64 is too large, conversion to datetime fails due to int64 -> int32 conversion.

# https://github.com/msgpack/msgpack-python/issues/696

print(msgpack.unpackb(b"\xd7\xff" + b"\x00" * 8, timestamp=3))

Back | FazBrowse Home | New Git URL