def _help_test_random():
m = random.randrange(1, 257)
k = random.randrange(1, m+1)
- l = random.randrange(0, 2**10)
+ l = random.randrange(0, 2**9)
ss = [ randstr(l/k) for x in range(k) ]
_h(k, m, ss)
blocks = [ x[1] for x in nums_and_blocks ]
nums = [ x[0] for x in nums_and_blocks ]
decer = zfec.easyfec.Decoder(k, m)
- decodeds = decer.decode(blocks, nums)
- assert len(decodeds) == len(s), (len(decodeds), len(s),)
- assert decodeds == s, (decodeds, s,)
+
+ decodeds = decer.decode(blocks, nums, padlen=k*len(blocks[0]) - len(s))
+ assert len(decodeds) == len(s), (ab(decodeds), ab(s), k, m)
+ assert decodeds == s, (ab(decodeds), ab(s),)
def _help_test_random_easy():
m = random.randrange(1, 257)
k = random.randrange(1, m+1)
- l = random.randrange(0, 2**10)
+ l = random.randrange(0, 2**9)
s = randstr(l)
_h_easy(k, m, s)
_h_easy(k, m, s)
class ZFecTest(unittest.TestCase):
+ def test_from_agl_c(self):
+ self.failUnless(zfec._fec.test_from_agl())
+
+ def test_from_agl_py(self):
+ e = zfec.Encoder(3, 5)
+ b0 = '\x01'*8 ; b1 = '\x02'*8 ; b2 = '\x03'*8
+ # print "_from_py before encoding:"
+ # print "b0: %s, b1: %s, b2: %s" % tuple(base64.b16encode(x) for x in [b0, b1, b2])
+
+ b3, b4 = e.encode([b0, b1, b2], (3, 4))
+ # print "after encoding:"
+ # print "b3: %s, b4: %s" % tuple(base64.b16encode(x) for x in [b3, b4])
+
+ d = zfec.Decoder(3, 5)
+ r0, r1, r2 = d.decode((b2, b3, b4), (1, 2, 3))
+
+ # print "after decoding:"
+ # print "b0: %s, b1: %s" % tuple(base64.b16encode(x) for x in [b0, b1])
+
def test_small(self):
for i in range(16):
_help_test_random_with_l(i)
if VERBOSE:
print "%d randomized tests pass." % (i+1)
+ def test_bad_args_construct_decoder(self):
+ try:
+ zfec.Decoder(-1, -1)
+ except zfec.Error, e:
+ assert "argument is required to be greater than or equal to 1" in str(e), e
+ else:
+ self.fail("Should have gotten an exception from out-of-range arguments.")
+
+ try:
+ zfec.Decoder(1, 257)
+ except zfec.Error, e:
+ assert "argument is required to be less than or equal to 256" in str(e), e
+ else:
+ self.fail("Should have gotten an exception from out-of-range arguments.")
+
+ try:
+ zfec.Decoder(3, 2)
+ except zfec.Error, e:
+ assert "first argument is required to be less than or equal to the second argument" in str(e), e
+ else:
+ self.fail("Should have gotten an exception from out-of-range arguments.")
+
+ def test_bad_args_construct_encoder(self):
+ try:
+ zfec.Encoder(-1, -1)
+ except zfec.Error, e:
+ assert "argument is required to be greater than or equal to 1" in str(e), e
+ else:
+ self.fail("Should have gotten an exception from out-of-range arguments.")
+
+ try:
+ zfec.Encoder(1, 257)
+ except zfec.Error, e:
+ assert "argument is required to be less than or equal to 256" in str(e), e
+ else:
+ self.fail("Should have gotten an exception from out-of-range arguments.")
+
def test_bad_args_dec(self):
decer = zfec.Decoder(2, 4)
try:
- decer.decode(98) # first argument is not a sequence
+ decer.decode(98, []) # first argument is not a sequence
except TypeError, e:
assert "First argument was not a sequence" in str(e), e
else:
class EasyFecTest(unittest.TestCase):
def test_small(self):
- for i in range(2**10):
+ for i in range(16):
_help_test_random_with_l_easy(i)
if VERBOSE:
print "%d randomized tests pass." % (i+1)
def test_random(self):
- for i in range(2**8):
+ for i in range(3):
_help_test_random_easy()
if VERBOSE:
print "%d randomized tests pass." % (i+1)
decer = zfec.easyfec.Decoder(2, 4)
try:
- decer.decode(98, [0, 1]) # first argument is not a sequence
+ decer.decode(98, [0, 1], 0) # first argument is not a sequence
except TypeError, e:
assert "First argument was not a sequence" in str(e), e
else:
raise "Should have gotten TypeError for wrong type of second argument."
try:
- decer.decode("ab", ["c", "d",])
+ decer.decode("ab", ["c", "d",], 0)
except zfec.Error, e:
assert "Precondition violation: second argument is required to contain int" in str(e), e
else:
raise "Should have gotten zfec.Error for wrong type of second argument."
try:
- decer.decode("ab", 98) # not a sequence at all
+ decer.decode("ab", 98, 0) # not a sequence at all
except TypeError, e:
assert "Second argument was not a sequence" in str(e), e
else:
try:
tempf = tempdir.file(TESTFNAME, 'w+b')
tempf.write(teststr)
+ tempf.flush()
tempf.seek(0)
# encode the file
# decode from the share files
outf = tempdir.file('recovered-testfile.txt', 'w+b')
zfec.filefec.decode_from_files(outf, sharefs, verbose=VERBOSE)
+ outf.flush()
outf.seek(0)
recovereddata = outf.read()
- assert recovereddata == teststr
+ assert recovereddata == teststr, (ab(recovereddata), ab(teststr),)
finally:
tempdir.shutdown()
if __name__ == "__main__":
unittest.main()
-
-# zfec -- fast forward error correction library with Python interface
-#
-# Copyright (C) 2007 Allmydata, Inc.
-# Author: Zooko Wilcox-O'Hearn
-#
-# This file is part of zfec.
-#
-# This program is free software; you can redistribute it and/or modify it
-# under the terms of the GNU General Public License as published by the Free
-# Software Foundation; either version 2 of the License, or (at your option)
-# any later version, with the added permission that, if you become obligated
-# to release a derived work under this licence (as per section 2.b), you may
-# delay the fulfillment of this obligation for up to 12 months. See the file
-# COPYING for details.
-#
-# If you would like to inquire about a commercial relationship with Allmydata,
-# Inc., please contact partnerships@allmydata.com and visit
-# http://allmydata.com/.
projects / tahoe-lafs / zfec.git / blobdiff