ExternalVendorCode/tahoe-lafs
1
0
Fork 0
mirror of https://github.com/tahoe-lafs/tahoe-lafs.git synced 2025-01-29 15:43:54 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into 3357.version_checks-py3

Browse Source
This commit is contained in:
Itamar Turner-Trauring 2020-07-24 14:50:39 -04:00
commit 11cd30c425
17 changed files with 678 additions and 498 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
misc/python3/ratchet-passing
View File

@ -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
@ -42,9 +52,20 @@ allmydata.test.test_observer.Observer.test_lazy_oneshot
allmydata.test.test_observer.Observer.test_observerlist
allmydata.test.test_observer.Observer.test_oneshot
allmydata.test.test_observer.Observer.test_oneshot_fireagain
allmydata.test.test_pipeline.Pipeline.test_basic
allmydata.test.test_pipeline.Pipeline.test_errors
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
allmydata.test.test_version.CheckRequirement.test_cross_check
allmydata.test.test_version.CheckRequirement.test_cross_check_unparseable_versions
allmydata.test.test_version.CheckRequirement.test_extract_openssl_version

0
newsfragments/3346.minor Normal file
View File

0
newsfragments/3353.minor Normal file
View File

42
src/allmydata/test/common_py3.py Normal file
View 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')

27
src/allmydata/test/common_util.py
View File

@ -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
View 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))

14
src/allmydata/test/test_observer.py
View File

@ -1,3 +1,17 @@
"""
Tests for allmydata.util.observer.
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 twisted.trial import unittest
from twisted.internet import defer, reactor

198
src/allmydata/test/test_pipeline.py Normal file
View File

@ -0,0 +1,198 @@
"""
Tests for allmydata.util.pipeline.
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
import gc
from twisted.internet import defer
from twisted.trial import unittest
from twisted.python import log
from twisted.python.failure import Failure
from allmydata.util import pipeline
class Pipeline(unittest.TestCase):
def pause(self, *args, **kwargs):
d = defer.Deferred()
self.calls.append( (d, args, kwargs) )
return d
def failUnlessCallsAre(self, expected):
#print self.calls
#print expected
self.failUnlessEqual(len(self.calls), len(expected), self.calls)
for i,c in enumerate(self.calls):
self.failUnlessEqual(c[1:], expected[i], str(i))
def test_basic(self):
self.calls = []
finished = []
p = pipeline.Pipeline(100)
d = p.flush() # fires immediately
d.addCallbacks(finished.append, log.err)
self.failUnlessEqual(len(finished), 1)
finished = []
d = p.add(10, self.pause, "one")
# the call should start right away, and our return Deferred should
# fire right away
d.addCallbacks(finished.append, log.err)
self.failUnlessEqual(len(finished), 1)
self.failUnlessEqual(finished[0], None)
self.failUnlessCallsAre([ ( ("one",) , {} ) ])
self.failUnlessEqual(p.gauge, 10)
# pipeline: [one]
finished = []
d = p.add(20, self.pause, "two", kw=2)
# pipeline: [one, two]
# the call and the Deferred should fire right away
d.addCallbacks(finished.append, log.err)
self.failUnlessEqual(len(finished), 1)
self.failUnlessEqual(finished[0], None)
self.failUnlessCallsAre([ ( ("one",) , {} ),
( ("two",) , {"kw": 2} ),
])
self.failUnlessEqual(p.gauge, 30)
self.calls[0][0].callback("one-result")
# pipeline: [two]
self.failUnlessEqual(p.gauge, 20)
finished = []
d = p.add(90, self.pause, "three", "posarg1")
# pipeline: [two, three]
flushed = []
fd = p.flush()
fd.addCallbacks(flushed.append, log.err)
self.failUnlessEqual(flushed, [])
# the call will be made right away, but the return Deferred will not,
# because the pipeline is now full.
d.addCallbacks(finished.append, log.err)
self.failUnlessEqual(len(finished), 0)
self.failUnlessCallsAre([ ( ("one",) , {} ),
( ("two",) , {"kw": 2} ),
( ("three", "posarg1"), {} ),
])
self.failUnlessEqual(p.gauge, 110)
self.failUnlessRaises(pipeline.SingleFileError, p.add, 10, self.pause)
# retiring either call will unblock the pipeline, causing the #3
# Deferred to fire
self.calls[2][0].callback("three-result")
# pipeline: [two]
self.failUnlessEqual(len(finished), 1)
self.failUnlessEqual(finished[0], None)
self.failUnlessEqual(flushed, [])
# retiring call#2 will finally allow the flush() Deferred to fire
self.calls[1][0].callback("two-result")
self.failUnlessEqual(len(flushed), 1)
def test_errors(self):
self.calls = []
p = pipeline.Pipeline(100)
d1 = p.add(200, self.pause, "one")
d2 = p.flush()
finished = []
d1.addBoth(finished.append)
self.failUnlessEqual(finished, [])
flushed = []
d2.addBoth(flushed.append)
self.failUnlessEqual(flushed, [])
self.calls[0][0].errback(ValueError("oops"))
self.failUnlessEqual(len(finished), 1)
f = finished[0]
self.failUnless(isinstance(f, Failure))
self.failUnless(f.check(pipeline.PipelineError))
self.failUnlessIn("PipelineError", str(f.value))
self.failUnlessIn("ValueError", str(f.value))
r = repr(f.value)
self.failUnless("ValueError" in r, r)
f2 = f.value.error
self.failUnless(f2.check(ValueError))
self.failUnlessEqual(len(flushed), 1)
f = flushed[0]
self.failUnless(isinstance(f, Failure))
self.failUnless(f.check(pipeline.PipelineError))
f2 = f.value.error
self.failUnless(f2.check(ValueError))
# now that the pipeline is in the failed state, any new calls will
# fail immediately
d3 = p.add(20, self.pause, "two")
finished = []
d3.addBoth(finished.append)
self.failUnlessEqual(len(finished), 1)
f = finished[0]
self.failUnless(isinstance(f, Failure))
self.failUnless(f.check(pipeline.PipelineError))
r = repr(f.value)
self.failUnless("ValueError" in r, r)
f2 = f.value.error
self.failUnless(f2.check(ValueError))
d4 = p.flush()
flushed = []
d4.addBoth(flushed.append)
self.failUnlessEqual(len(flushed), 1)
f = flushed[0]
self.failUnless(isinstance(f, Failure))
self.failUnless(f.check(pipeline.PipelineError))
f2 = f.value.error
self.failUnless(f2.check(ValueError))
def test_errors2(self):
self.calls = []
p = pipeline.Pipeline(100)
d1 = p.add(10, self.pause, "one")
d2 = p.add(20, self.pause, "two")
d3 = p.add(30, self.pause, "three")
d4 = p.flush()
# one call fails, then the second one succeeds: make sure
# ExpandableDeferredList tolerates the second one
flushed = []
d4.addBoth(flushed.append)
self.failUnlessEqual(flushed, [])
self.calls[0][0].errback(ValueError("oops"))
self.failUnlessEqual(len(flushed), 1)
f = flushed[0]
self.failUnless(isinstance(f, Failure))
self.failUnless(f.check(pipeline.PipelineError))
f2 = f.value.error
self.failUnless(f2.check(ValueError))
self.calls[1][0].callback("two-result")
self.calls[2][0].errback(ValueError("three-error"))
del d1,d2,d3,d4
gc.collect() # for PyPy

21
src/allmydata/test/test_storage.py
View File

@ -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
View 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')

449
src/allmydata/test/test_util.py
View File

@ -5,23 +5,20 @@ import six
import hashlib
import os, time, sys
import yaml
import gc # support PyPy
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 twisted.python import log
from allmydata.util import idlib, mathutil
from allmydata.util import fileutil, abbreviate
from allmydata.util import limiter, time_format, pollmixin
from allmydata.util import statistics, dictutil, pipeline, yamlutil
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
@ -582,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):
@ -785,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):
@ -1016,180 +749,6 @@ class DictUtil(unittest.TestCase):
self.failUnlessEqual(d["one"], 1)
self.failUnlessEqual(d.get_aux("one"), None)
class Pipeline(unittest.TestCase):
def pause(self, *args, **kwargs):
d = defer.Deferred()
self.calls.append( (d, args, kwargs) )
return d
def failUnlessCallsAre(self, expected):
#print self.calls
#print expected
self.failUnlessEqual(len(self.calls), len(expected), self.calls)
for i,c in enumerate(self.calls):
self.failUnlessEqual(c[1:], expected[i], str(i))
def test_basic(self):
self.calls = []
finished = []
p = pipeline.Pipeline(100)
d = p.flush() # fires immediately
d.addCallbacks(finished.append, log.err)
self.failUnlessEqual(len(finished), 1)
finished = []
d = p.add(10, self.pause, "one")
# the call should start right away, and our return Deferred should
# fire right away
d.addCallbacks(finished.append, log.err)
self.failUnlessEqual(len(finished), 1)
self.failUnlessEqual(finished[0], None)
self.failUnlessCallsAre([ ( ("one",) , {} ) ])
self.failUnlessEqual(p.gauge, 10)
# pipeline: [one]
finished = []
d = p.add(20, self.pause, "two", kw=2)
# pipeline: [one, two]
# the call and the Deferred should fire right away
d.addCallbacks(finished.append, log.err)
self.failUnlessEqual(len(finished), 1)
self.failUnlessEqual(finished[0], None)
self.failUnlessCallsAre([ ( ("one",) , {} ),
( ("two",) , {"kw": 2} ),
])
self.failUnlessEqual(p.gauge, 30)
self.calls[0][0].callback("one-result")
# pipeline: [two]
self.failUnlessEqual(p.gauge, 20)
finished = []
d = p.add(90, self.pause, "three", "posarg1")
# pipeline: [two, three]
flushed = []
fd = p.flush()
fd.addCallbacks(flushed.append, log.err)
self.failUnlessEqual(flushed, [])
# the call will be made right away, but the return Deferred will not,
# because the pipeline is now full.
d.addCallbacks(finished.append, log.err)
self.failUnlessEqual(len(finished), 0)
self.failUnlessCallsAre([ ( ("one",) , {} ),
( ("two",) , {"kw": 2} ),
( ("three", "posarg1"), {} ),
])
self.failUnlessEqual(p.gauge, 110)
self.failUnlessRaises(pipeline.SingleFileError, p.add, 10, self.pause)
# retiring either call will unblock the pipeline, causing the #3
# Deferred to fire
self.calls[2][0].callback("three-result")
# pipeline: [two]
self.failUnlessEqual(len(finished), 1)
self.failUnlessEqual(finished[0], None)
self.failUnlessEqual(flushed, [])
# retiring call#2 will finally allow the flush() Deferred to fire
self.calls[1][0].callback("two-result")
self.failUnlessEqual(len(flushed), 1)
def test_errors(self):
self.calls = []
p = pipeline.Pipeline(100)
d1 = p.add(200, self.pause, "one")
d2 = p.flush()
finished = []
d1.addBoth(finished.append)
self.failUnlessEqual(finished, [])
flushed = []
d2.addBoth(flushed.append)
self.failUnlessEqual(flushed, [])
self.calls[0][0].errback(ValueError("oops"))
self.failUnlessEqual(len(finished), 1)
f = finished[0]
self.failUnless(isinstance(f, Failure))
self.failUnless(f.check(pipeline.PipelineError))
self.failUnlessIn("PipelineError", str(f.value))
self.failUnlessIn("ValueError", str(f.value))
r = repr(f.value)
self.failUnless("ValueError" in r, r)
f2 = f.value.error
self.failUnless(f2.check(ValueError))
self.failUnlessEqual(len(flushed), 1)
f = flushed[0]
self.failUnless(isinstance(f, Failure))
self.failUnless(f.check(pipeline.PipelineError))
f2 = f.value.error
self.failUnless(f2.check(ValueError))
# now that the pipeline is in the failed state, any new calls will
# fail immediately
d3 = p.add(20, self.pause, "two")
finished = []
d3.addBoth(finished.append)
self.failUnlessEqual(len(finished), 1)
f = finished[0]
self.failUnless(isinstance(f, Failure))
self.failUnless(f.check(pipeline.PipelineError))
r = repr(f.value)
self.failUnless("ValueError" in r, r)
f2 = f.value.error
self.failUnless(f2.check(ValueError))
d4 = p.flush()
flushed = []
d4.addBoth(flushed.append)
self.failUnlessEqual(len(flushed), 1)
f = flushed[0]
self.failUnless(isinstance(f, Failure))
self.failUnless(f.check(pipeline.PipelineError))
f2 = f.value.error
self.failUnless(f2.check(ValueError))
def test_errors2(self):
self.calls = []
p = pipeline.Pipeline(100)
d1 = p.add(10, self.pause, "one")
d2 = p.add(20, self.pause, "two")
d3 = p.add(30, self.pause, "three")
d4 = p.flush()
# one call fails, then the second one succeeds: make sure
# ExpandableDeferredList tolerates the second one
flushed = []
d4.addBoth(flushed.append)
self.failUnlessEqual(flushed, [])
self.calls[0][0].errback(ValueError("oops"))
self.failUnlessEqual(len(flushed), 1)
f = flushed[0]
self.failUnless(isinstance(f, Failure))
self.failUnless(f.check(pipeline.PipelineError))
f2 = f.value.error
self.failUnless(f2.check(ValueError))
self.calls[1][0].callback("two-result")
self.calls[2][0].errback(ValueError("three-error"))
del d1,d2,d3,d4
gc.collect() # for PyPy
class SampleError(Exception):
pass

3
src/allmydata/test/web/test_web.py
View File

@ -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()

9
src/allmydata/util/_python3.py
View File

@ -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",
@ -25,11 +26,16 @@ PORTED_MODULES = [
"allmydata.util.mathutil",
"allmydata.util.namespace",
"allmydata.util.netstring",
"allmydata.util.observer",
"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",
@ -37,7 +43,10 @@ PORTED_TEST_MODULES = [
"allmydata.test.test_hashutil",
"allmydata.test.test_humanreadable",
"allmydata.test.test_netstring",
"allmydata.test.test_observer",
"allmydata.test.test_pipeline",
"allmydata.test.test_python3",
"allmydata.test.test_time_format",
"allmydata.test.test_version",
]

20
src/allmydata/util/abbreviate.py
View File

@ -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:

15
src/allmydata/util/observer.py
View File

@ -1,4 +1,17 @@
# -*- test-case-name: allmydata.test.test_observer -*-
"""
Observer for Twisted code.
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
import weakref
from twisted.internet import defer

15
src/allmydata/util/pipeline.py
View File

@ -1,9 +1,24 @@
"""
A pipeline of Deferreds.
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 twisted.internet import defer
from twisted.python.failure import Failure
from twisted.python import log
from allmydata.util.assertutil import precondition
class PipelineError(Exception):
"""One of the pipelined messages returned an error. The received Failure
object is stored in my .error attribute."""

24
src/allmydata/util/time_format.py
View File

@ -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: