src/allmydata/test/test_cli.py

   1
   2 from twisted.trial import unittest
   3 from cStringIO import StringIO
   4 import urllib
   5
   6 from allmydata.util import fileutil, hashutil
   7 from allmydata import uri
   8
   9 # at least import the CLI scripts, even if we don't have any real tests for
  10 # them yet.
  11 from allmydata.scripts import tahoe_ls, tahoe_get, tahoe_put, tahoe_rm
  12 _hush_pyflakes = [tahoe_ls, tahoe_get, tahoe_put, tahoe_rm]
  13
  14 from allmydata.scripts import cli, debug
  15
  16
  17 class CLI(unittest.TestCase):
  18     def test_options(self):
  19         fileutil.rm_dir("cli/test_options")
  20         fileutil.make_dirs("cli/test_options")
  21         fileutil.make_dirs("cli/test_options/private")
  22         open("cli/test_options/node.url","w").write("http://localhost:8080/\n")
  23         filenode_uri = uri.WriteableSSKFileURI(writekey="\x00"*16,
  24                                                fingerprint="\x00"*32)
  25         private_uri = uri.NewDirectoryURI(filenode_uri).to_string()
  26         open("cli/test_options/private/root_dir.cap", "w").write(private_uri + "\n")
  27         o = cli.ListOptions()
  28         o.parseOptions(["--node-directory", "cli/test_options"])
  29         self.failUnlessEqual(o['node-url'], "http://localhost:8080/")
  30         self.failUnlessEqual(o['dir-cap'], private_uri)
  31         self.failUnlessEqual(o['vdrive_pathname'], "")
  32
  33         o = cli.ListOptions()
  34         o.parseOptions(["--node-directory", "cli/test_options",
  35                         "--node-url", "http://example.org:8111/"])
  36         self.failUnlessEqual(o['node-url'], "http://example.org:8111/")
  37         self.failUnlessEqual(o['dir-cap'], private_uri)
  38         self.failUnlessEqual(o['vdrive_pathname'], "")
  39
  40         o = cli.ListOptions()
  41         o.parseOptions(["--node-directory", "cli/test_options",
  42                         "--dir-cap", "root"])
  43         self.failUnlessEqual(o['node-url'], "http://localhost:8080/")
  44         self.failUnlessEqual(o['dir-cap'], private_uri)
  45         self.failUnlessEqual(o['vdrive_pathname'], "")
  46
  47         o = cli.ListOptions()
  48         o.parseOptions(["--node-directory", "cli/test_options"])
  49         self.failUnlessEqual(o['node-url'], "http://localhost:8080/")
  50         self.failUnlessEqual(o['vdrive_pathname'], "")
  51
  52         o = cli.ListOptions()
  53         other_filenode_uri = uri.WriteableSSKFileURI(writekey="\x11"*16,
  54                                                      fingerprint="\x11"*32)
  55         other_uri = uri.NewDirectoryURI(other_filenode_uri).to_string()
  56         o.parseOptions(["--node-directory", "cli/test_options",
  57                         "--dir-cap", other_uri])
  58         self.failUnlessEqual(o['node-url'], "http://localhost:8080/")
  59         self.failUnlessEqual(o['dir-cap'], other_uri)
  60         self.failUnlessEqual(o['vdrive_pathname'], "")
  61
  62         o = cli.ListOptions()
  63         o.parseOptions(["--node-directory", "cli/test_options",
  64                         "--dir-cap", other_uri, "subdir"])
  65         self.failUnlessEqual(o['node-url'], "http://localhost:8080/")
  66         self.failUnlessEqual(o['dir-cap'], other_uri)
  67         self.failUnlessEqual(o['vdrive_pathname'], "subdir")
  68
  69     def _dump_cap(self, *args):
  70         out,err = StringIO(), StringIO()
  71         config = debug.DumpCapOptions()
  72         config.parseOptions(args)
  73         debug.dump_cap(config, out, err)
  74         self.failIf(err.getvalue())
  75         output = out.getvalue()
  76         return output
  77
  78     def test_dump_cap_chk(self):
  79         key = "\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f"
  80         storage_index = hashutil.storage_index_chk_hash(key)
  81         uri_extension_hash = hashutil.uri_extension_hash("stuff")
  82         needed_shares = 25
  83         total_shares = 100
  84         size = 1234
  85         u = uri.CHKFileURI(key=key,
  86                            uri_extension_hash=uri_extension_hash,
  87                            needed_shares=needed_shares,
  88                            total_shares=total_shares,
  89                            size=size)
  90         output = self._dump_cap(u.to_string())
  91         self.failUnless("CHK File:" in output)
  92         self.failUnless("key: yyyoryarywdyqnyjbefoadeqbh" in output)
  93         self.failUnless("UEB hash: hd7rwri6djiapo6itg5hcxa7ze5im7z9qwcdu8oka6qinahsbiuo" in output)
  94         self.failUnless("size: 1234" in output)
  95         self.failUnless("k/N: 25/100" in output)
  96         self.failUnless("storage index: p3w849k9whqhw6b9fkf4xjs5xc" in output)
  97
  98         output = self._dump_cap("--client-secret", "p3w849k9whqhw6b9fkf4xjs5xc",
  99                                 u.to_string())
 100         self.failUnless("client renewal secret: pu3oy5fu4irjsudwhn6c71g87anrxi1kokt4hmxz7qh5p1895zpy" in output)
 101
 102         output = self._dump_cap(u.get_verifier().to_string())
 103         self.failIf("key: " in output)
 104         self.failUnless("UEB hash: hd7rwri6djiapo6itg5hcxa7ze5im7z9qwcdu8oka6qinahsbiuo" in output)
 105         self.failUnless("size: 1234" in output)
 106         self.failUnless("k/N: 25/100" in output)
 107         self.failUnless("storage index: p3w849k9whqhw6b9fkf4xjs5xc" in output)
 108
 109         prefixed_u = "http://127.0.0.1/uri/%s" % urllib.quote(u.to_string())
 110         output = self._dump_cap(prefixed_u)
 111         self.failUnless("CHK File:" in output)
 112         self.failUnless("key: yyyoryarywdyqnyjbefoadeqbh" in output)
 113         self.failUnless("UEB hash: hd7rwri6djiapo6itg5hcxa7ze5im7z9qwcdu8oka6qinahsbiuo" in output)
 114         self.failUnless("size: 1234" in output)
 115         self.failUnless("k/N: 25/100" in output)
 116         self.failUnless("storage index: p3w849k9whqhw6b9fkf4xjs5xc" in output)
 117
 118     def test_dump_cap_lit(self):
 119         u = uri.LiteralFileURI("this is some data")
 120         output = self._dump_cap(u.to_string())
 121         self.failUnless("Literal File URI:" in output)
 122         self.failUnless("data: this is some data" in output)
 123
 124     def test_dump_cap_ssk(self):
 125         writekey = "\x01" * 16
 126         fingerprint = "\xfe" * 32
 127         u = uri.WriteableSSKFileURI(writekey, fingerprint)
 128
 129         output = self._dump_cap(u.to_string())
 130         self.failUnless("SSK Writeable URI:" in output)
 131         self.failUnless("writekey: yryonyebyryonyebyryonyebyr" in output)
 132         self.failUnless("readkey: zhgqsyrkuywo3rha41b1d7xrar" in output)
 133         self.failUnless("storage index: toz9zpxrzjzwoxtuq6xr3ygdze" in output)
 134         self.failUnless("fingerprint: 959x79z6959x79z6959x79z6959x79z6959x79z6959x79z6959y" in output)
 135
 136         output = self._dump_cap("--client-secret", "p3w849k9whqhw6b9fkf4xjs5xc",
 137                                 u.to_string())
 138         self.failUnless("file renewal secret: xy9p89q9pkitqn4ycwu5tpt9yia7s9izsqudnb4q5jdc3rawgcny" in output)
 139
 140         fileutil.make_dirs("cli/test_dump_cap/private")
 141         f = open("cli/test_dump_cap/private/secret", "w")
 142         f.write("p3w849k9whqhw6b9fkf4xjs5xc\n")
 143         f.close()
 144         output = self._dump_cap("--client-dir", "cli/test_dump_cap",
 145                                 u.to_string())
 146         self.failUnless("file renewal secret: xy9p89q9pkitqn4ycwu5tpt9yia7s9izsqudnb4q5jdc3rawgcny" in output)
 147
 148         output = self._dump_cap("--client-dir", "cli/test_dump_cap_BOGUS",
 149                                 u.to_string())
 150         self.failIf("file renewal secret:" in output)
 151
 152         output = self._dump_cap("--nodeid", "tqc35esocrvejvg4mablt6aowg6tl43j",
 153                                 u.to_string())
 154         self.failUnless("write_enabler: rqk9q6w46dim5ybshqk9kotkyhqcdqmp1z6498xniuz5kkjs1w7o" in output)
 155         self.failIf("file renewal secret:" in output)
 156
 157         output = self._dump_cap("--nodeid", "tqc35esocrvejvg4mablt6aowg6tl43j",
 158                                 "--client-secret", "p3w849k9whqhw6b9fkf4xjs5xc",
 159                                 u.to_string())
 160         self.failUnless("write_enabler: rqk9q6w46dim5ybshqk9kotkyhqcdqmp1z6498xniuz5kkjs1w7o" in output)
 161         self.failUnless("file renewal secret: xy9p89q9pkitqn4ycwu5tpt9yia7s9izsqudnb4q5jdc3rawgcny" in output)
 162         self.failUnless("lease renewal secret: r3fsw67mfji3c9mtsisqdumc1pz3gquzdrh4cpu63h8du4uuedgo" in output)
 163
 164         u = u.get_readonly()
 165         output = self._dump_cap(u.to_string())
 166         self.failUnless("SSK Read-only URI:" in output)
 167         self.failUnless("readkey: zhgqsyrkuywo3rha41b1d7xrar" in output)
 168         self.failUnless("storage index: toz9zpxrzjzwoxtuq6xr3ygdze" in output)
 169         self.failUnless("fingerprint: 959x79z6959x79z6959x79z6959x79z6959x79z6959x79z6959y" in output)
 170
 171         u = u.get_verifier()
 172         output = self._dump_cap(u.to_string())
 173         self.failUnless("SSK Verifier URI:" in output)
 174         self.failUnless("storage index: toz9zpxrzjzwoxtuq6xr3ygdze" in output)
 175         self.failUnless("fingerprint: 959x79z6959x79z6959x79z6959x79z6959x79z6959x79z6959y" in output)
 176
 177     def test_dump_cap_directory(self):
 178         writekey = "\x01" * 16
 179         fingerprint = "\xfe" * 32
 180         u1 = uri.WriteableSSKFileURI(writekey, fingerprint)
 181         u = uri.NewDirectoryURI(u1)
 182
 183         output = self._dump_cap(u.to_string())
 184         self.failUnless("Directory Writeable URI:" in output)
 185         self.failUnless("writekey: yryonyebyryonyebyryonyebyr" in output)
 186         self.failUnless("readkey: zhgqsyrkuywo3rha41b1d7xrar" in output)
 187         self.failUnless("storage index: toz9zpxrzjzwoxtuq6xr3ygdze" in output)
 188         self.failUnless("fingerprint: 959x79z6959x79z6959x79z6959x79z6959x79z6959x79z6959y" in output)
 189
 190         output = self._dump_cap("--client-secret", "p3w849k9whqhw6b9fkf4xjs5xc",
 191                                 u.to_string())
 192         self.failUnless("file renewal secret: xy9p89q9pkitqn4ycwu5tpt9yia7s9izsqudnb4q5jdc3rawgcny" in output)
 193
 194         output = self._dump_cap("--nodeid", "tqc35esocrvejvg4mablt6aowg6tl43j",
 195                                 u.to_string())
 196         self.failUnless("write_enabler: rqk9q6w46dim5ybshqk9kotkyhqcdqmp1z6498xniuz5kkjs1w7o" in output)
 197         self.failIf("file renewal secret:" in output)
 198
 199         output = self._dump_cap("--nodeid", "tqc35esocrvejvg4mablt6aowg6tl43j",
 200                                 "--client-secret", "p3w849k9whqhw6b9fkf4xjs5xc",
 201                                 u.to_string())
 202         self.failUnless("write_enabler: rqk9q6w46dim5ybshqk9kotkyhqcdqmp1z6498xniuz5kkjs1w7o" in output)
 203         self.failUnless("file renewal secret: xy9p89q9pkitqn4ycwu5tpt9yia7s9izsqudnb4q5jdc3rawgcny" in output)
 204         self.failUnless("lease renewal secret: r3fsw67mfji3c9mtsisqdumc1pz3gquzdrh4cpu63h8du4uuedgo" in output)
 205
 206         u = u.get_readonly()
 207         output = self._dump_cap(u.to_string())
 208         self.failUnless("Directory Read-only URI:" in output)
 209         self.failUnless("readkey: zhgqsyrkuywo3rha41b1d7xrar" in output)
 210         self.failUnless("storage index: toz9zpxrzjzwoxtuq6xr3ygdze" in output)
 211         self.failUnless("fingerprint: 959x79z6959x79z6959x79z6959x79z6959x79z6959x79z6959y" in output)
 212
 213         u = u.get_verifier()
 214         output = self._dump_cap(u.to_string())
 215         self.failUnless("Directory Verifier URI:" in output)
 216         self.failUnless("storage index: toz9zpxrzjzwoxtuq6xr3ygdze" in output)
 217         self.failUnless("fingerprint: 959x79z6959x79z6959x79z6959x79z6959x79z6959x79z6959y" in output)
 218