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
|
import pytest
import sys
import datetime
import msgpack
from msgpack.ext import Timestamp
if sys.version_info[0] > 2:
from msgpack.ext import _utc
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_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
@pytest.mark.skipif(sys.version_info[0] == 2, reason="datetime support is PY3+ only")
def test_timestamp_datetime():
t = Timestamp(42, 14)
assert t.to_datetime() == datetime.datetime(1970, 1, 1, 0, 0, 42, 0, tzinfo=_utc)
@pytest.mark.skipif(sys.version_info[0] == 2, reason="datetime support is PY3+ only")
def test_unpack_datetime():
t = Timestamp(42, 14)
packed = msgpack.packb(t)
unpacked = msgpack.unpackb(packed, timestamp=3)
assert unpacked == datetime.datetime(1970, 1, 1, 0, 0, 42, 0, tzinfo=_utc)
@pytest.mark.skipif(sys.version_info[0] == 2, reason="datetime support is PY3+ only")
def test_pack_datetime():
t = Timestamp(42, 14000)
dt = t.to_datetime()
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
|
