mirror of
https://github.com/msgpack/msgpack-python.git
synced 2025-10-19 20:03:16 +00:00
8fb709f2e0
Cython implementation still used datetime.from_timestamp method, which does not work on windows. Update the cython implementation to use utc time and delta and add a regression test to highlight the issue.
131 lines
4.3 KiB
Python
131 lines
4.3 KiB
Python
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_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
|
|
|
|
|
|
@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_unpack_before_epoch():
|
|
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
|
|
|
|
|
|
@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
|