src/allmydata/test/test_util.py

   1
   2 def foo(): pass # keep the line number constant
   3
   4 import os, time
   5 from twisted.trial import unittest
   6 from twisted.internet import defer, reactor
   7 from twisted.python import failure
   8
   9 from allmydata.util import base32, idlib, humanreadable, mathutil, hashutil
  10 from allmydata.util import assertutil, fileutil, deferredutil, abbreviate
  11 from allmydata.util import limiter, time_format, pollmixin, cachedir
  12
  13 class Base32(unittest.TestCase):
  14     def test_b2a_matches_Pythons(self):
  15         import base64
  16         y = "\x12\x34\x45\x67\x89\x0a\xbc\xde\xf0"
  17         x = base64.b32encode(y)
  18         while x and x[-1] == '=':
  19             x = x[:-1]
  20         x = x.lower()
  21         self.failUnlessEqual(base32.b2a(y), x)
  22     def test_b2a(self):
  23         self.failUnlessEqual(base32.b2a("\x12\x34"), "ci2a")
  24     def test_b2a_or_none(self):
  25         self.failUnlessEqual(base32.b2a_or_none(None), None)
  26         self.failUnlessEqual(base32.b2a_or_none("\x12\x34"), "ci2a")
  27     def test_a2b(self):
  28         self.failUnlessEqual(base32.a2b("ci2a"), "\x12\x34")
  29         self.failUnlessRaises(AssertionError, base32.a2b, "b0gus")
  30
  31 class IDLib(unittest.TestCase):
  32     def test_nodeid_b2a(self):
  33         self.failUnlessEqual(idlib.nodeid_b2a("\x00"*20), "a"*32)
  34
  35 class NoArgumentException(Exception):
  36     def __init__(self):
  37         pass
  38
  39 class HumanReadable(unittest.TestCase):
  40     def test_repr(self):
  41         hr = humanreadable.hr
  42         self.failUnlessEqual(hr(foo), "<foo() at test_util.py:2>")
  43         self.failUnlessEqual(hr(self.test_repr),
  44                              "<bound method HumanReadable.test_repr of <allmydata.test.test_util.HumanReadable testMethod=test_repr>>")
  45         self.failUnlessEqual(hr(1L), "1")
  46         self.failUnlessEqual(hr(10**40),
  47                              "100000000000000000...000000000000000000")
  48         self.failUnlessEqual(hr(self), "<allmydata.test.test_util.HumanReadable testMethod=test_repr>")
  49         self.failUnlessEqual(hr([1,2]), "[1, 2]")
  50         self.failUnlessEqual(hr({1:2}), "{1:2}")
  51         try:
  52             raise RuntimeError
  53         except Exception, e:
  54             self.failUnless(
  55                 hr(e) == "<RuntimeError: ()>" # python-2.4
  56                 or hr(e) == "RuntimeError()") # python-2.5
  57         try:
  58             raise RuntimeError("oops")
  59         except Exception, e:
  60             self.failUnless(
  61                 hr(e) == "<RuntimeError: 'oops'>" # python-2.4
  62                 or hr(e) == "RuntimeError('oops',)") # python-2.5
  63         try:
  64             raise NoArgumentException
  65         except Exception, e:
  66             self.failUnless(
  67                 hr(e) == "<NoArgumentException>" # python-2.4
  68                 or hr(e) == "NoArgumentException()") # python-2.5
  69
  70
  71 class MyList(list):
  72     pass
  73
  74 class Math(unittest.TestCase):
  75     def test_div_ceil(self):
  76         f = mathutil.div_ceil
  77         self.failUnlessEqual(f(0, 1), 0)
  78         self.failUnlessEqual(f(0, 2), 0)
  79         self.failUnlessEqual(f(0, 3), 0)
  80         self.failUnlessEqual(f(1, 3), 1)
  81         self.failUnlessEqual(f(2, 3), 1)
  82         self.failUnlessEqual(f(3, 3), 1)
  83         self.failUnlessEqual(f(4, 3), 2)
  84         self.failUnlessEqual(f(5, 3), 2)
  85         self.failUnlessEqual(f(6, 3), 2)
  86         self.failUnlessEqual(f(7, 3), 3)
  87
  88     def test_next_multiple(self):
  89         f = mathutil.next_multiple
  90         self.failUnlessEqual(f(5, 1), 5)
  91         self.failUnlessEqual(f(5, 2), 6)
  92         self.failUnlessEqual(f(5, 3), 6)
  93         self.failUnlessEqual(f(5, 4), 8)
  94         self.failUnlessEqual(f(5, 5), 5)
  95         self.failUnlessEqual(f(5, 6), 6)
  96         self.failUnlessEqual(f(32, 1), 32)
  97         self.failUnlessEqual(f(32, 2), 32)
  98         self.failUnlessEqual(f(32, 3), 33)
  99         self.failUnlessEqual(f(32, 4), 32)
 100         self.failUnlessEqual(f(32, 5), 35)
 101         self.failUnlessEqual(f(32, 6), 36)
 102         self.failUnlessEqual(f(32, 7), 35)
 103         self.failUnlessEqual(f(32, 8), 32)
 104         self.failUnlessEqual(f(32, 9), 36)
 105         self.failUnlessEqual(f(32, 10), 40)
 106         self.failUnlessEqual(f(32, 11), 33)
 107         self.failUnlessEqual(f(32, 12), 36)
 108         self.failUnlessEqual(f(32, 13), 39)
 109         self.failUnlessEqual(f(32, 14), 42)
 110         self.failUnlessEqual(f(32, 15), 45)
 111         self.failUnlessEqual(f(32, 16), 32)
 112         self.failUnlessEqual(f(32, 17), 34)
 113         self.failUnlessEqual(f(32, 18), 36)
 114         self.failUnlessEqual(f(32, 589), 589)
 115
 116     def test_pad_size(self):
 117         f = mathutil.pad_size
 118         self.failUnlessEqual(f(0, 4), 0)
 119         self.failUnlessEqual(f(1, 4), 3)
 120         self.failUnlessEqual(f(2, 4), 2)
 121         self.failUnlessEqual(f(3, 4), 1)
 122         self.failUnlessEqual(f(4, 4), 0)
 123         self.failUnlessEqual(f(5, 4), 3)
 124
 125     def test_is_power_of_k(self):
 126         f = mathutil.is_power_of_k
 127         for i in range(1, 100):
 128             if i in (1, 2, 4, 8, 16, 32, 64):
 129                 self.failUnless(f(i, 2), "but %d *is* a power of 2" % i)
 130             else:
 131                 self.failIf(f(i, 2), "but %d is *not* a power of 2" % i)
 132         for i in range(1, 100):
 133             if i in (1, 3, 9, 27, 81):
 134                 self.failUnless(f(i, 3), "but %d *is* a power of 3" % i)
 135             else:
 136                 self.failIf(f(i, 3), "but %d is *not* a power of 3" % i)
 137
 138     def test_next_power_of_k(self):
 139         f = mathutil.next_power_of_k
 140         self.failUnlessEqual(f(0,2), 1)
 141         self.failUnlessEqual(f(1,2), 1)
 142         self.failUnlessEqual(f(2,2), 2)
 143         self.failUnlessEqual(f(3,2), 4)
 144         self.failUnlessEqual(f(4,2), 4)
 145         for i in range(5, 8): self.failUnlessEqual(f(i,2), 8, "%d" % i)
 146         for i in range(9, 16): self.failUnlessEqual(f(i,2), 16, "%d" % i)
 147         for i in range(17, 32): self.failUnlessEqual(f(i,2), 32, "%d" % i)
 148         for i in range(33, 64): self.failUnlessEqual(f(i,2), 64, "%d" % i)
 149         for i in range(65, 100): self.failUnlessEqual(f(i,2), 128, "%d" % i)
 150
 151         self.failUnlessEqual(f(0,3), 1)
 152         self.failUnlessEqual(f(1,3), 1)
 153         self.failUnlessEqual(f(2,3), 3)
 154         self.failUnlessEqual(f(3,3), 3)
 155         for i in range(4, 9): self.failUnlessEqual(f(i,3), 9, "%d" % i)
 156         for i in range(10, 27): self.failUnlessEqual(f(i,3), 27, "%d" % i)
 157         for i in range(28, 81): self.failUnlessEqual(f(i,3), 81, "%d" % i)
 158         for i in range(82, 200): self.failUnlessEqual(f(i,3), 243, "%d" % i)
 159
 160     def test_ave(self):
 161         f = mathutil.ave
 162         self.failUnlessEqual(f([1,2,3]), 2)
 163         self.failUnlessEqual(f([0,0,0,4]), 1)
 164         self.failUnlessAlmostEqual(f([0.0, 1.0, 1.0]), .666666666666)
 165
 166
 167 class Asserts(unittest.TestCase):
 168     def should_assert(self, func, *args, **kwargs):
 169         try:
 170             func(*args, **kwargs)
 171         except AssertionError, e:
 172             return str(e)
 173         except Exception, e:
 174             self.fail("assert failed with non-AssertionError: %s" % e)
 175         self.fail("assert was not caught")
 176
 177     def should_not_assert(self, func, *args, **kwargs):
 178         if "re" in kwargs:
 179             regexp = kwargs["re"]
 180             del kwargs["re"]
 181         try:
 182             func(*args, **kwargs)
 183         except AssertionError, e:
 184             self.fail("assertion fired when it should not have: %s" % e)
 185         except Exception, e:
 186             self.fail("assertion (which shouldn't have failed) failed with non-AssertionError: %s" % e)
 187         return # we're happy
 188
 189
 190     def test_assert(self):
 191         f = assertutil._assert
 192         self.should_assert(f)
 193         self.should_assert(f, False)
 194         self.should_not_assert(f, True)
 195
 196         m = self.should_assert(f, False, "message")
 197         self.failUnlessEqual(m, "'message' <type 'str'>", m)
 198         m = self.should_assert(f, False, "message1", othermsg=12)
 199         self.failUnlessEqual("'message1' <type 'str'>, othermsg: 12 <type 'int'>", m)
 200         m = self.should_assert(f, False, othermsg="message2")
 201         self.failUnlessEqual("othermsg: 'message2' <type 'str'>", m)
 202
 203     def test_precondition(self):
 204         f = assertutil.precondition
 205         self.should_assert(f)
 206         self.should_assert(f, False)
 207         self.should_not_assert(f, True)
 208
 209         m = self.should_assert(f, False, "message")
 210         self.failUnlessEqual("precondition: 'message' <type 'str'>", m)
 211         m = self.should_assert(f, False, "message1", othermsg=12)
 212         self.failUnlessEqual("precondition: 'message1' <type 'str'>, othermsg: 12 <type 'int'>", m)
 213         m = self.should_assert(f, False, othermsg="message2")
 214         self.failUnlessEqual("precondition: othermsg: 'message2' <type 'str'>", m)
 215
 216     def test_postcondition(self):
 217         f = assertutil.postcondition
 218         self.should_assert(f)
 219         self.should_assert(f, False)
 220         self.should_not_assert(f, True)
 221
 222         m = self.should_assert(f, False, "message")
 223         self.failUnlessEqual("postcondition: 'message' <type 'str'>", m)
 224         m = self.should_assert(f, False, "message1", othermsg=12)
 225         self.failUnlessEqual("postcondition: 'message1' <type 'str'>, othermsg: 12 <type 'int'>", m)
 226         m = self.should_assert(f, False, othermsg="message2")
 227         self.failUnlessEqual("postcondition: othermsg: 'message2' <type 'str'>", m)
 228
 229 class FileUtil(unittest.TestCase):
 230     def mkdir(self, basedir, path, mode=0777):
 231         fn = os.path.join(basedir, path)
 232         fileutil.make_dirs(fn, mode)
 233
 234     def touch(self, basedir, path, mode=None, data="touch\n"):
 235         fn = os.path.join(basedir, path)
 236         f = open(fn, "w")
 237         f.write(data)
 238         f.close()
 239         if mode is not None:
 240             os.chmod(fn, mode)
 241
 242     def test_rm_dir(self):
 243         basedir = "util/FileUtil/test_rm_dir"
 244         fileutil.make_dirs(basedir)
 245         # create it again to test idempotency
 246         fileutil.make_dirs(basedir)
 247         d = os.path.join(basedir, "doomed")
 248         self.mkdir(d, "a/b")
 249         self.touch(d, "a/b/1.txt")
 250         self.touch(d, "a/b/2.txt", 0444)
 251         self.touch(d, "a/b/3.txt", 0)
 252         self.mkdir(d, "a/c")
 253         self.touch(d, "a/c/1.txt")
 254         self.touch(d, "a/c/2.txt", 0444)
 255         self.touch(d, "a/c/3.txt", 0)
 256         os.chmod(os.path.join(d, "a/c"), 0444)
 257         self.mkdir(d, "a/d")
 258         self.touch(d, "a/d/1.txt")
 259         self.touch(d, "a/d/2.txt", 0444)
 260         self.touch(d, "a/d/3.txt", 0)
 261         os.chmod(os.path.join(d, "a/d"), 0)
 262
 263         fileutil.rm_dir(d)
 264         self.failIf(os.path.exists(d))
 265         # remove it again to test idempotency
 266         fileutil.rm_dir(d)
 267
 268     def test_remove_if_possible(self):
 269         basedir = "util/FileUtil/test_remove_if_possible"
 270         fileutil.make_dirs(basedir)
 271         self.touch(basedir, "here")
 272         fn = os.path.join(basedir, "here")
 273         fileutil.remove_if_possible(fn)
 274         self.failIf(os.path.exists(fn))
 275         fileutil.remove_if_possible(fn) # should be idempotent
 276         fileutil.rm_dir(basedir)
 277         fileutil.remove_if_possible(fn) # should survive errors
 278
 279     def test_open_or_create(self):
 280         basedir = "util/FileUtil/test_open_or_create"
 281         fileutil.make_dirs(basedir)
 282         fn = os.path.join(basedir, "here")
 283         f = fileutil.open_or_create(fn)
 284         f.write("stuff.")
 285         f.close()
 286         f = fileutil.open_or_create(fn)
 287         f.seek(0, 2)
 288         f.write("more.")
 289         f.close()
 290         f = open(fn, "r")
 291         data = f.read()
 292         f.close()
 293         self.failUnlessEqual(data, "stuff.more.")
 294
 295     def test_NamedTemporaryDirectory(self):
 296         basedir = "util/FileUtil/test_NamedTemporaryDirectory"
 297         fileutil.make_dirs(basedir)
 298         td = fileutil.NamedTemporaryDirectory(dir=basedir)
 299         name = td.name
 300         self.failUnless(basedir in name)
 301         self.failUnless(basedir in repr(td))
 302         self.failUnless(os.path.isdir(name))
 303         del td
 304         # it is conceivable that we need to force gc here, but I'm not sure
 305         self.failIf(os.path.isdir(name))
 306
 307     def test_rename(self):
 308         basedir = "util/FileUtil/test_rename"
 309         fileutil.make_dirs(basedir)
 310         self.touch(basedir, "here")
 311         fn = os.path.join(basedir, "here")
 312         fn2 = os.path.join(basedir, "there")
 313         fileutil.rename(fn, fn2)
 314         self.failIf(os.path.exists(fn))
 315         self.failUnless(os.path.exists(fn2))
 316
 317     def test_du(self):
 318         basedir = "util/FileUtil/test_du"
 319         fileutil.make_dirs(basedir)
 320         d = os.path.join(basedir, "space-consuming")
 321         self.mkdir(d, "a/b")
 322         self.touch(d, "a/b/1.txt", data="a"*10)
 323         self.touch(d, "a/b/2.txt", data="b"*11)
 324         self.mkdir(d, "a/c")
 325         self.touch(d, "a/c/1.txt", data="c"*12)
 326         self.touch(d, "a/c/2.txt", data="d"*13)
 327
 328         used = fileutil.du(basedir)
 329         self.failUnlessEqual(10+11+12+13, used)
 330
 331 class PollMixinTests(unittest.TestCase):
 332     def setUp(self):
 333         self.pm = pollmixin.PollMixin()
 334
 335     def test_PollMixin_True(self):
 336         d = self.pm.poll(check_f=lambda : True,
 337                          pollinterval=0.1)
 338         return d
 339
 340     def test_PollMixin_False_then_True(self):
 341         i = iter([False, True])
 342         d = self.pm.poll(check_f=i.next,
 343                          pollinterval=0.1)
 344         return d
 345
 346     def test_timeout(self):
 347         d = self.pm.poll(check_f=lambda: False,
 348                          pollinterval=0.01,
 349                          timeout=1)
 350         def _suc(res):
 351             self.fail("poll should have failed, not returned %s" % (res,))
 352         def _err(f):
 353             f.trap(pollmixin.TimeoutError)
 354             return None # success
 355         d.addCallbacks(_suc, _err)
 356         return d
 357
 358 class DeferredUtilTests(unittest.TestCase):
 359     def test_gather_results(self):
 360         d1 = defer.Deferred()
 361         d2 = defer.Deferred()
 362         res = deferredutil.gatherResults([d1, d2])
 363         d1.errback(ValueError("BAD"))
 364         def _callb(res):
 365             self.fail("Should have errbacked, not resulted in %s" % (res,))
 366         def _errb(thef):
 367             thef.trap(ValueError)
 368         res.addCallbacks(_callb, _errb)
 369         return res
 370
 371     def test_success(self):
 372         d1, d2 = defer.Deferred(), defer.Deferred()
 373         good = []
 374         bad = []
 375         dlss = deferredutil.DeferredListShouldSucceed([d1,d2])
 376         dlss.addCallbacks(good.append, bad.append)
 377         d1.callback(1)
 378         d2.callback(2)
 379         self.failUnlessEqual(good, [[1,2]])
 380         self.failUnlessEqual(bad, [])
 381
 382     def test_failure(self):
 383         d1, d2 = defer.Deferred(), defer.Deferred()
 384         good = []
 385         bad = []
 386         dlss = deferredutil.DeferredListShouldSucceed([d1,d2])
 387         dlss.addCallbacks(good.append, bad.append)
 388         d1.addErrback(lambda _ignore: None)
 389         d2.addErrback(lambda _ignore: None)
 390         d1.callback(1)
 391         d2.errback(RuntimeError())
 392         self.failUnlessEqual(good, [])
 393         self.failUnlessEqual(len(bad), 1)
 394         f = bad[0]
 395         self.failUnless(isinstance(f, failure.Failure))
 396         self.failUnless(f.check(RuntimeError))
 397
 398 class HashUtilTests(unittest.TestCase):
 399
 400     def test_random_key(self):
 401         k = hashutil.random_key()
 402         self.failUnlessEqual(len(k), hashutil.KEYLEN)
 403
 404     def test_sha256d(self):
 405         h1 = hashutil.tagged_hash("tag1", "value")
 406         h2 = hashutil.tagged_hasher("tag1")
 407         h2.update("value")
 408         h2a = h2.digest()
 409         h2b = h2.digest()
 410         self.failUnlessEqual(h1, h2a)
 411         self.failUnlessEqual(h2a, h2b)
 412
 413     def test_sha256d_truncated(self):
 414         h1 = hashutil.tagged_hash("tag1", "value", 16)
 415         h2 = hashutil.tagged_hasher("tag1", 16)
 416         h2.update("value")
 417         h2 = h2.digest()
 418         self.failUnlessEqual(len(h1), 16)
 419         self.failUnlessEqual(len(h2), 16)
 420         self.failUnlessEqual(h1, h2)
 421
 422     def test_chk(self):
 423         h1 = hashutil.convergence_hash(3, 10, 1000, "data", "secret")
 424         h2 = hashutil.convergence_hasher(3, 10, 1000, "secret")
 425         h2.update("data")
 426         h2 = h2.digest()
 427         self.failUnlessEqual(h1, h2)
 428
 429     def test_hashers(self):
 430         h1 = hashutil.block_hash("foo")
 431         h2 = hashutil.block_hasher()
 432         h2.update("foo")
 433         self.failUnlessEqual(h1, h2.digest())
 434
 435         h1 = hashutil.uri_extension_hash("foo")
 436         h2 = hashutil.uri_extension_hasher()
 437         h2.update("foo")
 438         self.failUnlessEqual(h1, h2.digest())
 439
 440         h1 = hashutil.plaintext_hash("foo")
 441         h2 = hashutil.plaintext_hasher()
 442         h2.update("foo")
 443         self.failUnlessEqual(h1, h2.digest())
 444
 445         h1 = hashutil.crypttext_hash("foo")
 446         h2 = hashutil.crypttext_hasher()
 447         h2.update("foo")
 448         self.failUnlessEqual(h1, h2.digest())
 449
 450         h1 = hashutil.crypttext_segment_hash("foo")
 451         h2 = hashutil.crypttext_segment_hasher()
 452         h2.update("foo")
 453         self.failUnlessEqual(h1, h2.digest())
 454
 455         h1 = hashutil.plaintext_segment_hash("foo")
 456         h2 = hashutil.plaintext_segment_hasher()
 457         h2.update("foo")
 458         self.failUnlessEqual(h1, h2.digest())
 459
 460 class Abbreviate(unittest.TestCase):
 461     def test_time(self):
 462         a = abbreviate.abbreviate_time
 463         self.failUnlessEqual(a(None), "unknown")
 464         self.failUnlessEqual(a(0), "0 seconds")
 465         self.failUnlessEqual(a(1), "1 second")
 466         self.failUnlessEqual(a(2), "2 seconds")
 467         self.failUnlessEqual(a(119), "119 seconds")
 468         MIN = 60
 469         self.failUnlessEqual(a(2*MIN), "2 minutes")
 470         self.failUnlessEqual(a(60*MIN), "60 minutes")
 471         self.failUnlessEqual(a(179*MIN), "179 minutes")
 472         HOUR = 60*MIN
 473         self.failUnlessEqual(a(180*MIN), "3 hours")
 474         self.failUnlessEqual(a(4*HOUR), "4 hours")
 475         DAY = 24*HOUR
 476         MONTH = 30*DAY
 477         self.failUnlessEqual(a(2*DAY), "2 days")
 478         self.failUnlessEqual(a(2*MONTH), "2 months")
 479         YEAR = 365*DAY
 480         self.failUnlessEqual(a(5*YEAR), "5 years")
 481
 482     def test_space(self):
 483         tests_si = [(None, "unknown"),
 484                     (0, "0 B"),
 485                     (1, "1 B"),
 486                     (999, "999 B"),
 487                     (1000, "1000 B"),
 488                     (1023, "1023 B"),
 489                     (1024, "1.02 kB"),
 490                     (20*1000, "20.00 kB"),
 491                     (1024*1024, "1.05 MB"),
 492                     (1000*1000, "1.00 MB"),
 493                     (1000*1000*1000, "1.00 GB"),
 494                     (1000*1000*1000*1000, "1.00 TB"),
 495                     (1000*1000*1000*1000*1000, "1.00 PB"),
 496                     (1234567890123456, "1.23 PB"),
 497                     ]
 498         for (x, expected) in tests_si:
 499             got = abbreviate.abbreviate_space(x, SI=True)
 500             self.failUnlessEqual(got, expected)
 501
 502         tests_base1024 = [(None, "unknown"),
 503                           (0, "0 B"),
 504                           (1, "1 B"),
 505                           (999, "999 B"),
 506                           (1000, "1000 B"),
 507                           (1023, "1023 B"),
 508                           (1024, "1.00 kiB"),
 509                           (20*1024, "20.00 kiB"),
 510                           (1000*1000, "976.56 kiB"),
 511                           (1024*1024, "1.00 MiB"),
 512                           (1024*1024*1024, "1.00 GiB"),
 513                           (1024*1024*1024*1024, "1.00 TiB"),
 514                           (1000*1000*1000*1000*1000, "909.49 TiB"),
 515                           (1024*1024*1024*1024*1024, "1.00 PiB"),
 516                           (1234567890123456, "1.10 PiB"),
 517                     ]
 518         for (x, expected) in tests_base1024:
 519             got = abbreviate.abbreviate_space(x, SI=False)
 520             self.failUnlessEqual(got, expected)
 521
 522         self.failUnlessEqual(abbreviate.abbreviate_space_both(1234567),
 523                              "(1.23 MB, 1.18 MiB)")
 524
 525     def test_parse_space(self):
 526         p = abbreviate.parse_abbreviated_size
 527         self.failUnlessEqual(p(""), None)
 528         self.failUnlessEqual(p(None), None)
 529         self.failUnlessEqual(p("123"), 123)
 530         self.failUnlessEqual(p("123B"), 123)
 531         self.failUnlessEqual(p("2K"), 2000)
 532         self.failUnlessEqual(p("2kb"), 2000)
 533         self.failUnlessEqual(p("2KiB"), 2048)
 534         self.failUnlessEqual(p("10MB"), 10*1000*1000)
 535         self.failUnlessEqual(p("10MiB"), 10*1024*1024)
 536         self.failUnlessEqual(p("5G"), 5*1000*1000*1000)
 537         self.failUnlessEqual(p("4GiB"), 4*1024*1024*1024)
 538         e = self.failUnlessRaises(ValueError, p, "12 cubits")
 539         self.failUnless("12 cubits" in str(e))
 540
 541 class Limiter(unittest.TestCase):
 542     def job(self, i, foo):
 543         self.calls.append( (i, foo) )
 544         self.simultaneous += 1
 545         self.peak_simultaneous = max(self.simultaneous, self.peak_simultaneous)
 546         d = defer.Deferred()
 547         def _done():
 548             self.simultaneous -= 1
 549             d.callback("done %d" % i)
 550         reactor.callLater(1.0, _done)
 551         return d
 552
 553     def bad_job(self, i, foo):
 554         raise RuntimeError("bad_job %d" % i)
 555
 556     def test_limiter(self):
 557         self.calls = []
 558         self.simultaneous = 0
 559         self.peak_simultaneous = 0
 560         l = limiter.ConcurrencyLimiter()
 561         dl = []
 562         for i in range(20):
 563             dl.append(l.add(self.job, i, foo=str(i)))
 564         d = defer.DeferredList(dl, fireOnOneErrback=True)
 565         def _done(res):
 566             self.failUnlessEqual(self.simultaneous, 0)
 567             self.failUnless(self.peak_simultaneous <= 10)
 568             self.failUnlessEqual(len(self.calls), 20)
 569             for i in range(20):
 570                 self.failUnless( (i, str(i)) in self.calls)
 571         d.addCallback(_done)
 572         return d
 573
 574     def test_errors(self):
 575         self.calls = []
 576         self.simultaneous = 0
 577         self.peak_simultaneous = 0
 578         l = limiter.ConcurrencyLimiter()
 579         dl = []
 580         for i in range(20):
 581             dl.append(l.add(self.job, i, foo=str(i)))
 582         d2 = l.add(self.bad_job, 21, "21")
 583         d = defer.DeferredList(dl, fireOnOneErrback=True)
 584         def _most_done(res):
 585             results = []
 586             for (success, result) in res:
 587                 self.failUnlessEqual(success, True)
 588                 results.append(result)
 589             results.sort()
 590             expected_results = ["done %d" % i for i in range(20)]
 591             expected_results.sort()
 592             self.failUnlessEqual(results, expected_results)
 593             self.failUnless(self.peak_simultaneous <= 10)
 594             self.failUnlessEqual(len(self.calls), 20)
 595             for i in range(20):
 596                 self.failUnless( (i, str(i)) in self.calls)
 597             def _good(res):
 598                 self.fail("should have failed, not got %s" % (res,))
 599             def _err(f):
 600                 f.trap(RuntimeError)
 601                 self.failUnless("bad_job 21" in str(f))
 602             d2.addCallbacks(_good, _err)
 603             return d2
 604         d.addCallback(_most_done)
 605         def _all_done(res):
 606             self.failUnlessEqual(self.simultaneous, 0)
 607             self.failUnless(self.peak_simultaneous <= 10)
 608             self.failUnlessEqual(len(self.calls), 20)
 609             for i in range(20):
 610                 self.failUnless( (i, str(i)) in self.calls)
 611         d.addCallback(_all_done)
 612         return d
 613
 614 class TimeFormat(unittest.TestCase):
 615     def test_epoch(self):
 616         s = time_format.iso_utc_time_to_localseconds("1970-01-01T00:00:01")
 617         self.failUnlessEqual(s, 1.0)
 618         s = time_format.iso_utc_time_to_localseconds("1970-01-01_00:00:01")
 619         self.failUnlessEqual(s, 1.0)
 620         s = time_format.iso_utc_time_to_localseconds("1970-01-01 00:00:01")
 621         self.failUnlessEqual(s, 1.0)
 622
 623         self.failUnlessEqual(time_format.iso_utc(1.0), "1970-01-01_00:00:01")
 624         self.failUnlessEqual(time_format.iso_utc(1.0, sep=" "),
 625                              "1970-01-01 00:00:01")
 626         now = time.time()
 627         def my_time():
 628             return 1.0
 629         self.failUnlessEqual(time_format.iso_utc(t=my_time),
 630                              "1970-01-01_00:00:01")
 631         e = self.failUnlessRaises(ValueError,
 632                                   time_format.iso_utc_time_to_localseconds,
 633                                   "invalid timestring")
 634         self.failUnless("not a complete ISO8601 timestamp" in str(e))
 635         s = time_format.iso_utc_time_to_localseconds("1970-01-01_00:00:01.500")
 636         self.failUnlessEqual(s, 1.5)
 637
 638 class CacheDir(unittest.TestCase):
 639     def test_basic(self):
 640         basedir = "test_util/CacheDir/test_basic"
 641
 642         def _failIfExists(name):
 643             absfn = os.path.join(basedir, name)
 644             self.failIf(os.path.exists(absfn),
 645                         "%s exists but it shouldn't" % absfn)
 646
 647         def _failUnlessExists(name):
 648             absfn = os.path.join(basedir, name)
 649             self.failUnless(os.path.exists(absfn),
 650                             "%s doesn't exist but it should" % absfn)
 651
 652         cdm = cachedir.CacheDirectoryManager(basedir)
 653         a = cdm.get_file("a")
 654         b = cdm.get_file("b")
 655         c = cdm.get_file("c")
 656         f = open(a.get_filename(), "wb"); f.write("hi"); f.close(); del f
 657         f = open(b.get_filename(), "wb"); f.write("hi"); f.close(); del f
 658         f = open(c.get_filename(), "wb"); f.write("hi"); f.close(); del f
 659
 660         _failUnlessExists("a")
 661         _failUnlessExists("b")
 662         _failUnlessExists("c")
 663
 664         cdm.check()
 665
 666         _failUnlessExists("a")
 667         _failUnlessExists("b")
 668         _failUnlessExists("c")
 669
 670         del a
 671         # this file won't be deleted yet, because it isn't old enough
 672         cdm.check()
 673         _failUnlessExists("a")
 674         _failUnlessExists("b")
 675         _failUnlessExists("c")
 676
 677         # we change the definition of "old" to make everything old
 678         cdm.old = -10
 679
 680         cdm.check()
 681         _failIfExists("a")
 682         _failUnlessExists("b")
 683         _failUnlessExists("c")
 684
 685         cdm.old = 60*60
 686
 687         del b
 688
 689         cdm.check()
 690         _failIfExists("a")
 691         _failUnlessExists("b")
 692         _failUnlessExists("c")
 693
 694         b2 = cdm.get_file("b")
 695
 696         cdm.check()
 697         _failIfExists("a")
 698         _failUnlessExists("b")
 699         _failUnlessExists("c")