ExternalVendorCode/tahoe-lafs
1
0
Fork 0
mirror of https://github.com/tahoe-lafs/tahoe-lafs.git synced 2025-01-31 16:36:20 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Some progress towards passing tests.

Browse Source
This commit is contained in:
Itamar Turner-Trauring 2020-09-25 11:46:31 -04:00
parent 51709d5562
commit 5d9a1a5ab5
1 changed files with 15 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
src/allmydata/test/test_helper.py
View File

@ -1,3 +1,5 @@
from future.builtins import str
import os import os
from twisted.internet import defer from twisted.internet import defer
from twisted.trial import unittest from twisted.trial import unittest
@ -18,7 +20,7 @@ from .common import (
MiB = 1024*1024 MiB = 1024*1024
DATA = "I need help\n" * 1000 DATA = b"I need help\n" * 1000
class CHKUploadHelper_fake(offloaded.CHKUploadHelper): class CHKUploadHelper_fake(offloaded.CHKUploadHelper):
def start_encrypted(self, eu): def start_encrypted(self, eu):
@ -33,8 +35,8 @@ class CHKUploadHelper_fake(offloaded.CHKUploadHelper):
"segment_size": segsize, "segment_size": segsize,
"size": size, "size": size,
} }
ueb_hash = "fake" ueb_hash = b"fake"
v = uri.CHKFileVerifierURI(self._storage_index, "x"*32, v = uri.CHKFileVerifierURI(self._storage_index, b"x"*32,
needed_shares, total_shares, size) needed_shares, total_shares, size)
_UR = upload.UploadResults _UR = upload.UploadResults
ur = _UR(file_size=size, ur = _UR(file_size=size,
@ -56,7 +58,7 @@ class CHKUploadHelper_fake(offloaded.CHKUploadHelper):
class Helper_fake_upload(offloaded.Helper): class Helper_fake_upload(offloaded.Helper):
def _make_chk_upload_helper(self, storage_index, lp): def _make_chk_upload_helper(self, storage_index, lp):
si_s = si_b2a(storage_index) si_s = str(si_b2a(storage_index), "utf-8")
incoming_file = os.path.join(self._chk_incoming, si_s) incoming_file = os.path.join(self._chk_incoming, si_s)
encoding_file = os.path.join(self._chk_encoding, si_s) encoding_file = os.path.join(self._chk_encoding, si_s)
uh = CHKUploadHelper_fake(storage_index, self, uh = CHKUploadHelper_fake(storage_index, self,
@ -69,7 +71,7 @@ class Helper_fake_upload(offloaded.Helper):
class Helper_already_uploaded(Helper_fake_upload): class Helper_already_uploaded(Helper_fake_upload):
def _check_chk(self, storage_index, lp): def _check_chk(self, storage_index, lp):
res = upload.HelperUploadResults() res = upload.HelperUploadResults()
res.uri_extension_hash = hashutil.uri_extension_hash("") res.uri_extension_hash = hashutil.uri_extension_hash(b"")
# we're pretending that the file they're trying to upload was already # we're pretending that the file they're trying to upload was already
# present in the grid. We return some information about the file, so # present in the grid. We return some information about the file, so
@ -127,7 +129,7 @@ class AssistedUpload(unittest.TestCase):
lambda h: self.tub, lambda h: self.tub,
EMPTY_CLIENT_CONFIG, EMPTY_CLIENT_CONFIG,
) )
self.s.secret_holder = client.SecretHolder("lease secret", "converge") self.s.secret_holder = client.SecretHolder(b"lease secret", b"converge")
self.s.startService() self.s.startService()
t.setServiceParent(self.s) t.setServiceParent(self.s)
@ -162,11 +164,11 @@ class AssistedUpload(unittest.TestCase):
def _ready(res): def _ready(res):
assert u._helper assert u._helper
return upload_data(u, DATA, convergence="some convergence string") return upload_data(u, DATA, convergence=b"some convergence string")
d.addCallback(_ready) d.addCallback(_ready)
def _uploaded(results): def _uploaded(results):
the_uri = results.get_uri() the_uri = results.get_uri()
assert "CHK" in the_uri assert b"CHK" in the_uri
d.addCallback(_uploaded) d.addCallback(_uploaded)
def _check_empty(res): def _check_empty(res):
@ -195,11 +197,11 @@ class AssistedUpload(unittest.TestCase):
# this must be a multiple of 'required_shares'==k # this must be a multiple of 'required_shares'==k
segsize = mathutil.next_multiple(segsize, k) segsize = mathutil.next_multiple(segsize, k)
key = hashutil.convergence_hash(k, n, segsize, DATA, "test convergence string") key = hashutil.convergence_hash(k, n, segsize, DATA, b"test convergence string")
assert len(key) == 16 assert len(key) == 16
encryptor = aes.create_encryptor(key) encryptor = aes.create_encryptor(key)
SI = hashutil.storage_index_hash(key) SI = hashutil.storage_index_hash(key)
SI_s = si_b2a(SI) SI_s = str(si_b2a(SI), "utf-8")
encfile = os.path.join(self.basedir, "CHK_encoding", SI_s) encfile = os.path.join(self.basedir, "CHK_encoding", SI_s)
f = open(encfile, "wb") f = open(encfile, "wb")
f.write(aes.encrypt_data(encryptor, DATA)) f.write(aes.encrypt_data(encryptor, DATA))
@ -212,7 +214,7 @@ class AssistedUpload(unittest.TestCase):
def _ready(res): def _ready(res):
assert u._helper assert u._helper
return upload_data(u, DATA, convergence="test convergence string") return upload_data(u, DATA, convergence=b"test convergence string")
d.addCallback(_ready) d.addCallback(_ready)
def _uploaded(results): def _uploaded(results):
the_uri = results.get_uri() the_uri = results.get_uri()
@ -239,11 +241,11 @@ class AssistedUpload(unittest.TestCase):
def _ready(res): def _ready(res):
assert u._helper assert u._helper
return upload_data(u, DATA, convergence="some convergence string") return upload_data(u, DATA, convergence=b"some convergence string")
d.addCallback(_ready) d.addCallback(_ready)
def _uploaded(results): def _uploaded(results):
the_uri = results.get_uri() the_uri = results.get_uri()
assert "CHK" in the_uri assert b"CHK" in the_uri
d.addCallback(_uploaded) d.addCallback(_uploaded)
def _check_empty(res): def _check_empty(res):