def test_PUT_NEWFILEURL(self):
d = self.PUT("/vdrive/global/foo/new.txt", self.NEWFILE_CONTENTS)
def _check(res):
+ # TODO: we lose the response code, so we can't check this
+ #self.failUnlessEqual(responsecode, 201)
self.failUnless("new.txt" in self._foo_node.children)
new_uri = self._foo_node.children["new.txt"]
new_contents = self.files[new_uri]
d.addCallback(_check)
return d
+ def test_PUT_NEWFILEURL_replace(self):
+ d = self.PUT("/vdrive/global/foo/bar.txt", self.NEWFILE_CONTENTS)
+ def _check(res):
+ # TODO: we lose the response code, so we can't check this
+ #self.failUnlessEqual(responsecode, 200)
+ self.failUnless("bar.txt" in self._foo_node.children)
+ new_uri = self._foo_node.children["bar.txt"]
+ new_contents = self.files[new_uri]
+ self.failUnlessEqual(new_contents, self.NEWFILE_CONTENTS)
+ self.failUnlessEqual(res.strip(), new_uri)
+ d.addCallback(_check)
+ return d
+
+ def test_PUT_NEWFILEURL_no_replace(self):
+ d = self.PUT("/vdrive/global/foo/bar.txt?replace=false",
+ self.NEWFILE_CONTENTS)
+ d.addBoth(self.shouldFail, error.Error, "PUT_NEWFILEURL_no_replace",
+ "409 Conflict",
+ "There was already a child by that name, and you asked me "
+ "to not replace it")
+ return d
+
def test_PUT_NEWFILEURL_mkdirs(self):
d = self.PUT("/vdrive/global/foo/newdir/new.txt", self.NEWFILE_CONTENTS)
def _check(res):
d.addCallback(_check)
return d
+ def test_PUT_NEWDIRURL_replace(self):
+ d = self.PUT("/vdrive/global/foo/sub?t=mkdir", "")
+ def _check(res):
+ self.failUnless("sub" in self._foo_node.children)
+ newdir_uri = self._foo_node.children["sub"]
+ newdir_node = self.nodes[newdir_uri]
+ self.failIf(newdir_node.children)
+ d.addCallback(_check)
+ return d
+
+ def test_PUT_NEWDIRURL_no_replace(self):
+ d = self.PUT("/vdrive/global/foo/sub?t=mkdir&replace=false", "")
+ d.addBoth(self.shouldFail, error.Error, "PUT_NEWDIRURL_no_replace",
+ "409 Conflict",
+ "There was already a child by that name, and you asked me "
+ "to not replace it")
+ return d
+
def test_PUT_NEWDIRURL_mkdirs(self):
d = self.PUT("/vdrive/global/foo/subdir/newdir?t=mkdir", "")
def _check(res):
d.addCallback(_check)
return d
+ def test_PUT_NEWFILEURL_uri_replace(self):
+ new_uri = self.makefile(8)
+ d = self.PUT("/vdrive/global/foo/bar.txt?t=uri", new_uri)
+ def _check(res):
+ self.failUnless("bar.txt" in self._foo_node.children)
+ new_uri = self._foo_node.children["bar.txt"]
+ new_contents = self.files[new_uri]
+ self.failUnlessEqual(new_contents, self.files[new_uri])
+ self.failUnlessEqual(res.strip(), new_uri)
+ d.addCallback(_check)
+ return d
+
+ def test_PUT_NEWFILEURL_uri_no_replace(self):
+ new_uri = self.makefile(8)
+ d = self.PUT("/vdrive/global/foo/bar.txt?t=uri&replace=false", new_uri)
+ d.addBoth(self.shouldFail, error.Error, "PUT_NEWFILEURL_uri_no_replace",
+ "409 Conflict",
+ "There was already a child by that name, and you asked me "
+ "to not replace it")
+ return d
+
def test_XMLRPC(self):
raise unittest.SkipTest("not yet")
pass
class BlockingFileError(Exception):
"""We cannot auto-create a parent directory, because there is a file in
the way"""
+class NoReplacementError(Exception):
+ """There was already a child by that name, and you asked me to not replace it"""
LOCALHOST = "127.0.0.1"
return ctx.tag
class POSTHandler(rend.Page):
- def __init__(self, node):
+ def __init__(self, node, replace):
self._node = node
+ self._replace = replace
def renderHTTP(self, ctx):
req = inevow.IRequest(ctx)
return d
class PUTHandler(rend.Page):
- def __init__(self, node, path, t, localfile, localdir):
+ def __init__(self, node, path, t, localfile, localdir, replace):
self._node = node
self._path = path
self._t = t
self._localfile = localfile
self._localdir = localdir
+ self._replace = replace
def renderHTTP(self, ctx):
req = inevow.IRequest(ctx)
# we must traverse the path, creating new directories as necessary
d = self._get_or_create_directories(self._node, self._path[:-1])
name = self._path[-1]
+ d.addCallback(self._check_replacement, name, self._replace)
if t == "upload":
if localfile:
d.addCallback(self._upload_localfile, localfile, name)
req.setHeader("content-type", "text/plain")
return str(f.value)
d.addErrback(_check_blocking)
+ def _check_replacement(f):
+ f.trap(NoReplacementError)
+ req.setResponseCode(http.CONFLICT)
+ req.setHeader("content-type", "text/plain")
+ return str(f.value)
+ d.addErrback(_check_replacement)
return d
def _get_or_create_directories(self, node, path):
d.addCallback(self._get_or_create_directories, path[1:])
return d
+ def _check_replacement(self, node, name, replace):
+ if replace:
+ return node
+ d = node.has_child(name)
+ def _got(present):
+ if present:
+ raise NoReplacementError("There was already a child by that "
+ "name, and you asked me to not "
+ "replace it.")
+ return node
+ d.addCallback(_got)
+ return d
+
def _mkdir(self, node, name):
d = node.create_empty_directory(name)
def _done(newnode):
return NeedLocalhostError(), ()
# TODO: think about clobbering/revealing config files and node secrets
+ replace = True
+# if "replace" in req.fields:
+# if req.fields["replace"].value.lower() in ("false", "0"):
+# replace = False
+ if "replace" in req.args:
+ if req.args["replace"][0].lower() in ("false", "0"):
+ replace = False
+
if method == "GET":
# the node must exist, and our operation will be performed on the
# node itself.
# node itself.
d = self.get_child_at_path(path)
def _got(node):
- return POSTHandler(node), ()
+ return POSTHandler(node, replace), ()
d.addCallback(_got)
elif method == "DELETE":
# the node must exist, and our operation will be performed on its
elif method in ("PUT",):
# the node may or may not exist, and our operation may involve
# all the ancestors of the node.
- return PUTHandler(self.node, path, t, localfile, localdir), ()
+ return PUTHandler(self.node, path, t, localfile, localdir, replace), ()
else:
return rend.NotFound
def _trap_KeyError(f):
projects / tahoe-lafs / tahoe-lafs.git / commitdiff