from allmydata.stats import StatsProvider
from allmydata.history import History
from allmydata.interfaces import IURI, INewDirectoryURI, IStatsProducer, \
- IReadonlyNewDirectoryURI, IFileURI, IMutableFileURI, RIStubClient
+ IReadonlyNewDirectoryURI, IFileURI, IMutableFileURI, RIStubClient, \
+ UnhandledCapTypeError
KiB=1024
MiB=1024*KiB
node = LiteralFileNode(u, self) # LIT
else:
node = FileNode(u, self, self.download_cache_dirman) # CHK
- else:
- assert IMutableFileURI.providedBy(u), u
+ elif IMutableFileURI.providedBy(u):
node = MutableFileNode(self).init_from_uri(u)
+ else:
+ raise UnhandledCapTypeError("cap is recognized, but has no Node")
self._node_cache[u_s] = node # note: WeakValueDictionary
return self._node_cache[u_s]
"""UnknownNodes (using filecaps from the future that we don't understand)
cannot yet be copied safely, so I refuse to copy them."""
+class UnhandledCapTypeError(Exception):
+ """I recognize the cap/URI, but I cannot create an IFilesystemNode for
+ it."""
+
class IFilesystemNode(Interface):
def get_uri():
"""
self.shnum,
self.reason)
-
+class UnknownVersionError(Exception):
+ """The share we received was of a version we don't recognize."""
class ResponseCache:
"""I cache share data, to reduce the number of round trips used during
import struct
-from common import NeedMoreDataError
+from common import NeedMoreDataError, UnknownVersionError
PREFIX = ">BQ32s16s" # each version has a different prefix
SIGNED_PREFIX = ">BQ32s16s BBQQ" # this is covered by the signature
k, N, segsize, datalen,
o) = unpack_header(data)
- assert version == 0
+ if version != 0:
+ raise UnknownVersionError("got mutable share version %d, but I only understand version 0" % version)
+
if len(data) < o['share_hash_chain']:
raise NeedMoreDataError(o['share_hash_chain'],
o['enc_privkey'], o['EOF']-o['enc_privkey'])
o['enc_privkey'],
o['EOF']) = struct.unpack(HEADER, data[:HEADER_LENGTH])
- assert version == 0
+ if version != 0:
+ raise UnknownVersionError("got mutable share version %d, but I only understand version 0" % version)
+
if len(data) < o['EOF']:
raise NeedMoreDataError(o['EOF'],
o['enc_privkey'], o['EOF']-o['enc_privkey'])
def unpack_checkstring(checkstring):
cs_len = struct.calcsize(PREFIX)
version, seqnum, root_hash, IV = struct.unpack(PREFIX, checkstring[:cs_len])
- assert version == 0 # TODO: just ignore the share
+ if version != 0: # TODO: just ignore the share
+ raise UnknownVersionError("got mutable share version %d, but I only understand version 0" % version)
return (seqnum, root_hash, IV)
def pack_prefix(seqnum, root_hash, IV,
# should be noted in the servermap's list of problems.
if substring:
allproblems = [str(f) for f in servermap.problems]
- self.failUnless(substring in "".join(allproblems))
+ self.failUnlessIn(substring, "".join(allproblems))
return servermap
if should_succeed:
d1 = self._fn.download_version(servermap, ver)
return d
def test_corrupt_all_verbyte(self):
- # when the version byte is not 0, we hit an assertion error in
- # unpack_share().
- d = self._test_corrupt_all(0, "AssertionError")
+ # when the version byte is not 0, we hit an UnknownVersionError error
+ # in unpack_share().
+ d = self._test_corrupt_all(0, "UnknownVersionError")
def _check_servermap(servermap):
# and the dump should mention the problems
s = StringIO()
def test_constraint(self):
good="http://127.0.0.1:3456/uri/URI%3ADIR2%3Agh3l5rbvnv2333mrfvalmjfr4i%3Alz6l7u3z3b7g37s4zkdmfpx5ly4ib4m6thrpbusi6ys62qtc6mma/"
uri.NewDirectoryURI.init_from_human_encoding(good)
- self.failUnlessRaises(AssertionError, uri.NewDirectoryURI.init_from_string, good)
+ self.failUnlessRaises(uri.BadURIError, uri.NewDirectoryURI.init_from_string, good)
bad = good + '==='
- self.failUnlessRaises(AssertionError, uri.NewDirectoryURI.init_from_human_encoding, bad)
- self.failUnlessRaises(AssertionError, uri.NewDirectoryURI.init_from_string, bad)
+ self.failUnlessRaises(uri.BadURIError, uri.NewDirectoryURI.init_from_human_encoding, bad)
+ self.failUnlessRaises(uri.BadURIError, uri.NewDirectoryURI.init_from_string, bad)
fileURI = 'URI:CHK:gh3l5rbvnv2333mrfvalmjfr4i:lz6l7u3z3b7g37s4zkdmfpx5ly4ib4m6thrpbusi6ys62qtc6mma:3:10:345834'
uri.CHKFileURI.init_from_string(fileURI)
from allmydata.test.common import FakeDirectoryNode, FakeCHKFileNode, \
FakeMutableFileNode, create_chk_filenode, WebErrorMixin, ShouldFailMixin
from allmydata.interfaces import IURI, INewDirectoryURI, \
- IReadonlyNewDirectoryURI, IFileURI, IMutableFileURI, IMutableFileNode
+ IReadonlyNewDirectoryURI, IFileURI, IMutableFileURI, IMutableFileNode, \
+ UnhandledCapTypeError
from allmydata.mutable import servermap, publish, retrieve
import common_util as testutil
from allmydata.test.no_network import GridTestMixin
return FakeDirectoryNode(self).init_from_uri(u)
if IFileURI.providedBy(u):
return FakeCHKFileNode(u, self)
- assert IMutableFileURI.providedBy(u), u
- return FakeMutableFileNode(self).init_from_uri(u)
+ if IMutableFileURI.providedBy(u):
+ return FakeMutableFileNode(self).init_from_uri(u)
+ raise UnhandledCapTypeError("cap '%s' is recognized, but has no Node" % auri)
def create_empty_dirnode(self):
n = FakeDirectoryNode(self)
from allmydata.interfaces import IURI, IDirnodeURI, IFileURI, IImmutableFileURI, \
IVerifierURI, IMutableFileURI, INewDirectoryURI, IReadonlyNewDirectoryURI
+class BadURIError(Exception):
+ pass
+
# the URI shall be an ascii representation of the file. It shall contain
# enough information to retrieve and validate the contents. It shall be
# expressed in a limited character set (namely [TODO]).
self.total_shares = total_shares
self.size = size
self.storage_index = hashutil.storage_index_hash(self.key)
- assert len(self.storage_index) == 16
- self.storage_index = hashutil.storage_index_hash(key)
- assert len(self.storage_index) == 16 # sha256 hash truncated to 128
+ if not len(self.storage_index) == 16: # sha256 hash truncated to 128
+ raise BadURIError("storage index must be 16 bytes long")
@classmethod
def init_from_human_encoding(cls, uri):
mo = cls.HUMAN_RE.search(uri)
- assert mo, uri
+ if not mo:
+ raise BadURIError("%s doesn't look like a cap" % (uri,))
return cls(base32.a2b(mo.group(1)), base32.a2b(mo.group(2)),
int(mo.group(3)), int(mo.group(4)), int(mo.group(5)))
@classmethod
def init_from_string(cls, uri):
mo = cls.STRING_RE.search(uri)
- assert mo, uri
+ if not mo:
+ raise BadURIError("%s doesn't look like a cap" % (uri,))
return cls(base32.a2b(mo.group(1)), base32.a2b(mo.group(2)),
int(mo.group(3)), int(mo.group(4)), int(mo.group(5)))
@classmethod
def init_from_human_encoding(cls, uri):
mo = cls.HUMAN_RE.search(uri)
- assert mo, uri
+ if not mo:
+ raise BadURIError("'%s' doesn't look like a cap" % (uri,))
return cls(base32.a2b(mo.group(1)), base32.a2b(mo.group(2)))
@classmethod
def init_from_string(cls, uri):
mo = cls.STRING_RE.search(uri)
- assert mo, (uri, cls)
+ if not mo:
+ raise BadURIError("'%s' doesn't look like a %s cap" % (uri, cls))
return cls(base32.a2b(mo.group(1)), base32.a2b(mo.group(2)))
def to_string(self):
@classmethod
def init_from_human_encoding(cls, uri):
mo = cls.HUMAN_RE.search(uri)
- assert mo, uri
+ if not mo:
+ raise BadURIError("'%s' doesn't look like a cap" % (uri,))
return cls(base32.a2b(mo.group(1)), base32.a2b(mo.group(2)))
@classmethod
def init_from_string(cls, uri):
mo = cls.STRING_RE.search(uri)
- assert mo, uri
+ if not mo:
+ raise BadURIError("'%s' doesn't look like a cap" % (uri,))
return cls(base32.a2b(mo.group(1)), base32.a2b(mo.group(2)))
def to_string(self):
@classmethod
def init_from_string(cls, uri):
mo = cls.BASE_STRING_RE.search(uri)
- assert mo, (uri, cls)
+ if not mo:
+ raise BadURIError("'%s' doesn't look like a %s cap" % (uri, cls))
bits = uri[mo.end():]
fn = cls.INNER_URI_CLASS.init_from_string(
cls.INNER_URI_CLASS.BASE_STRING+bits)
@classmethod
def init_from_human_encoding(cls, uri):
mo = cls.BASE_HUMAN_RE.search(uri)
- assert mo, (uri, cls)
+ if not mo:
+ raise BadURIError("'%s' doesn't look like a %s cap" % (uri, cls))
bits = uri[mo.end():]
while bits and bits[-1] == '/':
bits = bits[:-1]
from allmydata import get_package_versions_string
from allmydata import provisioning
from allmydata.util import idlib, log
-from allmydata.interfaces import IFileNode
+from allmydata.interfaces import IFileNode, UnhandledCapTypeError
from allmydata.web import filenode, directory, unlinked, status, operations
from allmydata.web import reliability, storage
from allmydata.web.common import abbreviate_size, getxmlfile, WebError, \
try:
node = self.client.create_node_from_uri(name)
return directory.make_handler_for(node, self.client)
- except (TypeError, AssertionError):
+ except (TypeError, UnhandledCapTypeError, AssertionError):
raise WebError("'%s' is not a valid file- or directory- cap"
% name)
# 'name' must be a file URI
try:
node = self.client.create_node_from_uri(name)
- except (TypeError, AssertionError):
+ except (TypeError, UnhandledCapTypeError, AssertionError):
raise WebError("'%s' is not a valid file- or directory- cap"
% name)
if not IFileNode.providedBy(node):
projects / tahoe-lafs / tahoe-lafs.git / commitdiff