mirror of
https://github.com/tahoe-lafs/tahoe-lafs.git
synced 2025-04-08 19:34:18 +00:00
Merge pull request #750 from tahoe-lafs/3346.abbreviate-and-time_format-py3
Port abbreviate and time_format to Python 3. Fixes ticket:3346
This commit is contained in:
Itamar Turner-Trauring
GitHub
commit
7d0f7737de
@ -1,4 +1,14 @@
|
||||
allmydata.test.mutable.test_exceptions.Exceptions.test_repr
|
||||
allmydata.test.test_abbreviate.Abbreviate.test_abbrev_time_1s
|
||||
allmydata.test.test_abbreviate.Abbreviate.test_abbrev_time_25s
|
||||
allmydata.test.test_abbreviate.Abbreviate.test_abbrev_time_day
|
||||
allmydata.test.test_abbreviate.Abbreviate.test_abbrev_time_future_5_minutes
|
||||
allmydata.test.test_abbreviate.Abbreviate.test_abbrev_time_hours
|
||||
allmydata.test.test_abbreviate.Abbreviate.test_abbrev_time_month
|
||||
allmydata.test.test_abbreviate.Abbreviate.test_abbrev_time_year
|
||||
allmydata.test.test_abbreviate.Abbreviate.test_parse_space
|
||||
allmydata.test.test_abbreviate.Abbreviate.test_space
|
||||
allmydata.test.test_abbreviate.Abbreviate.test_time
|
||||
allmydata.test.test_base32.Base32.test_a2b
|
||||
allmydata.test.test_base32.Base32.test_a2b_b2a_match_Pythons
|
||||
allmydata.test.test_base32.Base32.test_b2a
|
||||
@ -48,3 +58,11 @@ allmydata.test.test_pipeline.Pipeline.test_errors2
|
||||
allmydata.test.test_python3.Python3PortingEffortTests.test_finished_porting
|
||||
allmydata.test.test_python3.Python3PortingEffortTests.test_ported_modules_distinct
|
||||
allmydata.test.test_python3.Python3PortingEffortTests.test_ported_modules_exist
|
||||
allmydata.test.test_time_format.TimeFormat.test_epoch
|
||||
allmydata.test.test_time_format.TimeFormat.test_epoch_in_London
|
||||
allmydata.test.test_time_format.TimeFormat.test_format_delta
|
||||
allmydata.test.test_time_format.TimeFormat.test_format_time
|
||||
allmydata.test.test_time_format.TimeFormat.test_format_time_y2038
|
||||
allmydata.test.test_time_format.TimeFormat.test_iso_utc
|
||||
allmydata.test.test_time_format.TimeFormat.test_parse_date
|
||||
allmydata.test.test_time_format.TimeFormat.test_parse_duration
|
||||
|
||||
0
newsfragments/3346.minor
Normal file
0
newsfragments/3346.minor
Normal file
42
src/allmydata/test/common_py3.py
Normal file
42
src/allmydata/test/common_py3.py
Normal file
@ -0,0 +1,42 @@
|
||||
"""
|
||||
Common utilities that have been ported to Python 3.
|
||||
|
||||
Ported to Python 3.
|
||||
"""
|
||||
|
||||
from __future__ import unicode_literals
|
||||
from __future__ import absolute_import
|
||||
from __future__ import division
|
||||
from __future__ import print_function
|
||||
|
||||
from future.utils import PY2
|
||||
if PY2:
|
||||
from builtins import filter, map, zip, ascii, chr, hex, input, next, oct, open, pow, round, super, bytes, dict, int, list, object, range, str, max, min # noqa: F401
|
||||
|
||||
import os
|
||||
import time
|
||||
|
||||
|
||||
class TimezoneMixin(object):
|
||||
|
||||
def setTimezone(self, timezone):
|
||||
def tzset_if_possible():
|
||||
# Windows doesn't have time.tzset().
|
||||
if hasattr(time, 'tzset'):
|
||||
time.tzset()
|
||||
|
||||
unset = object()
|
||||
originalTimezone = os.environ.get('TZ', unset)
|
||||
def restoreTimezone():
|
||||
if originalTimezone is unset:
|
||||
del os.environ['TZ']
|
||||
else:
|
||||
os.environ['TZ'] = originalTimezone
|
||||
tzset_if_possible()
|
||||
|
||||
os.environ['TZ'] = timezone
|
||||
self.addCleanup(restoreTimezone)
|
||||
tzset_if_possible()
|
||||
|
||||
def have_working_tzset(self):
|
||||
return hasattr(time, 'tzset')
|
||||
@ -1,6 +1,6 @@
|
||||
from __future__ import print_function
|
||||
|
||||
import os, signal, time
|
||||
import os, signal
|
||||
from random import randrange
|
||||
from six.moves import StringIO
|
||||
|
||||
@ -185,31 +185,6 @@ class TestMixin(SignalMixin):
|
||||
self.fail("Reactor was still active when it was required to be quiescent.")
|
||||
|
||||
|
||||
class TimezoneMixin(object):
|
||||
|
||||
def setTimezone(self, timezone):
|
||||
def tzset_if_possible():
|
||||
# Windows doesn't have time.tzset().
|
||||
if hasattr(time, 'tzset'):
|
||||
time.tzset()
|
||||
|
||||
unset = object()
|
||||
originalTimezone = os.environ.get('TZ', unset)
|
||||
def restoreTimezone():
|
||||
if originalTimezone is unset:
|
||||
del os.environ['TZ']
|
||||
else:
|
||||
os.environ['TZ'] = originalTimezone
|
||||
tzset_if_possible()
|
||||
|
||||
os.environ['TZ'] = timezone
|
||||
self.addCleanup(restoreTimezone)
|
||||
tzset_if_possible()
|
||||
|
||||
def have_working_tzset(self):
|
||||
return hasattr(time, 'tzset')
|
||||
|
||||
|
||||
try:
|
||||
import win32file
|
||||
import win32con
|
||||
|
||||
149
src/allmydata/test/test_abbreviate.py
Normal file
149
src/allmydata/test/test_abbreviate.py
Normal file
@ -0,0 +1,149 @@
|
||||
"""
|
||||
Tests for allmydata.util.abbreviate.
|
||||
|
||||
Ported to Python 3.
|
||||
"""
|
||||
from __future__ import absolute_import
|
||||
from __future__ import division
|
||||
from __future__ import print_function
|
||||
from __future__ import unicode_literals
|
||||
|
||||
from future.utils import PY2
|
||||
if PY2:
|
||||
from builtins import filter, map, zip, ascii, chr, hex, input, next, oct, open, pow, round, super, bytes, dict, int, list, object, range, str, max, min # noqa: F401
|
||||
|
||||
from datetime import timedelta
|
||||
|
||||
from twisted.trial import unittest
|
||||
|
||||
from allmydata.util import abbreviate
|
||||
|
||||
|
||||
class Abbreviate(unittest.TestCase):
|
||||
def test_abbrev_time_1s(self):
|
||||
diff = timedelta(seconds=1)
|
||||
s = abbreviate.abbreviate_time(diff)
|
||||
self.assertEqual('1 second ago', s)
|
||||
|
||||
def test_abbrev_time_25s(self):
|
||||
diff = timedelta(seconds=25)
|
||||
s = abbreviate.abbreviate_time(diff)
|
||||
self.assertEqual('25 seconds ago', s)
|
||||
|
||||
def test_abbrev_time_future_5_minutes(self):
|
||||
diff = timedelta(minutes=-5)
|
||||
s = abbreviate.abbreviate_time(diff)
|
||||
self.assertEqual('5 minutes in the future', s)
|
||||
|
||||
def test_abbrev_time_hours(self):
|
||||
diff = timedelta(hours=4)
|
||||
s = abbreviate.abbreviate_time(diff)
|
||||
self.assertEqual('4 hours ago', s)
|
||||
|
||||
def test_abbrev_time_day(self):
|
||||
diff = timedelta(hours=49) # must be more than 2 days
|
||||
s = abbreviate.abbreviate_time(diff)
|
||||
self.assertEqual('2 days ago', s)
|
||||
|
||||
def test_abbrev_time_month(self):
|
||||
diff = timedelta(days=91)
|
||||
s = abbreviate.abbreviate_time(diff)
|
||||
self.assertEqual('3 months ago', s)
|
||||
|
||||
def test_abbrev_time_year(self):
|
||||
diff = timedelta(weeks=(5 * 52) + 1)
|
||||
s = abbreviate.abbreviate_time(diff)
|
||||
self.assertEqual('5 years ago', s)
|
||||
|
||||
def test_time(self):
|
||||
a = abbreviate.abbreviate_time
|
||||
self.failUnlessEqual(a(None), "unknown")
|
||||
self.failUnlessEqual(a(0), "0 seconds")
|
||||
self.failUnlessEqual(a(1), "1 second")
|
||||
self.failUnlessEqual(a(2), "2 seconds")
|
||||
self.failUnlessEqual(a(119), "119 seconds")
|
||||
MIN = 60
|
||||
self.failUnlessEqual(a(2*MIN), "2 minutes")
|
||||
self.failUnlessEqual(a(60*MIN), "60 minutes")
|
||||
self.failUnlessEqual(a(179*MIN), "179 minutes")
|
||||
HOUR = 60*MIN
|
||||
self.failUnlessEqual(a(180*MIN), "3 hours")
|
||||
self.failUnlessEqual(a(4*HOUR), "4 hours")
|
||||
DAY = 24*HOUR
|
||||
MONTH = 30*DAY
|
||||
self.failUnlessEqual(a(2*DAY), "2 days")
|
||||
self.failUnlessEqual(a(2*MONTH), "2 months")
|
||||
YEAR = 365*DAY
|
||||
self.failUnlessEqual(a(5*YEAR), "5 years")
|
||||
|
||||
def test_space(self):
|
||||
tests_si = [(None, "unknown"),
|
||||
(0, "0 B"),
|
||||
(1, "1 B"),
|
||||
(999, "999 B"),
|
||||
(1000, "1000 B"),
|
||||
(1023, "1023 B"),
|
||||
(1024, "1.02 kB"),
|
||||
(20*1000, "20.00 kB"),
|
||||
(1024*1024, "1.05 MB"),
|
||||
(1000*1000, "1.00 MB"),
|
||||
(1000*1000*1000, "1.00 GB"),
|
||||
(1000*1000*1000*1000, "1.00 TB"),
|
||||
(1000*1000*1000*1000*1000, "1.00 PB"),
|
||||
(1000*1000*1000*1000*1000*1000, "1.00 EB"),
|
||||
(1234567890123456789, "1.23 EB"),
|
||||
]
|
||||
for (x, expected) in tests_si:
|
||||
got = abbreviate.abbreviate_space(x, SI=True)
|
||||
self.failUnlessEqual(got, expected)
|
||||
|
||||
tests_base1024 = [(None, "unknown"),
|
||||
(0, "0 B"),
|
||||
(1, "1 B"),
|
||||
(999, "999 B"),
|
||||
(1000, "1000 B"),
|
||||
(1023, "1023 B"),
|
||||
(1024, "1.00 kiB"),
|
||||
(20*1024, "20.00 kiB"),
|
||||
(1000*1000, "976.56 kiB"),
|
||||
(1024*1024, "1.00 MiB"),
|
||||
(1024*1024*1024, "1.00 GiB"),
|
||||
(1024*1024*1024*1024, "1.00 TiB"),
|
||||
(1000*1000*1000*1000*1000, "909.49 TiB"),
|
||||
(1024*1024*1024*1024*1024, "1.00 PiB"),
|
||||
(1024*1024*1024*1024*1024*1024, "1.00 EiB"),
|
||||
(1234567890123456789, "1.07 EiB"),
|
||||
]
|
||||
for (x, expected) in tests_base1024:
|
||||
got = abbreviate.abbreviate_space(x, SI=False)
|
||||
self.failUnlessEqual(got, expected)
|
||||
|
||||
self.failUnlessEqual(abbreviate.abbreviate_space_both(1234567),
|
||||
"(1.23 MB, 1.18 MiB)")
|
||||
|
||||
def test_parse_space(self):
|
||||
p = abbreviate.parse_abbreviated_size
|
||||
self.failUnlessEqual(p(""), None)
|
||||
self.failUnlessEqual(p(None), None)
|
||||
self.failUnlessEqual(p("123"), 123)
|
||||
self.failUnlessEqual(p("123B"), 123)
|
||||
self.failUnlessEqual(p("2K"), 2000)
|
||||
self.failUnlessEqual(p("2kb"), 2000)
|
||||
self.failUnlessEqual(p("2KiB"), 2048)
|
||||
self.failUnlessEqual(p("10MB"), 10*1000*1000)
|
||||
self.failUnlessEqual(p("10MiB"), 10*1024*1024)
|
||||
self.failUnlessEqual(p("5G"), 5*1000*1000*1000)
|
||||
self.failUnlessEqual(p("4GiB"), 4*1024*1024*1024)
|
||||
self.failUnlessEqual(p("3TB"), 3*1000*1000*1000*1000)
|
||||
self.failUnlessEqual(p("3TiB"), 3*1024*1024*1024*1024)
|
||||
self.failUnlessEqual(p("6PB"), 6*1000*1000*1000*1000*1000)
|
||||
self.failUnlessEqual(p("6PiB"), 6*1024*1024*1024*1024*1024)
|
||||
self.failUnlessEqual(p("9EB"), 9*1000*1000*1000*1000*1000*1000)
|
||||
self.failUnlessEqual(p("9EiB"), 9*1024*1024*1024*1024*1024*1024)
|
||||
|
||||
e = self.failUnlessRaises(ValueError, p, "12 cubits")
|
||||
self.failUnlessIn("12 cubits", str(e))
|
||||
e = self.failUnlessRaises(ValueError, p, "1 BB")
|
||||
self.failUnlessIn("1 BB", str(e))
|
||||
e = self.failUnlessRaises(ValueError, p, "fhtagn")
|
||||
self.failUnlessIn("fhtagn", str(e))
|
||||
@ -26,7 +26,7 @@ from zope.interface import implementer
|
||||
from foolscap.api import fireEventually
|
||||
import itertools
|
||||
from allmydata import interfaces
|
||||
from allmydata.util import fileutil, hashutil, base32, pollmixin, time_format
|
||||
from allmydata.util import fileutil, hashutil, base32, pollmixin
|
||||
from allmydata.storage.server import StorageServer
|
||||
from allmydata.storage.mutable import MutableShareFile
|
||||
from allmydata.storage.immutable import BucketWriter, BucketReader
|
||||
@ -3852,25 +3852,6 @@ class LeaseCrawler(unittest.TestCase, pollmixin.PollMixin):
|
||||
expiration_mode="bogus")
|
||||
self.failUnlessIn("GC mode 'bogus' must be 'age' or 'cutoff-date'", str(e))
|
||||
|
||||
def test_parse_duration(self):
|
||||
DAY = 24*60*60
|
||||
MONTH = 31*DAY
|
||||
YEAR = 365*DAY
|
||||
p = time_format.parse_duration
|
||||
self.failUnlessEqual(p("7days"), 7*DAY)
|
||||
self.failUnlessEqual(p("31day"), 31*DAY)
|
||||
self.failUnlessEqual(p("60 days"), 60*DAY)
|
||||
self.failUnlessEqual(p("2mo"), 2*MONTH)
|
||||
self.failUnlessEqual(p("3 month"), 3*MONTH)
|
||||
self.failUnlessEqual(p("2years"), 2*YEAR)
|
||||
e = self.failUnlessRaises(ValueError, p, "2kumquats")
|
||||
self.failUnlessIn("no unit (like day, month, or year) in '2kumquats'", str(e))
|
||||
|
||||
def test_parse_date(self):
|
||||
p = time_format.parse_date
|
||||
self.failUnless(isinstance(p("2009-03-18"), int), p("2009-03-18"))
|
||||
self.failUnlessEqual(p("2009-03-18"), 1237334400)
|
||||
|
||||
def test_limited_history(self):
|
||||
basedir = "storage/LeaseCrawler/limited_history"
|
||||
fileutil.make_dirs(basedir)
|
||||
|
||||
169
src/allmydata/test/test_time_format.py
Normal file
169
src/allmydata/test/test_time_format.py
Normal file
@ -0,0 +1,169 @@
|
||||
"""
|
||||
Tests for allmydata.util.time_format.
|
||||
"""
|
||||
from __future__ import absolute_import
|
||||
from __future__ import division
|
||||
from __future__ import print_function
|
||||
from __future__ import unicode_literals
|
||||
|
||||
from future.utils import PY2
|
||||
if PY2:
|
||||
from builtins import filter, map, zip, ascii, chr, hex, input, next, oct, open, pow, round, super, bytes, dict, int, list, object, range, str, max, min # noqa: F401
|
||||
|
||||
from past.builtins import long
|
||||
|
||||
import time
|
||||
|
||||
from twisted.trial import unittest
|
||||
|
||||
from allmydata.test.common_py3 import TimezoneMixin
|
||||
from allmydata.util import time_format
|
||||
|
||||
|
||||
class TimeFormat(unittest.TestCase, TimezoneMixin):
|
||||
def test_epoch(self):
|
||||
return self._help_test_epoch()
|
||||
|
||||
def test_epoch_in_London(self):
|
||||
# Europe/London is a particularly troublesome timezone. Nowadays, its
|
||||
# offset from GMT is 0. But in 1970, its offset from GMT was 1.
|
||||
# (Apparently in 1970 Britain had redefined standard time to be GMT+1
|
||||
# and stayed in standard time all year round, whereas today
|
||||
# Europe/London standard time is GMT and Europe/London Daylight
|
||||
# Savings Time is GMT+1.) The current implementation of
|
||||
# time_format.iso_utc_time_to_localseconds() breaks if the timezone is
|
||||
# Europe/London. (As soon as this unit test is done then I'll change
|
||||
# that implementation to something that works even in this case...)
|
||||
|
||||
if not self.have_working_tzset():
|
||||
raise unittest.SkipTest("This test can't be run on a platform without time.tzset().")
|
||||
|
||||
self.setTimezone("Europe/London")
|
||||
return self._help_test_epoch()
|
||||
|
||||
def _help_test_epoch(self):
|
||||
origtzname = time.tzname
|
||||
s = time_format.iso_utc_time_to_seconds("1970-01-01T00:00:01")
|
||||
self.failUnlessEqual(s, 1.0)
|
||||
s = time_format.iso_utc_time_to_seconds("1970-01-01_00:00:01")
|
||||
self.failUnlessEqual(s, 1.0)
|
||||
s = time_format.iso_utc_time_to_seconds("1970-01-01 00:00:01")
|
||||
self.failUnlessEqual(s, 1.0)
|
||||
|
||||
self.failUnlessEqual(time_format.iso_utc(1.0), "1970-01-01_00:00:01")
|
||||
self.failUnlessEqual(time_format.iso_utc(1.0, sep=" "),
|
||||
"1970-01-01 00:00:01")
|
||||
|
||||
now = time.time()
|
||||
isostr = time_format.iso_utc(now)
|
||||
timestamp = time_format.iso_utc_time_to_seconds(isostr)
|
||||
self.failUnlessEqual(int(timestamp), int(now))
|
||||
|
||||
def my_time():
|
||||
return 1.0
|
||||
self.failUnlessEqual(time_format.iso_utc(t=my_time),
|
||||
"1970-01-01_00:00:01")
|
||||
e = self.failUnlessRaises(ValueError,
|
||||
time_format.iso_utc_time_to_seconds,
|
||||
"invalid timestring")
|
||||
self.failUnless("not a complete ISO8601 timestamp" in str(e))
|
||||
s = time_format.iso_utc_time_to_seconds("1970-01-01_00:00:01.500")
|
||||
self.failUnlessEqual(s, 1.5)
|
||||
|
||||
# Look for daylight-savings-related errors.
|
||||
thatmomentinmarch = time_format.iso_utc_time_to_seconds("2009-03-20 21:49:02.226536")
|
||||
self.failUnlessEqual(thatmomentinmarch, 1237585742.226536)
|
||||
self.failUnlessEqual(origtzname, time.tzname)
|
||||
|
||||
def test_iso_utc(self):
|
||||
when = 1266760143.7841301
|
||||
out = time_format.iso_utc_date(when)
|
||||
self.failUnlessEqual(out, "2010-02-21")
|
||||
out = time_format.iso_utc_date(t=lambda: when)
|
||||
self.failUnlessEqual(out, "2010-02-21")
|
||||
out = time_format.iso_utc(when)
|
||||
self.failUnlessEqual(out, "2010-02-21_13:49:03.784130")
|
||||
out = time_format.iso_utc(when, sep="-")
|
||||
self.failUnlessEqual(out, "2010-02-21-13:49:03.784130")
|
||||
|
||||
def test_parse_duration(self):
|
||||
p = time_format.parse_duration
|
||||
DAY = 24*60*60
|
||||
MONTH = 31*DAY
|
||||
YEAR = 365*DAY
|
||||
self.failUnlessEqual(p("1 day"), DAY)
|
||||
self.failUnlessEqual(p("2 days"), 2*DAY)
|
||||
self.failUnlessEqual(p("3 months"), 3*MONTH)
|
||||
self.failUnlessEqual(p("4 mo"), 4*MONTH)
|
||||
self.failUnlessEqual(p("5 years"), 5*YEAR)
|
||||
e = self.failUnlessRaises(ValueError, p, "123")
|
||||
self.failUnlessIn("no unit (like day, month, or year) in '123'",
|
||||
str(e))
|
||||
self.failUnlessEqual(p("7days"), 7*DAY)
|
||||
self.failUnlessEqual(p("31day"), 31*DAY)
|
||||
self.failUnlessEqual(p("60 days"), 60*DAY)
|
||||
self.failUnlessEqual(p("2mo"), 2*MONTH)
|
||||
self.failUnlessEqual(p("3 month"), 3*MONTH)
|
||||
self.failUnlessEqual(p("2years"), 2*YEAR)
|
||||
e = self.failUnlessRaises(ValueError, p, "2kumquats")
|
||||
self.failUnlessIn("no unit (like day, month, or year) in '2kumquats'", str(e))
|
||||
|
||||
def test_parse_date(self):
|
||||
p = time_format.parse_date
|
||||
self.failUnlessEqual(p("2010-02-21"), 1266710400)
|
||||
self.failUnless(isinstance(p("2009-03-18"), (int, long)), p("2009-03-18"))
|
||||
self.failUnlessEqual(p("2009-03-18"), 1237334400)
|
||||
|
||||
def test_format_time(self):
|
||||
self.failUnlessEqual(time_format.format_time(time.gmtime(0)), '1970-01-01 00:00:00')
|
||||
self.failUnlessEqual(time_format.format_time(time.gmtime(60)), '1970-01-01 00:01:00')
|
||||
self.failUnlessEqual(time_format.format_time(time.gmtime(60*60)), '1970-01-01 01:00:00')
|
||||
seconds_per_day = 60*60*24
|
||||
leap_years_1970_to_2014_inclusive = ((2012 - 1968) // 4)
|
||||
self.failUnlessEqual(time_format.format_time(time.gmtime(seconds_per_day*((2015 - 1970)*365+leap_years_1970_to_2014_inclusive))), '2015-01-01 00:00:00')
|
||||
|
||||
def test_format_time_y2038(self):
|
||||
seconds_per_day = 60*60*24
|
||||
leap_years_1970_to_2047_inclusive = ((2044 - 1968) // 4)
|
||||
t = (seconds_per_day*
|
||||
((2048 - 1970)*365 + leap_years_1970_to_2047_inclusive))
|
||||
try:
|
||||
gm_t = time.gmtime(t)
|
||||
except ValueError:
|
||||
raise unittest.SkipTest("Note: this system cannot handle dates after 2037.")
|
||||
self.failUnlessEqual(time_format.format_time(gm_t),
|
||||
'2048-01-01 00:00:00')
|
||||
|
||||
def test_format_delta(self):
|
||||
time_1 = 1389812723
|
||||
time_5s_delta = 1389812728
|
||||
time_28m7s_delta = 1389814410
|
||||
time_1h_delta = 1389816323
|
||||
time_1d21h46m49s_delta = 1389977532
|
||||
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1, time_1), '0s')
|
||||
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1, time_5s_delta), '5s')
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1, time_28m7s_delta), '28m 7s')
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1, time_1h_delta), '1h 0m 0s')
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1, time_1d21h46m49s_delta), '1d 21h 46m 49s')
|
||||
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1d21h46m49s_delta, time_1), '-')
|
||||
|
||||
# time_1 with a decimal fraction will make the delta 1s less
|
||||
time_1decimal = 1389812723.383963
|
||||
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1decimal, time_5s_delta), '4s')
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1decimal, time_28m7s_delta), '28m 6s')
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1decimal, time_1h_delta), '59m 59s')
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1decimal, time_1d21h46m49s_delta), '1d 21h 46m 48s')
|
||||
@ -7,19 +7,18 @@ import os, time, sys
|
||||
import yaml
|
||||
|
||||
from six.moves import StringIO
|
||||
from datetime import timedelta
|
||||
from twisted.trial import unittest
|
||||
from twisted.internet import defer, reactor
|
||||
from twisted.python.failure import Failure
|
||||
|
||||
from allmydata.util import idlib, mathutil
|
||||
from allmydata.util import fileutil, abbreviate
|
||||
from allmydata.util import limiter, time_format, pollmixin
|
||||
from allmydata.util import fileutil
|
||||
from allmydata.util import limiter, pollmixin
|
||||
from allmydata.util import statistics, dictutil, yamlutil
|
||||
from allmydata.util import log as tahoe_log
|
||||
from allmydata.util.spans import Spans, overlap, DataSpans
|
||||
from allmydata.util.fileutil import EncryptedTemporaryFile
|
||||
from allmydata.test.common_util import ReallyEqualMixin, TimezoneMixin
|
||||
from allmydata.test.common_util import ReallyEqualMixin
|
||||
|
||||
if six.PY3:
|
||||
long = int
|
||||
@ -580,135 +579,6 @@ class PollMixinTests(unittest.TestCase):
|
||||
return d
|
||||
|
||||
|
||||
class Abbreviate(unittest.TestCase):
|
||||
def test_abbrev_time_1s(self):
|
||||
diff = timedelta(seconds=1)
|
||||
s = abbreviate.abbreviate_time(diff)
|
||||
self.assertEqual('1 second ago', s)
|
||||
|
||||
def test_abbrev_time_25s(self):
|
||||
diff = timedelta(seconds=25)
|
||||
s = abbreviate.abbreviate_time(diff)
|
||||
self.assertEqual('25 seconds ago', s)
|
||||
|
||||
def test_abbrev_time_future_5_minutes(self):
|
||||
diff = timedelta(minutes=-5)
|
||||
s = abbreviate.abbreviate_time(diff)
|
||||
self.assertEqual('5 minutes in the future', s)
|
||||
|
||||
def test_abbrev_time_hours(self):
|
||||
diff = timedelta(hours=4)
|
||||
s = abbreviate.abbreviate_time(diff)
|
||||
self.assertEqual('4 hours ago', s)
|
||||
|
||||
def test_abbrev_time_day(self):
|
||||
diff = timedelta(hours=49) # must be more than 2 days
|
||||
s = abbreviate.abbreviate_time(diff)
|
||||
self.assertEqual('2 days ago', s)
|
||||
|
||||
def test_abbrev_time_month(self):
|
||||
diff = timedelta(days=91)
|
||||
s = abbreviate.abbreviate_time(diff)
|
||||
self.assertEqual('3 months ago', s)
|
||||
|
||||
def test_abbrev_time_year(self):
|
||||
diff = timedelta(weeks=(5 * 52) + 1)
|
||||
s = abbreviate.abbreviate_time(diff)
|
||||
self.assertEqual('5 years ago', s)
|
||||
|
||||
def test_time(self):
|
||||
a = abbreviate.abbreviate_time
|
||||
self.failUnlessEqual(a(None), "unknown")
|
||||
self.failUnlessEqual(a(0), "0 seconds")
|
||||
self.failUnlessEqual(a(1), "1 second")
|
||||
self.failUnlessEqual(a(2), "2 seconds")
|
||||
self.failUnlessEqual(a(119), "119 seconds")
|
||||
MIN = 60
|
||||
self.failUnlessEqual(a(2*MIN), "2 minutes")
|
||||
self.failUnlessEqual(a(60*MIN), "60 minutes")
|
||||
self.failUnlessEqual(a(179*MIN), "179 minutes")
|
||||
HOUR = 60*MIN
|
||||
self.failUnlessEqual(a(180*MIN), "3 hours")
|
||||
self.failUnlessEqual(a(4*HOUR), "4 hours")
|
||||
DAY = 24*HOUR
|
||||
MONTH = 30*DAY
|
||||
self.failUnlessEqual(a(2*DAY), "2 days")
|
||||
self.failUnlessEqual(a(2*MONTH), "2 months")
|
||||
YEAR = 365*DAY
|
||||
self.failUnlessEqual(a(5*YEAR), "5 years")
|
||||
|
||||
def test_space(self):
|
||||
tests_si = [(None, "unknown"),
|
||||
(0, "0 B"),
|
||||
(1, "1 B"),
|
||||
(999, "999 B"),
|
||||
(1000, "1000 B"),
|
||||
(1023, "1023 B"),
|
||||
(1024, "1.02 kB"),
|
||||
(20*1000, "20.00 kB"),
|
||||
(1024*1024, "1.05 MB"),
|
||||
(1000*1000, "1.00 MB"),
|
||||
(1000*1000*1000, "1.00 GB"),
|
||||
(1000*1000*1000*1000, "1.00 TB"),
|
||||
(1000*1000*1000*1000*1000, "1.00 PB"),
|
||||
(1000*1000*1000*1000*1000*1000, "1.00 EB"),
|
||||
(1234567890123456789, "1.23 EB"),
|
||||
]
|
||||
for (x, expected) in tests_si:
|
||||
got = abbreviate.abbreviate_space(x, SI=True)
|
||||
self.failUnlessEqual(got, expected)
|
||||
|
||||
tests_base1024 = [(None, "unknown"),
|
||||
(0, "0 B"),
|
||||
(1, "1 B"),
|
||||
(999, "999 B"),
|
||||
(1000, "1000 B"),
|
||||
(1023, "1023 B"),
|
||||
(1024, "1.00 kiB"),
|
||||
(20*1024, "20.00 kiB"),
|
||||
(1000*1000, "976.56 kiB"),
|
||||
(1024*1024, "1.00 MiB"),
|
||||
(1024*1024*1024, "1.00 GiB"),
|
||||
(1024*1024*1024*1024, "1.00 TiB"),
|
||||
(1000*1000*1000*1000*1000, "909.49 TiB"),
|
||||
(1024*1024*1024*1024*1024, "1.00 PiB"),
|
||||
(1024*1024*1024*1024*1024*1024, "1.00 EiB"),
|
||||
(1234567890123456789, "1.07 EiB"),
|
||||
]
|
||||
for (x, expected) in tests_base1024:
|
||||
got = abbreviate.abbreviate_space(x, SI=False)
|
||||
self.failUnlessEqual(got, expected)
|
||||
|
||||
self.failUnlessEqual(abbreviate.abbreviate_space_both(1234567),
|
||||
"(1.23 MB, 1.18 MiB)")
|
||||
|
||||
def test_parse_space(self):
|
||||
p = abbreviate.parse_abbreviated_size
|
||||
self.failUnlessEqual(p(""), None)
|
||||
self.failUnlessEqual(p(None), None)
|
||||
self.failUnlessEqual(p("123"), 123)
|
||||
self.failUnlessEqual(p("123B"), 123)
|
||||
self.failUnlessEqual(p("2K"), 2000)
|
||||
self.failUnlessEqual(p("2kb"), 2000)
|
||||
self.failUnlessEqual(p("2KiB"), 2048)
|
||||
self.failUnlessEqual(p("10MB"), 10*1000*1000)
|
||||
self.failUnlessEqual(p("10MiB"), 10*1024*1024)
|
||||
self.failUnlessEqual(p("5G"), 5*1000*1000*1000)
|
||||
self.failUnlessEqual(p("4GiB"), 4*1024*1024*1024)
|
||||
self.failUnlessEqual(p("3TB"), 3*1000*1000*1000*1000)
|
||||
self.failUnlessEqual(p("3TiB"), 3*1024*1024*1024*1024)
|
||||
self.failUnlessEqual(p("6PB"), 6*1000*1000*1000*1000*1000)
|
||||
self.failUnlessEqual(p("6PiB"), 6*1024*1024*1024*1024*1024)
|
||||
self.failUnlessEqual(p("9EB"), 9*1000*1000*1000*1000*1000*1000)
|
||||
self.failUnlessEqual(p("9EiB"), 9*1024*1024*1024*1024*1024*1024)
|
||||
|
||||
e = self.failUnlessRaises(ValueError, p, "12 cubits")
|
||||
self.failUnlessIn("12 cubits", str(e))
|
||||
e = self.failUnlessRaises(ValueError, p, "1 BB")
|
||||
self.failUnlessIn("1 BB", str(e))
|
||||
e = self.failUnlessRaises(ValueError, p, "fhtagn")
|
||||
self.failUnlessIn("fhtagn", str(e))
|
||||
|
||||
class Limiter(unittest.TestCase):
|
||||
|
||||
def job(self, i, foo):
|
||||
@ -783,141 +653,6 @@ class Limiter(unittest.TestCase):
|
||||
d.addCallback(_all_done)
|
||||
return d
|
||||
|
||||
class TimeFormat(unittest.TestCase, TimezoneMixin):
|
||||
def test_epoch(self):
|
||||
return self._help_test_epoch()
|
||||
|
||||
def test_epoch_in_London(self):
|
||||
# Europe/London is a particularly troublesome timezone. Nowadays, its
|
||||
# offset from GMT is 0. But in 1970, its offset from GMT was 1.
|
||||
# (Apparently in 1970 Britain had redefined standard time to be GMT+1
|
||||
# and stayed in standard time all year round, whereas today
|
||||
# Europe/London standard time is GMT and Europe/London Daylight
|
||||
# Savings Time is GMT+1.) The current implementation of
|
||||
# time_format.iso_utc_time_to_localseconds() breaks if the timezone is
|
||||
# Europe/London. (As soon as this unit test is done then I'll change
|
||||
# that implementation to something that works even in this case...)
|
||||
|
||||
if not self.have_working_tzset():
|
||||
raise unittest.SkipTest("This test can't be run on a platform without time.tzset().")
|
||||
|
||||
self.setTimezone("Europe/London")
|
||||
return self._help_test_epoch()
|
||||
|
||||
def _help_test_epoch(self):
|
||||
origtzname = time.tzname
|
||||
s = time_format.iso_utc_time_to_seconds("1970-01-01T00:00:01")
|
||||
self.failUnlessEqual(s, 1.0)
|
||||
s = time_format.iso_utc_time_to_seconds("1970-01-01_00:00:01")
|
||||
self.failUnlessEqual(s, 1.0)
|
||||
s = time_format.iso_utc_time_to_seconds("1970-01-01 00:00:01")
|
||||
self.failUnlessEqual(s, 1.0)
|
||||
|
||||
self.failUnlessEqual(time_format.iso_utc(1.0), "1970-01-01_00:00:01")
|
||||
self.failUnlessEqual(time_format.iso_utc(1.0, sep=" "),
|
||||
"1970-01-01 00:00:01")
|
||||
|
||||
now = time.time()
|
||||
isostr = time_format.iso_utc(now)
|
||||
timestamp = time_format.iso_utc_time_to_seconds(isostr)
|
||||
self.failUnlessEqual(int(timestamp), int(now))
|
||||
|
||||
def my_time():
|
||||
return 1.0
|
||||
self.failUnlessEqual(time_format.iso_utc(t=my_time),
|
||||
"1970-01-01_00:00:01")
|
||||
e = self.failUnlessRaises(ValueError,
|
||||
time_format.iso_utc_time_to_seconds,
|
||||
"invalid timestring")
|
||||
self.failUnless("not a complete ISO8601 timestamp" in str(e))
|
||||
s = time_format.iso_utc_time_to_seconds("1970-01-01_00:00:01.500")
|
||||
self.failUnlessEqual(s, 1.5)
|
||||
|
||||
# Look for daylight-savings-related errors.
|
||||
thatmomentinmarch = time_format.iso_utc_time_to_seconds("2009-03-20 21:49:02.226536")
|
||||
self.failUnlessEqual(thatmomentinmarch, 1237585742.226536)
|
||||
self.failUnlessEqual(origtzname, time.tzname)
|
||||
|
||||
def test_iso_utc(self):
|
||||
when = 1266760143.7841301
|
||||
out = time_format.iso_utc_date(when)
|
||||
self.failUnlessEqual(out, "2010-02-21")
|
||||
out = time_format.iso_utc_date(t=lambda: when)
|
||||
self.failUnlessEqual(out, "2010-02-21")
|
||||
out = time_format.iso_utc(when)
|
||||
self.failUnlessEqual(out, "2010-02-21_13:49:03.784130")
|
||||
out = time_format.iso_utc(when, sep="-")
|
||||
self.failUnlessEqual(out, "2010-02-21-13:49:03.784130")
|
||||
|
||||
def test_parse_duration(self):
|
||||
p = time_format.parse_duration
|
||||
DAY = 24*60*60
|
||||
self.failUnlessEqual(p("1 day"), DAY)
|
||||
self.failUnlessEqual(p("2 days"), 2*DAY)
|
||||
self.failUnlessEqual(p("3 months"), 3*31*DAY)
|
||||
self.failUnlessEqual(p("4 mo"), 4*31*DAY)
|
||||
self.failUnlessEqual(p("5 years"), 5*365*DAY)
|
||||
e = self.failUnlessRaises(ValueError, p, "123")
|
||||
self.failUnlessIn("no unit (like day, month, or year) in '123'",
|
||||
str(e))
|
||||
|
||||
def test_parse_date(self):
|
||||
self.failUnlessEqual(time_format.parse_date("2010-02-21"), 1266710400)
|
||||
|
||||
def test_format_time(self):
|
||||
self.failUnlessEqual(time_format.format_time(time.gmtime(0)), '1970-01-01 00:00:00')
|
||||
self.failUnlessEqual(time_format.format_time(time.gmtime(60)), '1970-01-01 00:01:00')
|
||||
self.failUnlessEqual(time_format.format_time(time.gmtime(60*60)), '1970-01-01 01:00:00')
|
||||
seconds_per_day = 60*60*24
|
||||
leap_years_1970_to_2014_inclusive = ((2012 - 1968) // 4)
|
||||
self.failUnlessEqual(time_format.format_time(time.gmtime(seconds_per_day*((2015 - 1970)*365+leap_years_1970_to_2014_inclusive))), '2015-01-01 00:00:00')
|
||||
|
||||
def test_format_time_y2038(self):
|
||||
seconds_per_day = 60*60*24
|
||||
leap_years_1970_to_2047_inclusive = ((2044 - 1968) // 4)
|
||||
t = (seconds_per_day*
|
||||
((2048 - 1970)*365 + leap_years_1970_to_2047_inclusive))
|
||||
try:
|
||||
gm_t = time.gmtime(t)
|
||||
except ValueError:
|
||||
raise unittest.SkipTest("Note: this system cannot handle dates after 2037.")
|
||||
self.failUnlessEqual(time_format.format_time(gm_t),
|
||||
'2048-01-01 00:00:00')
|
||||
|
||||
def test_format_delta(self):
|
||||
time_1 = 1389812723
|
||||
time_5s_delta = 1389812728
|
||||
time_28m7s_delta = 1389814410
|
||||
time_1h_delta = 1389816323
|
||||
time_1d21h46m49s_delta = 1389977532
|
||||
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1, time_1), '0s')
|
||||
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1, time_5s_delta), '5s')
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1, time_28m7s_delta), '28m 7s')
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1, time_1h_delta), '1h 0m 0s')
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1, time_1d21h46m49s_delta), '1d 21h 46m 49s')
|
||||
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1d21h46m49s_delta, time_1), '-')
|
||||
|
||||
# time_1 with a decimal fraction will make the delta 1s less
|
||||
time_1decimal = 1389812723.383963
|
||||
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1decimal, time_5s_delta), '4s')
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1decimal, time_28m7s_delta), '28m 6s')
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1decimal, time_1h_delta), '59m 59s')
|
||||
self.failUnlessEqual(
|
||||
time_format.format_delta(time_1decimal, time_1d21h46m49s_delta), '1d 21h 46m 48s')
|
||||
|
||||
|
||||
ctr = [0]
|
||||
class EqButNotIs(object):
|
||||
|
||||
@ -59,6 +59,7 @@ from .common import (
|
||||
from allmydata.interfaces import IMutableFileNode, SDMF_VERSION, MDMF_VERSION
|
||||
from allmydata.mutable import servermap, publish, retrieve
|
||||
from .. import common_util as testutil
|
||||
from ..common_py3 import TimezoneMixin
|
||||
from ..common_web import (
|
||||
do_http,
|
||||
Error,
|
||||
@ -310,7 +311,7 @@ class FakeClient(_Client):
|
||||
|
||||
MUTABLE_SIZELIMIT = FakeMutableFileNode.MUTABLE_SIZELIMIT
|
||||
|
||||
class WebMixin(testutil.TimezoneMixin):
|
||||
class WebMixin(TimezoneMixin):
|
||||
def setUp(self):
|
||||
self.setTimezone('UTC-13:00')
|
||||
self.s = FakeClient()
|
||||
|
||||
@ -16,6 +16,7 @@ if PY2:
|
||||
# Keep these sorted alphabetically, to reduce merge conflicts:
|
||||
PORTED_MODULES = [
|
||||
"allmydata.hashtree",
|
||||
"allmydata.util.abbreviate",
|
||||
"allmydata.util.assertutil",
|
||||
"allmydata.util.base32",
|
||||
"allmydata.util.base62",
|
||||
@ -29,9 +30,12 @@ PORTED_MODULES = [
|
||||
"allmydata.util.pipeline",
|
||||
"allmydata.util.pollmixin",
|
||||
"allmydata.util._python3",
|
||||
"allmydata.util.time_format",
|
||||
"allmydata.test.common_py3",
|
||||
]
|
||||
|
||||
PORTED_TEST_MODULES = [
|
||||
"allmydata.test.test_abbreviate",
|
||||
"allmydata.test.test_base32",
|
||||
"allmydata.test.test_base62",
|
||||
"allmydata.test.test_deferredutil",
|
||||
@ -42,6 +46,7 @@ PORTED_TEST_MODULES = [
|
||||
"allmydata.test.test_observer",
|
||||
"allmydata.test.test_pipeline",
|
||||
"allmydata.test.test_python3",
|
||||
"allmydata.test.test_time_format",
|
||||
]
|
||||
|
||||
|
||||
|
||||
@ -1,3 +1,16 @@
|
||||
"""
|
||||
Convert timestamps to abbreviated English text.
|
||||
|
||||
Ported to Python 3.
|
||||
"""
|
||||
from __future__ import division
|
||||
from __future__ import absolute_import
|
||||
from __future__ import print_function
|
||||
from __future__ import unicode_literals
|
||||
|
||||
from future.utils import PY2
|
||||
if PY2:
|
||||
from builtins import filter, map, zip, ascii, chr, hex, input, next, oct, open, pow, round, super, bytes, dict, int, list, object, range, str, max, min # noqa: F401
|
||||
|
||||
import re
|
||||
from datetime import timedelta
|
||||
@ -9,6 +22,10 @@ MONTH = 30*DAY
|
||||
YEAR = 365*DAY
|
||||
|
||||
def abbreviate_time(s):
|
||||
"""
|
||||
Given time in seconds (float or int) or timedelta, summarize as English by
|
||||
returning unicode string.
|
||||
"""
|
||||
postfix = ''
|
||||
if isinstance(s, timedelta):
|
||||
# this feels counter-intuitive that positive numbers in a
|
||||
@ -45,6 +62,9 @@ def abbreviate_time(s):
|
||||
return _plural(s / YEAR, "year")
|
||||
|
||||
def abbreviate_space(s, SI=True):
|
||||
"""
|
||||
Given size in bytes summarize as English by returning unicode string.
|
||||
"""
|
||||
if s is None:
|
||||
return "unknown"
|
||||
if SI:
|
||||
|
||||
@ -1,5 +1,18 @@
|
||||
# ISO-8601:
|
||||
# http://www.cl.cam.ac.uk/~mgk25/iso-time.html
|
||||
"""
|
||||
Time formatting utilities.
|
||||
|
||||
ISO-8601:
|
||||
http://www.cl.cam.ac.uk/~mgk25/iso-time.html
|
||||
"""
|
||||
from __future__ import absolute_import
|
||||
from __future__ import division
|
||||
from __future__ import print_function
|
||||
from __future__ import unicode_literals
|
||||
|
||||
from future.utils import PY2
|
||||
if PY2:
|
||||
from builtins import filter, map, zip, ascii, chr, hex, input, next, oct, open, pow, round, super, bytes, dict, int, list, object, range, str, max, min # noqa: F401
|
||||
from future.utils import native_str
|
||||
|
||||
import calendar, datetime, re, time
|
||||
|
||||
@ -14,6 +27,7 @@ def iso_utc_date(now=None, t=time.time):
|
||||
def iso_utc(now=None, sep='_', t=time.time):
|
||||
if now is None:
|
||||
now = t()
|
||||
sep = native_str(sep) # Python 2 doesn't allow unicode input to isoformat
|
||||
return datetime.datetime.utcfromtimestamp(now).isoformat(sep)
|
||||
|
||||
def iso_utc_time_to_seconds(isotime, _conversion_re=re.compile(r"(?P<year>\d{4})-(?P<month>\d{2})-(?P<day>\d{2})[T_ ](?P<hour>\d{2}):(?P<minute>\d{2}):(?P<second>\d{2})(?P<subsecond>\.\d+)?")):
|
||||
@ -74,11 +88,11 @@ def format_delta(time_1, time_2):
|
||||
delta = int(time_2 - time_1)
|
||||
seconds = delta % 60
|
||||
delta -= seconds
|
||||
minutes = (delta / 60) % 60
|
||||
minutes = (delta // 60) % 60
|
||||
delta -= minutes * 60
|
||||
hours = delta / (60*60) % 24
|
||||
hours = delta // (60*60) % 24
|
||||
delta -= hours * 24
|
||||
days = delta / (24*60*60)
|
||||
days = delta // (24*60*60)
|
||||
if not days:
|
||||
if not hours:
|
||||
if not minutes:
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user