class FakeUploader(service.Service):
name = "uploader"
- def upload(self, uploadable, history=None):
+ def upload(self, uploadable):
d = uploadable.get_size()
d.addCallback(lambda size: uploadable.read(size))
def _got_data(datav):
self.history = FakeHistory()
self.uploader = FakeUploader()
self.uploader.setServiceParent(self)
+ self.blacklist = None
self.nodemaker = FakeNodeMaker(None, self._secret_holder, None,
self.uploader, None,
None, None)
return d
def test_GET_FILE_URI_mdmf_extensions(self):
- base = "/uri/%s" % urllib.quote("%s:3:131073" % self._quux_txt_uri)
- d = self.GET(base)
- d.addCallback(self.failUnlessIsQuuxDotTxt)
- return d
-
- def test_GET_FILE_URI_mdmf_bare_cap(self):
- cap_elements = self._quux_txt_uri.split(":")
- # 6 == expected cap length with two extensions.
- self.failUnlessEqual(len(cap_elements), 6)
-
- # Now lop off the extension parameters and stitch everything
- # back together
- quux_uri = ":".join(cap_elements[:len(cap_elements) - 2])
-
- # Now GET that. We should get back quux.
- base = "/uri/%s" % urllib.quote(quux_uri)
+ base = "/uri/%s" % urllib.quote("%s:RANDOMSTUFF" % self._quux_txt_uri)
d = self.GET(base)
d.addCallback(self.failUnlessIsQuuxDotTxt)
return d
return d
def test_PUT_FILE_URI_mdmf_extensions(self):
- base = "/uri/%s" % urllib.quote("%s:3:131073" % self._quux_txt_uri)
+ base = "/uri/%s" % urllib.quote("%s:EXTENSIONSTUFF" % self._quux_txt_uri)
self._quux_new_contents = "new_contents"
d = self.GET(base)
d.addCallback(lambda res: self.failUnlessIsQuuxDotTxt(res))
res))
return d
- def test_PUT_FILE_URI_mdmf_bare_cap(self):
- elements = self._quux_txt_uri.split(":")
- self.failUnlessEqual(len(elements), 6)
-
- quux_uri = ":".join(elements[:len(elements) - 2])
- base = "/uri/%s" % urllib.quote(quux_uri)
- self._quux_new_contents = "new_contents" * 50000
-
- d = self.GET(base)
- d.addCallback(self.failUnlessIsQuuxDotTxt)
- d.addCallback(lambda ignored: self.PUT(base, self._quux_new_contents))
- d.addCallback(lambda ignored: self.GET(base))
- d.addCallback(lambda res:
- self.failUnlessEqual(res, self._quux_new_contents))
- return d
-
def test_PUT_FILE_URI_mdmf_readonly(self):
# We're not allowed to PUT things to a readonly cap.
base = "/uri/%s" % self._quux_txt_readonly_uri
return d
def test_GET_FILEURL_info_mdmf_extensions(self):
- d = self.GET("/uri/%s:3:131073?t=info" % self._quux_txt_uri)
+ d = self.GET("/uri/%s:STUFF?t=info" % self._quux_txt_uri)
def _got(res):
self.failUnlessIn("mutable file (mdmf)", res)
self.failUnlessIn(self._quux_txt_uri, res)
d.addCallback(_got)
return d
- def test_GET_FILEURL_info_mdmf_bare_cap(self):
- elements = self._quux_txt_uri.split(":")
- self.failUnlessEqual(len(elements), 6)
-
- quux_uri = ":".join(elements[:len(elements) - 2])
- base = "/uri/%s?t=info" % urllib.quote(quux_uri)
- d = self.GET(base)
- def _got(res):
- self.failUnlessIn("mutable file (mdmf)", res)
- self.failUnlessIn(quux_uri, res)
- d.addCallback(_got)
- return d
-
def test_PUT_overwrite_only_files(self):
# create a directory, put a file in that directory.
contents, n, filecap = self.makefile(8)
# this should get us a few segments of an MDMF mutable file,
# which we can then test for.
contents = self.NEWFILE_CONTENTS * 300000
- d = self.PUT("/uri?mutable=true&mutable-type=mdmf",
+ d = self.PUT("/uri?format=mdmf",
contents)
def _got_filecap(filecap):
self.failUnless(filecap.startswith("URI:MDMF"))
return filecap
d.addCallback(_got_filecap)
d.addCallback(lambda filecap: self.GET("/uri/%s?t=json" % filecap))
- d.addCallback(lambda json: self.failUnlessIn("mdmf", json))
+ d.addCallback(lambda json: self.failUnlessIn("MDMF", json))
return d
def test_PUT_NEWFILEURL_unlinked_sdmf(self):
contents = self.NEWFILE_CONTENTS * 300000
- d = self.PUT("/uri?mutable=true&mutable-type=sdmf",
+ d = self.PUT("/uri?format=sdmf",
contents)
d.addCallback(lambda filecap: self.GET("/uri/%s?t=json" % filecap))
- d.addCallback(lambda json: self.failUnlessIn("sdmf", json))
+ d.addCallback(lambda json: self.failUnlessIn("SDMF", json))
return d
- def test_PUT_NEWFILEURL_unlinked_bad_mutable_type(self):
+ def test_PUT_NEWFILEURL_unlinked_bad_format(self):
contents = self.NEWFILE_CONTENTS * 300000
- return self.shouldHTTPError("test bad mutable type",
- 400, "Bad Request", "Unknown type: foo",
- self.PUT, "/uri?mutable=true&mutable-type=foo",
+ return self.shouldHTTPError("PUT_NEWFILEURL_unlinked_bad_format",
+ 400, "Bad Request", "Unknown format: foo",
+ self.PUT, "/uri?format=foo",
contents)
def test_PUT_NEWFILEURL_range_bad(self):
return d
def test_GET_FILEURL_json_mutable_type(self):
- # The JSON should include mutable-type, which says whether the
+ # The JSON should include format, which says whether the
# file is SDMF or MDMF
- d = self.PUT("/uri?mutable=true&mutable-type=mdmf",
+ d = self.PUT("/uri?format=mdmf",
self.NEWFILE_CONTENTS * 300000)
d.addCallback(lambda filecap: self.GET("/uri/%s?t=json" % filecap))
def _got_json(json, version):
data = data[1]
assert isinstance(data, dict)
- self.failUnlessIn("mutable-type", data)
- self.failUnlessEqual(data['mutable-type'], version)
+ self.failUnlessIn("format", data)
+ self.failUnlessEqual(data["format"], version)
- d.addCallback(_got_json, "mdmf")
+ d.addCallback(_got_json, "MDMF")
# Now make an SDMF file and check that it is reported correctly.
d.addCallback(lambda ignored:
- self.PUT("/uri?mutable=true&mutable-type=sdmf",
+ self.PUT("/uri?format=sdmf",
self.NEWFILE_CONTENTS * 300000))
d.addCallback(lambda filecap: self.GET("/uri/%s?t=json" % filecap))
- d.addCallback(_got_json, "sdmf")
- return d
-
- def test_GET_FILEURL_json_mdmf_extensions(self):
- # A GET invoked against a URL that includes an MDMF cap with
- # extensions should fetch the same JSON information as a GET
- # invoked against a bare cap.
- self._quux_txt_uri = "%s:3:131073" % self._quux_txt_uri
- self._quux_txt_readonly_uri = "%s:3:131073" % self._quux_txt_readonly_uri
- d = self.GET("/uri/%s?t=json" % urllib.quote(self._quux_txt_uri))
- d.addCallback(self.failUnlessIsQuuxJSON)
- return d
-
- def test_GET_FILEURL_json_mdmf_bare_cap(self):
- elements = self._quux_txt_uri.split(":")
- self.failUnlessEqual(len(elements), 6)
-
- quux_uri = ":".join(elements[:len(elements) - 2])
- # so failUnlessIsQuuxJSON will work.
- self._quux_txt_uri = quux_uri
-
- # we need to alter the readonly URI in the same way, again so
- # failUnlessIsQuuxJSON will work
- elements = self._quux_txt_readonly_uri.split(":")
- self.failUnlessEqual(len(elements), 6)
- quux_ro_uri = ":".join(elements[:len(elements) - 2])
- self._quux_txt_readonly_uri = quux_ro_uri
-
- base = "/uri/%s?t=json" % urllib.quote(quux_uri)
- d = self.GET(base)
- d.addCallback(self.failUnlessIsQuuxJSON)
- return d
-
- def test_GET_FILEURL_json_mdmf_bare_readonly_cap(self):
- elements = self._quux_txt_readonly_uri.split(":")
- self.failUnlessEqual(len(elements), 6)
-
- quux_readonly_uri = ":".join(elements[:len(elements) - 2])
- # so failUnlessIsQuuxJSON will work
- self._quux_txt_readonly_uri = quux_readonly_uri
- base = "/uri/%s?t=json" % quux_readonly_uri
- d = self.GET(base)
- # XXX: We may need to make a method that knows how to check for
- # readonly JSON, or else alter that one so that it knows how to
- # do that.
- d.addCallback(self.failUnlessIsQuuxJSON, readonly=True)
+ d.addCallback(_got_json, "SDMF")
return d
def test_GET_FILEURL_json_mdmf(self):
d.addBoth(self.should404, "test_GET_FILEURL_uri_missing")
return d
+ def _check_upload_and_mkdir_forms(self, html):
+ # We should have a form to create a file, with radio buttons that allow
+ # the user to toggle whether it is a CHK/LIT (default), SDMF, or MDMF file.
+ self.failUnlessIn('name="t" value="upload"', html)
+ self.failUnlessIn('input checked="checked" type="radio" id="upload-chk" value="chk" name="format"', html)
+ self.failUnlessIn('input type="radio" id="upload-sdmf" value="sdmf" name="format"', html)
+ self.failUnlessIn('input type="radio" id="upload-mdmf" value="mdmf" name="format"', html)
+
+ # We should also have the ability to create a mutable directory, with
+ # radio buttons that allow the user to toggle whether it is an SDMF (default)
+ # or MDMF directory.
+ self.failUnlessIn('name="t" value="mkdir"', html)
+ self.failUnlessIn('input checked="checked" type="radio" id="mkdir-sdmf" value="sdmf" name="format"', html)
+ self.failUnlessIn('input type="radio" id="mkdir-mdmf" value="mdmf" name="format"', html)
+
def test_GET_DIRECTORY_html(self):
d = self.GET(self.public_url + "/foo", followRedirect=True)
- def _check(res):
- self.failUnlessIn('<div class="toolbar-item"><a href="../../..">Return to Welcome page</a></div>',res)
- # These are radio buttons that allow a user to toggle
- # whether a particular mutable file is SDMF or MDMF.
- self.failUnlessIn("mutable-type-mdmf", res)
- self.failUnlessIn("mutable-type-sdmf", res)
- # Similarly, these toggle whether a particular directory
- # should be MDMF or SDMF.
- self.failUnlessIn("mutable-directory-mdmf", res)
- self.failUnlessIn("mutable-directory-sdmf", res)
- self.failUnlessIn("quux", res)
+ def _check(html):
+ self.failUnlessIn('<div class="toolbar-item"><a href="../../..">Return to Welcome page</a></div>', html)
+ self._check_upload_and_mkdir_forms(html)
+ self.failUnlessIn("quux", html)
d.addCallback(_check)
return d
def test_GET_root_html(self):
- # make sure that we have the option to upload an unlinked
- # mutable file in SDMF and MDMF formats.
- d = self.GET("/")
- def _got_html(html):
- # These are radio buttons that allow the user to toggle
- # whether a particular mutable file is MDMF or SDMF.
- self.failUnlessIn("mutable-type-mdmf", html)
- self.failUnlessIn("mutable-type-sdmf", html)
- # We should also have the ability to create a mutable directory.
- self.failUnlessIn("mkdir", html)
- # ...and we should have the ability to say whether that's an
- # MDMF or SDMF directory
- self.failUnlessIn("mutable-directory-mdmf", html)
- self.failUnlessIn("mutable-directory-sdmf", html)
- d.addCallback(_got_html)
- return d
-
- def test_mutable_type_defaults(self):
- # The checked="checked" attribute of the inputs corresponding to
- # the mutable-type parameter should change as expected with the
- # value configured in tahoe.cfg.
- #
- # By default, the value configured with the client is
- # SDMF_VERSION, so that should be checked.
- assert self.s.mutable_file_default == SDMF_VERSION
-
d = self.GET("/")
- def _got_html(html, value):
- i = 'input checked="checked" type="radio" id="mutable-type-%s"'
- self.failUnlessIn(i % value, html)
- d.addCallback(_got_html, "sdmf")
- d.addCallback(lambda ignored:
- self.GET(self.public_url + "/foo", followRedirect=True))
- d.addCallback(_got_html, "sdmf")
- # Now switch the configuration value to MDMF. The MDMF radio
- # buttons should now be checked on these pages.
- def _swap_values(ignored):
- self.s.mutable_file_default = MDMF_VERSION
- d.addCallback(_swap_values)
- d.addCallback(lambda ignored: self.GET("/"))
- d.addCallback(_got_html, "mdmf")
- d.addCallback(lambda ignored:
- self.GET(self.public_url + "/foo", followRedirect=True))
- d.addCallback(_got_html, "mdmf")
+ d.addCallback(self._check_upload_and_mkdir_forms)
return d
def test_GET_DIRURL(self):
d.addCallback(self.failUnlessIsFooJSON)
return d
- def test_GET_DIRURL_json_mutable_type(self):
+ def test_GET_DIRURL_json_format(self):
d = self.PUT(self.public_url + \
- "/foo/sdmf.txt?mutable=true&mutable-type=sdmf",
+ "/foo/sdmf.txt?format=sdmf",
self.NEWFILE_CONTENTS * 300000)
d.addCallback(lambda ignored:
self.PUT(self.public_url + \
- "/foo/mdmf.txt?mutable=true&mutable-type=mdmf",
+ "/foo/mdmf.txt?format=mdmf",
self.NEWFILE_CONTENTS * 300000))
# Now we have an MDMF and SDMF file in the directory. If we GET
# its JSON, we should see their encodings.
kids = data['children']
mdmf_data = kids['mdmf.txt'][1]
- self.failUnlessIn("mutable-type", mdmf_data)
- self.failUnlessEqual(mdmf_data['mutable-type'], "mdmf")
+ self.failUnlessIn("format", mdmf_data)
+ self.failUnlessEqual(mdmf_data["format"], "MDMF")
sdmf_data = kids['sdmf.txt'][1]
- self.failUnlessIn("mutable-type", sdmf_data)
- self.failUnlessEqual(sdmf_data['mutable-type'], "sdmf")
+ self.failUnlessIn("format", sdmf_data)
+ self.failUnlessEqual(sdmf_data["format"], "SDMF")
d.addCallback(_got_json)
return d
return d
def test_PUT_NEWDIRURL_mdmf(self):
- d = self.PUT(self.public_url + "/foo/newdir?t=mkdir&mutable-type=mdmf", "")
+ d = self.PUT(self.public_url + "/foo/newdir?t=mkdir&format=mdmf", "")
d.addCallback(lambda res:
self.failUnlessNodeHasChild(self._foo_node, u"newdir"))
d.addCallback(lambda res: self._foo_node.get(u"newdir"))
return d
def test_PUT_NEWDIRURL_sdmf(self):
- d = self.PUT(self.public_url + "/foo/newdir?t=mkdir&mutable-type=sdmf",
+ d = self.PUT(self.public_url + "/foo/newdir?t=mkdir&format=sdmf",
"")
d.addCallback(lambda res:
self.failUnlessNodeHasChild(self._foo_node, u"newdir"))
self.failUnlessEqual(node._node.get_version(), SDMF_VERSION))
return d
- def test_PUT_NEWDIRURL_bad_mutable_type(self):
- return self.shouldHTTPError("test bad mutable type",
- 400, "Bad Request", "Unknown type: foo",
- self.PUT, self.public_url + \
- "/foo/newdir=?t=mkdir&mutable-type=foo", "")
+ def test_PUT_NEWDIRURL_bad_format(self):
+ return self.shouldHTTPError("PUT_NEWDIRURL_bad_format",
+ 400, "Bad Request", "Unknown format: foo",
+ self.PUT, self.public_url +
+ "/foo/newdir=?t=mkdir&format=foo", "")
def test_POST_NEWDIRURL(self):
d = self.POST2(self.public_url + "/foo/newdir?t=mkdir", "")
return d
def test_POST_NEWDIRURL_mdmf(self):
- d = self.POST2(self.public_url + "/foo/newdir?t=mkdir&mutable-type=mdmf", "")
+ d = self.POST2(self.public_url + "/foo/newdir?t=mkdir&format=mdmf", "")
d.addCallback(lambda res:
self.failUnlessNodeHasChild(self._foo_node, u"newdir"))
d.addCallback(lambda res: self._foo_node.get(u"newdir"))
return d
def test_POST_NEWDIRURL_sdmf(self):
- d = self.POST2(self.public_url + "/foo/newdir?t=mkdir&mutable-type=sdmf", "")
+ d = self.POST2(self.public_url + "/foo/newdir?t=mkdir&format=sdmf", "")
d.addCallback(lambda res:
self.failUnlessNodeHasChild(self._foo_node, u"newdir"))
d.addCallback(lambda res: self._foo_node.get(u"newdir"))
self.failUnlessEqual(node._node.get_version(), SDMF_VERSION))
return d
- def test_POST_NEWDIRURL_bad_mutable_type(self):
- return self.shouldHTTPError("test bad mutable type",
- 400, "Bad Request", "Unknown type: foo",
+ def test_POST_NEWDIRURL_bad_format(self):
+ return self.shouldHTTPError("POST_NEWDIRURL_bad_format",
+ 400, "Bad Request", "Unknown format: foo",
self.POST2, self.public_url + \
- "/foo/newdir?t=mkdir&mutable-type=foo", "")
+ "/foo/newdir?t=mkdir&format=foo", "")
def test_POST_NEWDIRURL_emptyname(self):
# an empty pathname component (i.e. a double-slash) is disallowed
- d = self.shouldFail2(error.Error, "test_POST_NEWDIRURL_emptyname",
+ d = self.shouldFail2(error.Error, "POST_NEWDIRURL_emptyname",
"400 Bad Request",
"The webapi does not allow empty pathname components, i.e. a double slash",
self.POST, self.public_url + "//?t=mkdir")
(newkids, caps) = self._create_initial_children()
query = "/foo/newdir?t=mkdir-with-children"
if version == MDMF_VERSION:
- query += "&mutable-type=mdmf"
+ query += "&format=mdmf"
elif version == SDMF_VERSION:
- query += "&mutable-type=sdmf"
+ query += "&format=sdmf"
else:
version = SDMF_VERSION # for later
d = self.POST2(self.public_url + query,
def test_POST_NEWDIRURL_initial_children_sdmf(self):
return self._do_POST_NEWDIRURL_initial_children_test(SDMF_VERSION)
- def test_POST_NEWDIRURL_initial_children_bad_mutable_type(self):
+ def test_POST_NEWDIRURL_initial_children_bad_format(self):
(newkids, caps) = self._create_initial_children()
- return self.shouldHTTPError("test bad mutable type",
- 400, "Bad Request", "Unknown type: foo",
+ return self.shouldHTTPError("POST_NEWDIRURL_initial_children_bad_format",
+ 400, "Bad Request", "Unknown format: foo",
self.POST2, self.public_url + \
- "/foo/newdir?t=mkdir-with-children&mutable-type=foo",
+ "/foo/newdir?t=mkdir-with-children&format=foo",
simplejson.dumps(newkids))
def test_POST_NEWDIRURL_immutable(self):
return d
def test_PUT_NEWDIRURL_mkdirs_mdmf(self):
- d = self.PUT(self.public_url + "/foo/subdir/newdir?t=mkdir&mutable-type=mdmf", "")
+ d = self.PUT(self.public_url + "/foo/subdir/newdir?t=mkdir&format=mdmf", "")
d.addCallback(lambda ignored:
self.failUnlessNodeHasChild(self._foo_node, u"subdir"))
d.addCallback(lambda ignored:
return d
def test_PUT_NEWDIRURL_mkdirs_sdmf(self):
- d = self.PUT(self.public_url + "/foo/subdir/newdir?t=mkdir&mutable-type=sdmf", "")
+ d = self.PUT(self.public_url + "/foo/subdir/newdir?t=mkdir&format=sdmf", "")
d.addCallback(lambda ignored:
self.failUnlessNodeHasChild(self._foo_node, u"subdir"))
d.addCallback(lambda ignored:
self.failUnlessEqual(newdir._node.get_version(), SDMF_VERSION))
return d
- def test_PUT_NEWDIRURL_mkdirs_bad_mutable_type(self):
- return self.shouldHTTPError("test bad mutable type",
- 400, "Bad Request", "Unknown type: foo",
+ def test_PUT_NEWDIRURL_mkdirs_bad_format(self):
+ return self.shouldHTTPError("PUT_NEWDIRURL_mkdirs_bad_format",
+ 400, "Bad Request", "Unknown format: foo",
self.PUT, self.public_url + \
- "/foo/subdir/newdir?t=mkdir&mutable-type=foo",
+ "/foo/subdir/newdir?t=mkdir&format=foo",
"")
def test_DELETE_DIRURL(self):
return d
- def test_POST_upload_mutable_type_unlinked(self):
- d = self.POST("/uri?t=upload&mutable=true&mutable-type=sdmf",
- file=("sdmf.txt", self.NEWFILE_CONTENTS * 300000))
- d.addCallback(lambda filecap: self.GET("/uri/%s?t=json" % filecap))
- def _got_json(json, version):
- data = simplejson.loads(json)
- data = data[1]
-
- self.failUnlessIn("mutable-type", data)
- self.failUnlessEqual(data['mutable-type'], version)
- d.addCallback(_got_json, "sdmf")
- d.addCallback(lambda ignored:
- self.POST("/uri?t=upload&mutable=true&mutable-type=mdmf",
- file=('mdmf.txt', self.NEWFILE_CONTENTS * 300000)))
- def _got_filecap(filecap):
- self.failUnless(filecap.startswith("URI:MDMF"))
- return filecap
- d.addCallback(_got_filecap)
- d.addCallback(lambda filecap: self.GET("/uri/%s?t=json" % filecap))
- d.addCallback(_got_json, "mdmf")
+ def test_POST_upload_format_unlinked(self):
+ def _check_upload_unlinked(ign, format, uri_prefix):
+ filename = format + ".txt"
+ d = self.POST("/uri?t=upload&format=" + format,
+ file=(filename, self.NEWFILE_CONTENTS * 300000))
+ def _got_results(results):
+ if format.upper() in ("SDMF", "MDMF"):
+ # webapi.rst says this returns a filecap
+ filecap = results
+ else:
+ # for immutable, it returns an "upload results page", and
+ # the filecap is buried inside
+ line = [l for l in results.split("\n") if "URI: " in l][0]
+ mo = re.search(r'<span>([^<]+)</span>', line)
+ filecap = mo.group(1)
+ self.failUnless(filecap.startswith(uri_prefix),
+ (uri_prefix, filecap))
+ return self.GET("/uri/%s?t=json" % filecap)
+ d.addCallback(_got_results)
+ def _got_json(json):
+ data = simplejson.loads(json)
+ data = data[1]
+ self.failUnlessIn("format", data)
+ self.failUnlessEqual(data["format"], format.upper())
+ d.addCallback(_got_json)
+ return d
+ d = defer.succeed(None)
+ d.addCallback(_check_upload_unlinked, "chk", "URI:CHK")
+ d.addCallback(_check_upload_unlinked, "CHK", "URI:CHK")
+ d.addCallback(_check_upload_unlinked, "sdmf", "URI:SSK")
+ d.addCallback(_check_upload_unlinked, "mdmf", "URI:MDMF")
return d
- def test_POST_upload_mutable_type_unlinked_bad_mutable_type(self):
- return self.shouldHTTPError("test bad mutable type",
- 400, "Bad Request", "Unknown type: foo",
+ def test_POST_upload_bad_format_unlinked(self):
+ return self.shouldHTTPError("POST_upload_bad_format_unlinked",
+ 400, "Bad Request", "Unknown format: foo",
self.POST,
- "/uri?5=upload&mutable=true&mutable-type=foo",
+ "/uri?t=upload&format=foo",
file=("foo.txt", self.NEWFILE_CONTENTS * 300000))
- def test_POST_upload_mutable_type(self):
- d = self.POST(self.public_url + \
- "/foo?t=upload&mutable=true&mutable-type=sdmf",
- file=("sdmf.txt", self.NEWFILE_CONTENTS * 300000))
- fn = self._foo_node
- def _got_cap(filecap, filename):
- filenameu = unicode(filename)
- self.failUnlessURIMatchesRWChild(filecap, fn, filenameu)
- return self.GET(self.public_url + "/foo/%s?t=json" % filename)
- def _got_mdmf_cap(filecap):
- self.failUnless(filecap.startswith("URI:MDMF"))
- return filecap
- d.addCallback(_got_cap, "sdmf.txt")
- def _got_json(json, version):
- data = simplejson.loads(json)
- data = data[1]
+ def test_POST_upload_format(self):
+ def _check_upload(ign, format, uri_prefix, fn=None):
+ filename = format + ".txt"
+ d = self.POST(self.public_url +
+ "/foo?t=upload&format=" + format,
+ file=(filename, self.NEWFILE_CONTENTS * 300000))
+ def _got_filecap(filecap):
+ if fn is not None:
+ filenameu = unicode(filename)
+ self.failUnlessURIMatchesRWChild(filecap, fn, filenameu)
+ self.failUnless(filecap.startswith(uri_prefix))
+ return self.GET(self.public_url + "/foo/%s?t=json" % filename)
+ d.addCallback(_got_filecap)
+ def _got_json(json):
+ data = simplejson.loads(json)
+ data = data[1]
+ self.failUnlessIn("format", data)
+ self.failUnlessEqual(data["format"], format.upper())
+ d.addCallback(_got_json)
+ return d
- self.failUnlessIn("mutable-type", data)
- self.failUnlessEqual(data['mutable-type'], version)
- d.addCallback(_got_json, "sdmf")
- d.addCallback(lambda ignored:
- self.POST(self.public_url + \
- "/foo?t=upload&mutable=true&mutable-type=mdmf",
- file=("mdmf.txt", self.NEWFILE_CONTENTS * 300000)))
- d.addCallback(_got_mdmf_cap)
- d.addCallback(_got_cap, "mdmf.txt")
- d.addCallback(_got_json, "mdmf")
+ d = defer.succeed(None)
+ d.addCallback(_check_upload, "chk", "URI:CHK")
+ d.addCallback(_check_upload, "sdmf", "URI:SSK", self._foo_node)
+ d.addCallback(_check_upload, "mdmf", "URI:MDMF")
+ d.addCallback(_check_upload, "MDMF", "URI:MDMF")
return d
- def test_POST_upload_bad_mutable_type(self):
- return self.shouldHTTPError("test bad mutable type",
- 400, "Bad Request", "Unknown type: foo",
+ def test_POST_upload_bad_format(self):
+ return self.shouldHTTPError("POST_upload_bad_format",
+ 400, "Bad Request", "Unknown format: foo",
self.POST, self.public_url + \
- "/foo?t=upload&mutable=true&mutable-type=foo",
+ "/foo?t=upload&format=foo",
file=("foo.txt", self.NEWFILE_CONTENTS * 300000))
def test_POST_upload_mutable(self):
return d
def test_POST_mkdir_mdmf(self):
- d = self.POST(self.public_url + "/foo?t=mkdir&name=newdir&mutable-type=mdmf")
+ d = self.POST(self.public_url + "/foo?t=mkdir&name=newdir&format=mdmf")
d.addCallback(lambda res: self._foo_node.get(u"newdir"))
d.addCallback(lambda node:
self.failUnlessEqual(node._node.get_version(), MDMF_VERSION))
return d
def test_POST_mkdir_sdmf(self):
- d = self.POST(self.public_url + "/foo?t=mkdir&name=newdir&mutable-type=sdmf")
+ d = self.POST(self.public_url + "/foo?t=mkdir&name=newdir&format=sdmf")
d.addCallback(lambda res: self._foo_node.get(u"newdir"))
d.addCallback(lambda node:
self.failUnlessEqual(node._node.get_version(), SDMF_VERSION))
return d
- def test_POST_mkdir_bad_mutable_type(self):
- return self.shouldHTTPError("test bad mutable type",
- 400, "Bad Request", "Unknown type: foo",
- self.POST, self.public_url + \
- "/foo?t=mkdir&name=newdir&mutable-type=foo")
+ def test_POST_mkdir_bad_format(self):
+ return self.shouldHTTPError("POST_mkdir_bad_format",
+ 400, "Bad Request", "Unknown format: foo",
+ self.POST, self.public_url +
+ "/foo?t=mkdir&name=newdir&format=foo")
def test_POST_mkdir_initial_children(self):
(newkids, caps) = self._create_initial_children()
def test_POST_mkdir_initial_children_mdmf(self):
(newkids, caps) = self._create_initial_children()
d = self.POST2(self.public_url +
- "/foo?t=mkdir-with-children&name=newdir&mutable-type=mdmf",
+ "/foo?t=mkdir-with-children&name=newdir&format=mdmf",
simplejson.dumps(newkids))
d.addCallback(lambda res:
self.failUnlessNodeHasChild(self._foo_node, u"newdir"))
def test_POST_mkdir_initial_children_sdmf(self):
(newkids, caps) = self._create_initial_children()
d = self.POST2(self.public_url +
- "/foo?t=mkdir-with-children&name=newdir&mutable-type=sdmf",
+ "/foo?t=mkdir-with-children&name=newdir&format=sdmf",
simplejson.dumps(newkids))
d.addCallback(lambda res:
self.failUnlessNodeHasChild(self._foo_node, u"newdir"))
caps['filecap1'])
return d
- def test_POST_mkdir_initial_children_bad_mutable_type(self):
+ def test_POST_mkdir_initial_children_bad_format(self):
(newkids, caps) = self._create_initial_children()
- return self.shouldHTTPError("test bad mutable type",
- 400, "Bad Request", "Unknown type: foo",
+ return self.shouldHTTPError("POST_mkdir_initial_children_bad_format",
+ 400, "Bad Request", "Unknown format: foo",
self.POST, self.public_url + \
- "/foo?t=mkdir-with-children&name=newdir&mutable-type=foo",
+ "/foo?t=mkdir-with-children&name=newdir&format=foo",
simplejson.dumps(newkids))
def test_POST_mkdir_immutable(self):
def test_POST_mkdir_immutable_bad(self):
(newkids, caps) = self._create_initial_children()
- d = self.shouldFail2(error.Error, "test_POST_mkdir_immutable_bad",
+ d = self.shouldFail2(error.Error, "POST_mkdir_immutable_bad",
"400 Bad Request",
"needed to be immutable but was not",
self.POST2,
return d
def test_POST_mkdir_no_parentdir_noredirect_mdmf(self):
- d = self.POST("/uri?t=mkdir&mutable-type=mdmf")
+ d = self.POST("/uri?t=mkdir&format=mdmf")
def _after_mkdir(res):
u = uri.from_string(res)
# Check that this is an MDMF writecap
return d
def test_POST_mkdir_no_parentdir_noredirect_sdmf(self):
- d = self.POST("/uri?t=mkdir&mutable-type=sdmf")
+ d = self.POST("/uri?t=mkdir&format=sdmf")
def _after_mkdir(res):
u = uri.from_string(res)
self.failUnlessIsInstance(u, uri.DirectoryURI)
d.addCallback(_after_mkdir)
return d
- def test_POST_mkdir_no_parentdir_noredirect_bad_mutable_type(self):
- return self.shouldHTTPError("test bad mutable type",
- 400, "Bad Request", "Unknown type: foo",
- self.POST, self.public_url + \
- "/uri?t=mkdir&mutable-type=foo")
+ def test_POST_mkdir_no_parentdir_noredirect_bad_format(self):
+ return self.shouldHTTPError("POST_mkdir_no_parentdir_noredirect_bad_format",
+ 400, "Bad Request", "Unknown format: foo",
+ self.POST, self.public_url +
+ "/uri?t=mkdir&format=foo")
def test_POST_mkdir_no_parentdir_noredirect2(self):
# make sure form-based arguments (as on the welcome page) still work
# the regular /uri?t=mkdir operation is specified to ignore its body.
# Only t=mkdir-with-children pays attention to it.
(newkids, caps) = self._create_initial_children()
- d = self.shouldHTTPError("POST t=mkdir unexpected children",
+ d = self.shouldHTTPError("POST_mkdir_no_parentdir_unexpected_children",
400, "Bad Request",
"t=mkdir does not accept children=, "
"try t=mkdir-with-children instead",
return d
def test_POST_noparent_bad(self):
- d = self.shouldHTTPError("POST /uri?t=bogus", 400, "Bad Request",
+ d = self.shouldHTTPError("POST_noparent_bad",
+ 400, "Bad Request",
"/uri accepts only PUT, PUT?t=mkdir, "
"POST?t=upload, and POST?t=mkdir",
self.POST, "/uri?t=bogus")
return d
def test_POST_bad_t(self):
- d = self.shouldFail2(error.Error, "POST_bad_t", "400 Bad Request",
+ d = self.shouldFail2(error.Error, "POST_bad_t",
+ "400 Bad Request",
"POST to a directory with bad t=BOGUS",
self.POST, self.public_url + "/foo", t="BOGUS")
return d
def test_PUT_DIRURL_bad_t(self):
d = self.shouldFail2(error.Error, "test_PUT_DIRURL_bad_t",
- "400 Bad Request", "PUT to a directory",
- self.PUT, self.public_url + "/foo?t=BOGUS", "")
+ "400 Bad Request", "PUT to a directory",
+ self.PUT, self.public_url + "/foo?t=BOGUS", "")
d.addCallback(lambda res:
self.failUnlessRWChildURIIs(self.public_root,
u"foo",
def test_PUT_NEWFILEURL_mdmf(self):
new_contents = self.NEWFILE_CONTENTS * 300000
d = self.PUT(self.public_url + \
- "/foo/mdmf.txt?mutable=true&mutable-type=mdmf",
+ "/foo/mdmf.txt?format=mdmf",
new_contents)
d.addCallback(lambda ignored:
self.GET(self.public_url + "/foo/mdmf.txt?t=json"))
def _got_json(json):
data = simplejson.loads(json)
data = data[1]
- self.failUnlessIn("mutable-type", data)
- self.failUnlessEqual(data['mutable-type'], "mdmf")
+ self.failUnlessIn("format", data)
+ self.failUnlessEqual(data["format"], "MDMF")
self.failUnless(data['rw_uri'].startswith("URI:MDMF"))
self.failUnless(data['ro_uri'].startswith("URI:MDMF"))
d.addCallback(_got_json)
def test_PUT_NEWFILEURL_sdmf(self):
new_contents = self.NEWFILE_CONTENTS * 300000
d = self.PUT(self.public_url + \
- "/foo/sdmf.txt?mutable=true&mutable-type=sdmf",
+ "/foo/sdmf.txt?format=sdmf",
new_contents)
d.addCallback(lambda ignored:
self.GET(self.public_url + "/foo/sdmf.txt?t=json"))
def _got_json(json):
data = simplejson.loads(json)
data = data[1]
- self.failUnlessIn("mutable-type", data)
- self.failUnlessEqual(data['mutable-type'], "sdmf")
+ self.failUnlessIn("format", data)
+ self.failUnlessEqual(data["format"], "SDMF")
d.addCallback(_got_json)
return d
- def test_PUT_NEWFILEURL_bad_mutable_type(self):
+ def test_PUT_NEWFILEURL_bad_format(self):
new_contents = self.NEWFILE_CONTENTS * 300000
- return self.shouldHTTPError("test bad mutable type",
- 400, "Bad Request", "Unknown type: foo",
+ return self.shouldHTTPError("PUT_NEWFILEURL_bad_format",
+ 400, "Bad Request", "Unknown format: foo",
self.PUT, self.public_url + \
- "/foo/foo.txt?mutable=true&mutable-type=foo",
+ "/foo/foo.txt?format=foo",
new_contents)
def test_PUT_NEWFILEURL_uri_replace(self):
def test_PUT_NEWFILEURL_uri_no_replace(self):
contents, n, new_uri = self.makefile(8)
d = self.PUT(self.public_url + "/foo/bar.txt?t=uri&replace=false", new_uri)
- d.addBoth(self.shouldFail, error.Error, "PUT_NEWFILEURL_uri_no_replace",
+ 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_PUT_mkdir_mdmf(self):
- d = self.PUT("/uri?t=mkdir&mutable-type=mdmf", "")
+ d = self.PUT("/uri?t=mkdir&format=mdmf", "")
def _got(res):
u = uri.from_string(res)
# Check that this is an MDMF writecap
return d
def test_PUT_mkdir_sdmf(self):
- d = self.PUT("/uri?t=mkdir&mutable-type=sdmf", "")
+ d = self.PUT("/uri?t=mkdir&format=sdmf", "")
def _got(res):
u = uri.from_string(res)
self.failUnlessIsInstance(u, uri.DirectoryURI)
d.addCallback(_got)
return d
- def test_PUT_mkdir_bad_mutable_type(self):
- return self.shouldHTTPError("bad mutable type",
- 400, "Bad Request", "Unknown type: foo",
- self.PUT, "/uri?t=mkdir&mutable-type=foo",
+ def test_PUT_mkdir_bad_format(self):
+ return self.shouldHTTPError("PUT_mkdir_bad_format",
+ 400, "Bad Request", "Unknown format: foo",
+ self.PUT, "/uri?t=mkdir&format=foo",
"")
def test_POST_check(self):
d.addCallback(_then)
# Negative offsets should cause an error.
d.addCallback(lambda ignored:
- self.shouldHTTPError("test mutable invalid offset negative",
+ self.shouldHTTPError("PUT_update_at_invalid_offset",
400, "Bad Request",
"Invalid offset",
self.PUT,
self.filecap = filecap
d.addCallback(_then)
d.addCallback(lambda ignored:
- self.shouldHTTPError("test immutable update",
+ self.shouldHTTPError("PUT_update_at_offset_immutable",
400, "Bad Request",
"immutable",
self.PUT,
def test_bad_method(self):
url = self.webish_url + self.public_url + "/foo/bar.txt"
- d = self.shouldHTTPError("test_bad_method",
+ d = self.shouldHTTPError("bad_method",
501, "Not Implemented",
"I don't know how to treat a BOGUS request.",
client.getPage, url, method="BOGUS")
def test_short_url(self):
url = self.webish_url + "/uri"
- d = self.shouldHTTPError("test_short_url", 501, "Not Implemented",
+ d = self.shouldHTTPError("short_url", 501, "Not Implemented",
"I don't know how to treat a DELETE request.",
client.getPage, url, method="DELETE")
return d
def test_ophandle_bad(self):
url = self.webish_url + "/operations/bogus?t=status"
- d = self.shouldHTTPError("test_ophandle_bad", 404, "404 Not Found",
+ d = self.shouldHTTPError("ophandle_bad", 404, "404 Not Found",
"unknown/expired handle 'bogus'",
client.getPage, url)
return d
return d
d.addCallback(_check1)
d.addCallback(lambda ignored:
- self.shouldHTTPError("test_ophandle_cancel",
+ self.shouldHTTPError("ophandle_cancel",
404, "404 Not Found",
"unknown/expired handle '128'",
self.GET,
d.addCallback(lambda ign:
self.clock.advance(2.0))
d.addCallback(lambda ignored:
- self.shouldHTTPError("test_ophandle_retainfor",
+ self.shouldHTTPError("ophandle_retainfor",
404, "404 Not Found",
"unknown/expired handle '129'",
self.GET,
self.GET("/operations/130?t=status&output=JSON&release-after-complete=true"))
# the release-after-complete=true will cause the handle to be expired
d.addCallback(lambda ignored:
- self.shouldHTTPError("test_ophandle_release_after_complete",
+ self.shouldHTTPError("ophandle_release_after_complete",
404, "404 Not Found",
"unknown/expired handle '130'",
self.GET,
d.addCallback(lambda ign:
self.clock.advance(96*60*60))
d.addCallback(lambda ign:
- self.shouldHTTPError("test_uncollected_ophandle_expired_after_100_hours",
+ self.shouldHTTPError("uncollected_ophandle_expired_after_100_hours",
404, "404 Not Found",
"unknown/expired handle '132'",
self.GET,
d.addCallback(lambda ign:
self.clock.advance(24*60*60))
d.addCallback(lambda ign:
- self.shouldHTTPError("test_collected_ophandle_expired_after_1000_minutes",
+ self.shouldHTTPError("collected_ophandle_expired_after_1_day",
404, "404 Not Found",
"unknown/expired handle '134'",
self.GET,
for shnum, serverid, fn in shares:
sf = get_share_file(fn)
num_leases = len(list(sf.get_leases()))
- lease_counts.append( (fn, num_leases) )
+ lease_counts.append( (fn, num_leases) )
return lease_counts
def _assert_leasecount(self, lease_counts, expected):
return d
+ def test_blacklist(self):
+ # download from a blacklisted URI, get an error
+ self.basedir = "web/Grid/blacklist"
+ self.set_up_grid()
+ c0 = self.g.clients[0]
+ c0_basedir = c0.basedir
+ fn = os.path.join(c0_basedir, "access.blacklist")
+ self.uris = {}
+ DATA = "off-limits " * 50
+
+ d = c0.upload(upload.Data(DATA, convergence=""))
+ def _stash_uri_and_create_dir(ur):
+ self.uri = ur.uri
+ self.url = "uri/"+self.uri
+ u = uri.from_string_filenode(self.uri)
+ self.si = u.get_storage_index()
+ childnode = c0.create_node_from_uri(self.uri, None)
+ return c0.create_dirnode({u"blacklisted.txt": (childnode,{}) })
+ d.addCallback(_stash_uri_and_create_dir)
+ def _stash_dir(node):
+ self.dir_node = node
+ self.dir_uri = node.get_uri()
+ self.dir_url = "uri/"+self.dir_uri
+ d.addCallback(_stash_dir)
+ d.addCallback(lambda ign: self.GET(self.dir_url, followRedirect=True))
+ def _check_dir_html(body):
+ self.failUnlessIn("<html>", body)
+ self.failUnlessIn("blacklisted.txt</a>", body)
+ d.addCallback(_check_dir_html)
+ d.addCallback(lambda ign: self.GET(self.url))
+ d.addCallback(lambda body: self.failUnlessEqual(DATA, body))
+
+ def _blacklist(ign):
+ f = open(fn, "w")
+ f.write(" # this is a comment\n")
+ f.write(" \n")
+ f.write("\n") # also exercise blank lines
+ f.write("%s %s\n" % (base32.b2a(self.si), "off-limits to you"))
+ f.close()
+ # clients should be checking the blacklist each time, so we don't
+ # need to restart the client
+ d.addCallback(_blacklist)
+ d.addCallback(lambda ign: self.shouldHTTPError("get_from_blacklisted_uri",
+ 403, "Forbidden",
+ "Access Prohibited: off-limits",
+ self.GET, self.url))
+
+ # We should still be able to list the parent directory, in HTML...
+ d.addCallback(lambda ign: self.GET(self.dir_url, followRedirect=True))
+ def _check_dir_html2(body):
+ self.failUnlessIn("<html>", body)
+ self.failUnlessIn("blacklisted.txt</strike>", body)
+ d.addCallback(_check_dir_html2)
+
+ # ... and in JSON (used by CLI).
+ d.addCallback(lambda ign: self.GET(self.dir_url+"?t=json", followRedirect=True))
+ def _check_dir_json(res):
+ data = simplejson.loads(res)
+ self.failUnless(isinstance(data, list), data)
+ self.failUnlessEqual(data[0], "dirnode")
+ self.failUnless(isinstance(data[1], dict), data)
+ self.failUnlessIn("children", data[1])
+ self.failUnlessIn("blacklisted.txt", data[1]["children"])
+ childdata = data[1]["children"]["blacklisted.txt"]
+ self.failUnless(isinstance(childdata, list), data)
+ self.failUnlessEqual(childdata[0], "filenode")
+ self.failUnless(isinstance(childdata[1], dict), data)
+ d.addCallback(_check_dir_json)
+
+ def _unblacklist(ign):
+ open(fn, "w").close()
+ # the Blacklist object watches mtime to tell when the file has
+ # changed, but on windows this test will run faster than the
+ # filesystem's mtime resolution. So we edit Blacklist.last_mtime
+ # to force a reload.
+ self.g.clients[0].blacklist.last_mtime -= 2.0
+ d.addCallback(_unblacklist)
+
+ # now a read should work
+ d.addCallback(lambda ign: self.GET(self.url))
+ d.addCallback(lambda body: self.failUnlessEqual(DATA, body))
+
+ # read again to exercise the blacklist-is-unchanged logic
+ d.addCallback(lambda ign: self.GET(self.url))
+ d.addCallback(lambda body: self.failUnlessEqual(DATA, body))
+
+ # now add a blacklisted directory, and make sure files under it are
+ # refused too
+ def _add_dir(ign):
+ childnode = c0.create_node_from_uri(self.uri, None)
+ return c0.create_dirnode({u"child": (childnode,{}) })
+ d.addCallback(_add_dir)
+ def _get_dircap(dn):
+ self.dir_si_b32 = base32.b2a(dn.get_storage_index())
+ self.dir_url_base = "uri/"+dn.get_write_uri()
+ self.dir_url_json1 = "uri/"+dn.get_write_uri()+"?t=json"
+ self.dir_url_json2 = "uri/"+dn.get_write_uri()+"/?t=json"
+ self.dir_url_json_ro = "uri/"+dn.get_readonly_uri()+"/?t=json"
+ self.child_url = "uri/"+dn.get_readonly_uri()+"/child"
+ d.addCallback(_get_dircap)
+ d.addCallback(lambda ign: self.GET(self.dir_url_base, followRedirect=True))
+ d.addCallback(lambda body: self.failUnlessIn("<html>", body))
+ d.addCallback(lambda ign: self.GET(self.dir_url_json1))
+ d.addCallback(lambda res: simplejson.loads(res)) # just check it decodes
+ d.addCallback(lambda ign: self.GET(self.dir_url_json2))
+ d.addCallback(lambda res: simplejson.loads(res)) # just check it decodes
+ d.addCallback(lambda ign: self.GET(self.dir_url_json_ro))
+ d.addCallback(lambda res: simplejson.loads(res)) # just check it decodes
+ d.addCallback(lambda ign: self.GET(self.child_url))
+ d.addCallback(lambda body: self.failUnlessEqual(DATA, body))
+
+ def _block_dir(ign):
+ f = open(fn, "w")
+ f.write("%s %s\n" % (self.dir_si_b32, "dir-off-limits to you"))
+ f.close()
+ self.g.clients[0].blacklist.last_mtime -= 2.0
+ d.addCallback(_block_dir)
+ d.addCallback(lambda ign: self.shouldHTTPError("get_from_blacklisted_dir base",
+ 403, "Forbidden",
+ "Access Prohibited: dir-off-limits",
+ self.GET, self.dir_url_base))
+ d.addCallback(lambda ign: self.shouldHTTPError("get_from_blacklisted_dir json1",
+ 403, "Forbidden",
+ "Access Prohibited: dir-off-limits",
+ self.GET, self.dir_url_json1))
+ d.addCallback(lambda ign: self.shouldHTTPError("get_from_blacklisted_dir json2",
+ 403, "Forbidden",
+ "Access Prohibited: dir-off-limits",
+ self.GET, self.dir_url_json2))
+ d.addCallback(lambda ign: self.shouldHTTPError("get_from_blacklisted_dir json_ro",
+ 403, "Forbidden",
+ "Access Prohibited: dir-off-limits",
+ self.GET, self.dir_url_json_ro))
+ d.addCallback(lambda ign: self.shouldHTTPError("get_from_blacklisted_dir child",
+ 403, "Forbidden",
+ "Access Prohibited: dir-off-limits",
+ self.GET, self.child_url))
+ return d
+
+
class CompletelyUnhandledError(Exception):
pass
class ErrorBoom(rend.Page):
projects / tahoe-lafs / tahoe-lafs.git / blobdiff