self.s = FakeClient()
self.s.startService()
self.ws = s = webish.WebishServer("0")
- s.allow_local_access(True)
s.setServiceParent(self.s)
port = s.listener._port.getHost().port
self.webish_url = "http://localhost:%d" % port
d.addBoth(self.should404, "test_GET_FILEURL_json_missing")
return d
- def disable_local_access(self, res=None):
- self.ws.allow_local_access(False)
- return res
-
- def test_GET_FILEURL_localfile(self):
- localfile = os.path.abspath("web/GET_FILEURL_local file")
- url = (self.public_url + "/foo/bar.txt?t=download&localfile=%s" %
- urllib.quote(localfile))
- fileutil.make_dirs("web")
- d = self.GET(url)
- def _done(res):
- self.failUnless(os.path.exists(localfile))
- data = open(localfile, "rb").read()
- self.failUnlessEqual(data, self.BAR_CONTENTS)
- d.addCallback(_done)
- return d
-
- def test_GET_FILEURL_localfile_disabled(self):
- localfile = os.path.abspath("web/GET_FILEURL_local file_disabled")
- url = (self.public_url + "/foo/bar.txt?t=download&localfile=%s" %
- urllib.quote(localfile))
- fileutil.make_dirs("web")
- self.disable_local_access()
- d = self.GET(url)
- d.addBoth(self.shouldFail, error.Error, "localfile disabled",
- "403 Forbidden",
- "local file access is disabled")
- return d
-
- def test_GET_FILEURL_localfile_nonlocal(self):
- # TODO: somehow pretend that we aren't local, and verify that the
- # server refuses to write to local files, probably by changing the
- # server's idea of what counts as "local".
- old_LOCALHOST = webish.LOCALHOST
- webish.LOCALHOST = "127.0.0.2"
- localfile = os.path.abspath("web/GET_FILEURL_local file_nonlocal")
- fileutil.make_dirs("web")
- d = self.GET(self.public_url + "/foo/bar.txt?t=download&localfile=%s"
- % urllib.quote(localfile))
- d.addBoth(self.shouldFail, error.Error, "localfile non-local",
- "403 Forbidden",
- "localfile= or localdir= requires a local connection")
- def _check(res):
- self.failIf(os.path.exists(localfile))
- d.addCallback(_check)
- def _reset(res):
- webish.LOCALHOST = old_LOCALHOST
- return res
- d.addBoth(_reset)
- return d
-
- def test_GET_FILEURL_localfile_nonabsolute(self):
- localfile = "web/nonabsolute/path"
- fileutil.make_dirs("web/nonabsolute")
- d = self.GET(self.public_url + "/foo/bar.txt?t=download&localfile=%s"
- % urllib.quote(localfile))
- d.addBoth(self.shouldFail, error.Error, "localfile non-absolute",
- "403 Forbidden",
- "localfile= or localdir= requires an absolute path")
- def _check(res):
- self.failIf(os.path.exists(localfile))
- d.addCallback(_check)
- return d
-
- def test_PUT_NEWFILEURL_localfile(self):
- localfile = os.path.abspath("web/PUT_NEWFILEURL_local file")
- url = (self.public_url + "/foo/new.txt?t=upload&localfile=%s" %
- urllib.quote(localfile))
- fileutil.make_dirs("web")
- f = open(localfile, "wb")
- f.write(self.NEWFILE_CONTENTS)
- f.close()
- d = self.PUT(url, "")
- d.addCallback(self.failUnlessURIMatchesChild, self._foo_node, u"new.txt")
- d.addCallback(lambda res:
- self.failUnlessChildContentsAre(self._foo_node, u"new.txt",
- self.NEWFILE_CONTENTS))
- return d
-
- def test_PUT_NEWFILEURL_localfile_missingarg(self):
- url = self.public_url + "/foo/new.txt?t=upload"
- d = self.shouldHTTPError2("test_PUT_NEWFILEURL_localfile_missing",
- 400, "Bad Request",
- "t=upload requires localfile= or localdir=",
- self.PUT, url, "")
- return d
-
- def test_PUT_NEWFILEURL_localfile_disabled(self):
- localfile = os.path.abspath("web/PUT_NEWFILEURL_local file_disabled")
- url = (self.public_url + "/foo/new.txt?t=upload&localfile=%s" %
- urllib.quote(localfile))
- fileutil.make_dirs("web")
- f = open(localfile, "wb")
- f.write(self.NEWFILE_CONTENTS)
- f.close()
- self.disable_local_access()
- d = self.PUT(url, "")
- d.addBoth(self.shouldFail, error.Error, "put localfile disabled",
- "403 Forbidden",
- "local file access is disabled")
- return d
-
- def test_PUT_NEWFILEURL_localfile_mkdirs(self):
- localfile = os.path.abspath("web/PUT_NEWFILEURL_local file_mkdirs")
- fileutil.make_dirs("web")
- f = open(localfile, "wb")
- f.write(self.NEWFILE_CONTENTS)
- f.close()
- d = self.PUT(self.public_url + "/foo/newdir/new.txt?t=upload&localfile=%s"
- % urllib.quote(localfile), "")
- d.addCallback(self.failUnlessURIMatchesChild,
- self._foo_node, u"newdir/new.txt")
- d.addCallback(lambda res:
- self.failIfNodeHasChild(self._foo_node, u"new.txt"))
- d.addCallback(lambda res:
- self.failUnlessNodeHasChild(self._foo_node, u"newdir"))
- d.addCallback(lambda res:
- self.failUnlessChildContentsAre(self._foo_node,
- u"newdir/new.txt",
- self.NEWFILE_CONTENTS))
- return d
-
def test_GET_FILEURL_uri(self):
d = self.GET(self.public_url + "/foo/bar.txt?t=uri")
def _check(res):
d.addBoth(self.should404, "test_DELETE_DIRURL_missing2")
return d
- def test_walker(self):
- out = []
- def _visitor(path, node, metadata):
- out.append((path, node))
- return defer.succeed(None)
- w = webish.DirnodeWalkerMixin()
- d = w.walk(self.public_root, _visitor)
- def _check(res):
- names = [path for (path,node) in out]
- self.failUnlessEqual(sorted(names),
- [(u'foo',),
- (u'foo',u'bar.txt'),
- (u'foo',u'blockingfile'),
- (u'foo', u'empty'),
- (u'foo', u"n\u00fc.txt"),
- (u'foo', u'sub'),
- (u'foo',u'sub',u'baz.txt'),
- (u'reedownlee',),
- (u'reedownlee', u'nor'),
- ])
- subindex = names.index( (u'foo', u'sub') )
- bazindex = names.index( (u'foo', u'sub', u'baz.txt') )
- self.failUnless(subindex < bazindex)
- for path,node in out:
- if path[-1] in (u'bar.txt', u"n\u00fc.txt", u'blockingfile',
- u'baz.txt', u'nor'):
- self.failUnless(interfaces.IFileNode.providedBy(node))
- else:
- self.failUnless(interfaces.IDirectoryNode.providedBy(node))
- d.addCallback(_check)
- return d
-
- def test_GET_DIRURL_localdir(self):
- localdir = os.path.abspath("web/GET_DIRURL_local dir")
- fileutil.make_dirs("web")
- d = self.GET(self.public_url + "/foo?t=download&localdir=%s" %
- urllib.quote(localdir))
- def _check(res):
- barfile = os.path.join(localdir, "bar.txt")
- self.failUnless(os.path.exists(barfile))
- data = open(barfile, "rb").read()
- self.failUnlessEqual(data, self.BAR_CONTENTS)
- blockingfile = os.path.join(localdir, "blockingfile")
- self.failUnless(os.path.exists(blockingfile))
- subdir = os.path.join(localdir, "sub")
- self.failUnless(os.path.isdir(subdir))
- d.addCallback(_check)
- return d
-
- def test_GET_DIRURL_localdir_disabled(self):
- localdir = os.path.abspath("web/GET_DIRURL_local dir_disabled")
- fileutil.make_dirs("web")
- self.disable_local_access()
- d = self.GET(self.public_url + "/foo?t=download&localdir=%s" %
- urllib.quote(localdir))
- d.addBoth(self.shouldFail, error.Error, "localfile disabled",
- "403 Forbidden",
- "local file access is disabled")
- return d
-
- def test_GET_DIRURL_localdir_nonabsolute(self):
- localdir = "web/nonabsolute/dir path"
- fileutil.make_dirs("web/nonabsolute")
- d = self.GET(self.public_url + "/foo?t=download&localdir=%s" %
- urllib.quote(localdir))
- d.addBoth(self.shouldFail, error.Error, "localdir non-absolute",
- "403 Forbidden",
- "localfile= or localdir= requires an absolute path")
- def _check(res):
- self.failIf(os.path.exists(localdir))
- d.addCallback(_check)
- return d
-
- def test_GET_DIRURL_localdir_nolocaldir(self):
- d = self.shouldHTTPError2("GET_DIRURL_localdir_nolocaldir",
- 400, "Bad Request",
- "t=download requires localdir=",
- self.GET,
- self.public_url + "/foo?t=download")
- return d
-
- def touch(self, localdir, filename):
- path = os.path.join(localdir, filename)
- f = open(path, "wb")
- f.write("contents of %s\n" % filename)
- f.close()
-
def dump_root(self):
print "NODEWALK"
w = webish.DirnodeWalkerMixin()
def failUnlessCHKURIHasContents(self, got_uri, contents):
self.failUnless(FakeCHKFileNode.all_contents[got_uri] == contents)
- def test_PUT_NEWDIRURL_localdir(self):
- localdir = os.path.abspath("web/PUT_NEWDIRURL_local dir")
- # create some files there
- fileutil.make_dirs(os.path.join(localdir, "one"))
- fileutil.make_dirs(os.path.join(localdir, "one/sub"))
- fileutil.make_dirs(os.path.join(localdir, "two"))
- fileutil.make_dirs(os.path.join(localdir, "three"))
- self.touch(localdir, "three/foo.txt")
- self.touch(localdir, "three/bar.txt")
- self.touch(localdir, "zap.zip")
-
- d = self.PUT(self.public_url + "/newdir?t=upload&localdir=%s"
- % urllib.quote(localdir), "")
- pr = self.public_root
- d.addCallback(lambda res: self.failUnlessNodeHasChild(pr, u"newdir"))
- d.addCallback(lambda res: pr.get(u"newdir"))
- d.addCallback(self.failUnlessNodeKeysAre,
- [u"one", u"two", u"three", u"zap.zip"])
- d.addCallback(lambda res: pr.get_child_at_path(u"newdir/one"))
- d.addCallback(self.failUnlessNodeKeysAre, [u"sub"])
- d.addCallback(lambda res: pr.get_child_at_path(u"newdir/three"))
- d.addCallback(self.failUnlessNodeKeysAre, [u"foo.txt", u"bar.txt"])
- d.addCallback(lambda res: pr.get_child_at_path(u"newdir/three/bar.txt"))
- d.addCallback(lambda barnode: barnode.download_to_data())
- d.addCallback(lambda contents:
- self.failUnlessEqual(contents,
- "contents of three/bar.txt\n"))
- return d
-
- def test_PUT_NEWDIRURL_localdir_disabled(self):
- localdir = os.path.abspath("web/PUT_NEWDIRURL_local dir_disabled")
- # create some files there
- fileutil.make_dirs(os.path.join(localdir, "one"))
- fileutil.make_dirs(os.path.join(localdir, "one/sub"))
- fileutil.make_dirs(os.path.join(localdir, "two"))
- fileutil.make_dirs(os.path.join(localdir, "three"))
- self.touch(localdir, "three/foo.txt")
- self.touch(localdir, "three/bar.txt")
- self.touch(localdir, "zap.zip")
-
- self.disable_local_access()
- d = self.PUT(self.public_url + "/newdir?t=upload&localdir=%s"
- % urllib.quote(localdir), "")
- d.addBoth(self.shouldFail, error.Error, "localfile disabled",
- "403 Forbidden",
- "local file access is disabled")
- return d
-
- def test_PUT_NEWDIRURL_localdir_mkdirs(self):
- localdir = os.path.abspath("web/PUT_NEWDIRURL_local dir_mkdirs")
- # create some files there
- fileutil.make_dirs(os.path.join(localdir, "one"))
- fileutil.make_dirs(os.path.join(localdir, "one/sub"))
- fileutil.make_dirs(os.path.join(localdir, "two"))
- fileutil.make_dirs(os.path.join(localdir, "three"))
- self.touch(localdir, "three/foo.txt")
- self.touch(localdir, "three/bar.txt")
- self.touch(localdir, "zap.zip")
-
- d = self.PUT(self.public_url + "/foo/subdir/newdir?t=upload&localdir=%s"
- % urllib.quote(localdir),
- "")
- fn = self._foo_node
- d.addCallback(lambda res: self.failUnlessNodeHasChild(fn, u"subdir"))
- d.addCallback(lambda res: fn.get_child_at_path(u"subdir/newdir"))
- d.addCallback(self.failUnlessNodeKeysAre,
- [u"one", u"two", u"three", u"zap.zip"])
- d.addCallback(lambda res: fn.get_child_at_path(u"subdir/newdir/one"))
- d.addCallback(self.failUnlessNodeKeysAre, [u"sub"])
- d.addCallback(lambda res: fn.get_child_at_path(u"subdir/newdir/three"))
- d.addCallback(self.failUnlessNodeKeysAre, [u"foo.txt", u"bar.txt"])
- d.addCallback(lambda res:
- fn.get_child_at_path(u"subdir/newdir/three/bar.txt"))
- d.addCallback(lambda barnode: barnode.download_to_data())
- d.addCallback(lambda contents:
- self.failUnlessEqual(contents,
- "contents of three/bar.txt\n"))
- return d
-
- def test_PUT_NEWDIRURL_localdir_missing(self):
- localdir = os.path.abspath("web/PUT_NEWDIRURL_localdir_missing")
- # we do *not* create it, to trigger an error
- url = (self.public_url + "/foo/subdir/newdir?t=upload&localdir=%s"
- % urllib.quote(localdir))
- d = self.shouldHTTPError2("test_PUT_NEWDIRURL_localdir_missing",
- 400, "Bad Request",
- "%s doesn't exist!" % localdir,
- self.PUT, url, "")
- return d
-
def test_POST_upload(self):
d = self.POST(self.public_url + "/foo", t="upload",
file=("new.txt", self.NEWFILE_CONTENTS))
projects / tahoe-lafs / tahoe-lafs.git / commitdiff