|
@@ -10,7 +10,6 @@ from ..archiver.prune_cmd import prune_split
|
|
|
from ..constants import * # NOQA
|
|
|
from ..helpers import ChunkIteratorFileWrapper, ChunkerParams
|
|
|
from ..helpers import chunkit
|
|
|
-from ..helpers import safe_ns, safe_s, SUPPORT_32BIT_PLATFORMS
|
|
|
from ..helpers import popen_with_error_handling
|
|
|
from ..helpers import iter_separated
|
|
|
from ..helpers import is_slow_msgpack
|
|
@@ -216,40 +215,6 @@ def test_chunkit():
|
|
|
assert list(it) == []
|
|
|
|
|
|
|
|
|
-def utcfromtimestamp(timestamp):
|
|
|
- """Returns a naive datetime instance representing the timestamp in the UTC timezone"""
|
|
|
- return datetime.fromtimestamp(timestamp, timezone.utc).replace(tzinfo=None)
|
|
|
-
|
|
|
-
|
|
|
-def test_safe_timestamps():
|
|
|
- if SUPPORT_32BIT_PLATFORMS:
|
|
|
- # ns fit into int64
|
|
|
- assert safe_ns(2**64) <= 2**63 - 1
|
|
|
- assert safe_ns(-1) == 0
|
|
|
- # s fit into int32
|
|
|
- assert safe_s(2**64) <= 2**31 - 1
|
|
|
- assert safe_s(-1) == 0
|
|
|
- # datetime won't fall over its y10k problem
|
|
|
- beyond_y10k = 2**100
|
|
|
- with pytest.raises(OverflowError):
|
|
|
- utcfromtimestamp(beyond_y10k)
|
|
|
- assert utcfromtimestamp(safe_s(beyond_y10k)) > datetime(2038, 1, 1)
|
|
|
- assert utcfromtimestamp(safe_ns(beyond_y10k) / 1000000000) > datetime(2038, 1, 1)
|
|
|
- else:
|
|
|
- # ns fit into int64
|
|
|
- assert safe_ns(2**64) <= 2**63 - 1
|
|
|
- assert safe_ns(-1) == 0
|
|
|
- # s are so that their ns conversion fits into int64
|
|
|
- assert safe_s(2**64) * 1000000000 <= 2**63 - 1
|
|
|
- assert safe_s(-1) == 0
|
|
|
- # datetime won't fall over its y10k problem
|
|
|
- beyond_y10k = 2**100
|
|
|
- with pytest.raises(OverflowError):
|
|
|
- utcfromtimestamp(beyond_y10k)
|
|
|
- assert utcfromtimestamp(safe_s(beyond_y10k)) > datetime(2262, 1, 1)
|
|
|
- assert utcfromtimestamp(safe_ns(beyond_y10k) / 1000000000) > datetime(2262, 1, 1)
|
|
|
-
|
|
|
-
|
|
|
class TestPopenWithErrorHandling:
|
|
|
@pytest.mark.skipif(not shutil.which("test"), reason='"test" binary is needed')
|
|
|
def test_simple(self):
|