d.addCallback(self.failUnlessNodeKeysAre, [u"baz.txt"])
return d
- def test_PUT_NEWDIRURL_mkdir_p(self):
- d = defer.succeed(None)
- d.addCallback(lambda res: self.POST(self.public_url + "/foo", t='mkdir', name='mkp'))
- d.addCallback(lambda res: self.failUnlessNodeHasChild(self._foo_node, u"mkp"))
- d.addCallback(lambda res: self._foo_node.get(u"mkp"))
- def mkdir_p(mkpnode):
- url = '/uri/%s?t=mkdir-p&path=/sub1/sub2' % urllib.quote(mkpnode.get_uri())
- d = self.POST(url)
- def made_subsub(ssuri):
- d = self._foo_node.get_child_at_path(u"mkp/sub1/sub2")
- d.addCallback(lambda ssnode: self.failUnlessReallyEqual(ssnode.get_uri(), ssuri))
- d = self.POST(url)
- d.addCallback(lambda uri2: self.failUnlessReallyEqual(uri2, ssuri))
- return d
- d.addCallback(made_subsub)
- return d
- d.addCallback(mkdir_p)
- return d
-
def test_PUT_NEWDIRURL_mkdirs(self):
d = self.PUT(self.public_url + "/foo/subdir/newdir?t=mkdir", "")
d.addCallback(lambda res:
d = self._POST_mkdir_with_children(req)
elif t == "mkdir-immutable":
d = self._POST_mkdir_immutable(req)
- elif t == "mkdir-p":
- # TODO: docs, tests
- d = self._POST_mkdir_p(req)
elif t == "upload":
d = self._POST_upload(ctx) # this one needs the context
elif t == "uri":
d.addCallback(lambda child: child.get_uri()) # TODO: urlencode
return d
- def _POST_mkdir_p(self, req):
- path = get_arg(req, "path")
- if not path:
- raise WebError("mkdir-p requires a path")
- path_ = tuple([seg.decode("utf-8") for seg in path.split('/') if seg ])
- # TODO: replace
- d = self._get_or_create_directories(self.node, path_)
- d.addCallback(lambda node: node.get_uri())
- return d
-
- def _get_or_create_directories(self, node, path):
- if not IDirectoryNode.providedBy(node):
- # unfortunately it is too late to provide the name of the
- # blocking directory in the error message.
- raise BlockingFileError("cannot create directory because there "
- "is a file in the way")
- if not path:
- return defer.succeed(node)
- d = node.get(path[0])
- def _maybe_create(f):
- f.trap(NoSuchChildError)
- return node.create_subdirectory(path[0])
- d.addErrback(_maybe_create)
- d.addCallback(self._get_or_create_directories, path[1:])
- return d
-
def _POST_upload(self, ctx):
req = IRequest(ctx)
charset = get_arg(req, "_charset", "utf-8")