tahoe-lafs/tahoe-lafs
tahoe-lafs
/
tahoe-lafs
2
0
Fork 0
Code Issues Pull Requests 23 Packages Projects Releases 1 Wiki Activity

#509: test_system.py: add test for streamed-manifest

This commit is contained in:
Brian Warner 2009-01-23 16:32:47 -07:00
parent 9ade197c3b
commit 5a0e3f3a40
1 changed files with 45 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
src/allmydata/test/test_system.py
View File

@ -2108,6 +2108,7 @@ class DeepCheckWebGood(DeepCheckBase, unittest.TestCase):
d.addCallback(self.set_up_tree) d.addCallback(self.set_up_tree)
d.addCallback(self.do_stats) d.addCallback(self.do_stats)
d.addCallback(self.do_web_stream_manifest) d.addCallback(self.do_web_stream_manifest)
d.addCallback(self.do_web_stream_check)
d.addCallback(self.do_test_check_good) d.addCallback(self.do_test_check_good)
d.addCallback(self.do_test_web_good) d.addCallback(self.do_test_web_good)
d.addCallback(self.do_test_cli_good) d.addCallback(self.do_test_cli_good)
@ -2146,8 +2147,12 @@ class DeepCheckWebGood(DeepCheckBase, unittest.TestCase):
def do_web_stream_manifest(self, ignored): def do_web_stream_manifest(self, ignored):
d = self.web(self.root, method="POST", t="stream-manifest") d = self.web(self.root, method="POST", t="stream-manifest")
def _check((res,url)): d.addCallback(lambda (output,url):
units = list(self.parse_streamed_json(res)) self._check_streamed_manifest(output))
return d
def _check_streamed_manifest(self, output):
units = list(self.parse_streamed_json(output))
files = [u for u in units if u["type"] in ("file", "directory")] files = [u for u in units if u["type"] in ("file", "directory")]
assert units[-1]["type"] == "stats" assert units[-1]["type"] == "stats"
stats = units[-1]["stats"] stats = units[-1]["stats"]
@ -2180,6 +2185,16 @@ class DeepCheckWebGood(DeepCheckBase, unittest.TestCase):
self.failIfEqual(large["cap"], large["verifycap"]) self.failIfEqual(large["cap"], large["verifycap"])
self.failUnlessEqual(large["verifycap"], large["repaircap"]) self.failUnlessEqual(large["verifycap"], large["repaircap"])
self.check_stats_good(stats) self.check_stats_good(stats)
def do_web_stream_check(self, ignored):
return
d = self.web(self.root, t="stream-deep-check")
def _check(res):
units = list(self.parse_streamed_json(res))
files = [u for u in units if u["type"] in ("file", "directory")]
assert units[-1]["type"] == "stats"
stats = units[-1]["stats"]
# ...
d.addCallback(_check) d.addCallback(_check)
return d return d