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_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, output)
  92         self.failUnless("key: aaaqeayeaudaocajbifqydiob4" in output, output)
  93         self.failUnless("UEB hash: nf3nimquen7aeqm36ekgxomalstenpkvsdmf6fplj7swdatbv5oa" in output, output)
  94         self.failUnless("size: 1234" in output, output)
  95         self.failUnless("k/N: 25/100" in output, output)
  96         self.failUnless("storage index: hdis5iaveku6lnlaiccydyid7q" in output, output)
  97
  98         output = self._dump_cap("--client-secret", "5s33nk3qpvnj2fw3z4mnm2y6fa",
  99                                 u.to_string())
 100         self.failUnless("client renewal secret: znxmki5zdibb5qlt46xbdvk2t55j7hibejq3i5ijyurkr6m6jkhq" in output, output)
 101
 102         output = self._dump_cap(u.get_verifier().to_string())
 103         self.failIf("key: " in output, output)
 104         self.failUnless("UEB hash: nf3nimquen7aeqm36ekgxomalstenpkvsdmf6fplj7swdatbv5oa" in output, output)
 105         self.failUnless("size: 1234" in output, output)
 106         self.failUnless("k/N: 25/100" in output, output)
 107         self.failUnless("storage index: hdis5iaveku6lnlaiccydyid7q" in output, 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, output)
 112         self.failUnless("key: aaaqeayeaudaocajbifqydiob4" in output, output)
 113         self.failUnless("UEB hash: nf3nimquen7aeqm36ekgxomalstenpkvsdmf6fplj7swdatbv5oa" in output, output)
 114         self.failUnless("size: 1234" in output, output)
 115         self.failUnless("k/N: 25/100" in output, output)
 116         self.failUnless("storage index: hdis5iaveku6lnlaiccydyid7q" in output, 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, output)
 122         self.failUnless("data: this is some data" in output, 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, output)
 131         self.failUnless("writekey: aeaqcaibaeaqcaibaeaqcaibae" in output, output)
 132         self.failUnless("readkey: nvgh5vj2ekzzkim5fgtb4gey5y" in output, output)
 133         self.failUnless("storage index: nt4fwemuw7flestsezvo2eveke" in output, output)
 134         self.failUnless("fingerprint: 737p57x6737p57x6737p57x6737p57x6737p57x6737p57x6737a" in output, output)
 135
 136         output = self._dump_cap("--client-secret", "5s33nk3qpvnj2fw3z4mnm2y6fa",
 137                                 u.to_string())
 138         self.failUnless("file renewal secret: arpszxzc2t6kb4okkg7sp765xgkni5z7caavj7lta73vmtymjlxq" in output, output)
 139
 140         fileutil.make_dirs("cli/test_dump_cap/private")
 141         f = open("cli/test_dump_cap/private/secret", "w")
 142         f.write("5s33nk3qpvnj2fw3z4mnm2y6fa\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: arpszxzc2t6kb4okkg7sp765xgkni5z7caavj7lta73vmtymjlxq" in output, 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, output)
 151
 152         output = self._dump_cap("--nodeid", "tqc35esocrvejvg4mablt6aowg6tl43j",
 153                                 u.to_string())
 154         self.failUnless("write_enabler: mgcavriox2wlb5eer26unwy5cw56elh3sjweffckkmivvsxtaknq" in output, output)
 155         self.failIf("file renewal secret:" in output, output)
 156
 157         output = self._dump_cap("--nodeid", "tqc35esocrvejvg4mablt6aowg6tl43j",
 158                                 "--client-secret", "5s33nk3qpvnj2fw3z4mnm2y6fa",
 159                                 u.to_string())
 160         self.failUnless("write_enabler: mgcavriox2wlb5eer26unwy5cw56elh3sjweffckkmivvsxtaknq" in output, output)
 161         self.failUnless("file renewal secret: arpszxzc2t6kb4okkg7sp765xgkni5z7caavj7lta73vmtymjlxq" in output, output)
 162         self.failUnless("lease renewal secret: 7pjtaumrb7znzkkbvekkmuwpqfjyfyamznfz4bwwvmh4nw33lorq" in output, output)
 163
 164         u = u.get_readonly()
 165         output = self._dump_cap(u.to_string())
 166         self.failUnless("SSK Read-only URI:" in output, output)
 167         self.failUnless("readkey: nvgh5vj2ekzzkim5fgtb4gey5y" in output, output)
 168         self.failUnless("storage index: nt4fwemuw7flestsezvo2eveke" in output, output)
 169         self.failUnless("fingerprint: 737p57x6737p57x6737p57x6737p57x6737p57x6737p57x6737a" in output, output)
 170
 171         u = u.get_verifier()
 172         output = self._dump_cap(u.to_string())
 173         self.failUnless("SSK Verifier URI:" in output, output)
 174         self.failUnless("storage index: nt4fwemuw7flestsezvo2eveke" in output, output)
 175         self.failUnless("fingerprint: 737p57x6737p57x6737p57x6737p57x6737p57x6737p57x6737a" in output, 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, output)
 185         self.failUnless("writekey: aeaqcaibaeaqcaibaeaqcaibae" in output,
 186                         output)
 187         self.failUnless("readkey: nvgh5vj2ekzzkim5fgtb4gey5y" in output, output)
 188         self.failUnless("storage index: nt4fwemuw7flestsezvo2eveke" in output,
 189                         output)
 190         self.failUnless("fingerprint: 737p57x6737p57x6737p57x6737p57x6737p57x6737p57x6737a" in output, output)
 191
 192         output = self._dump_cap("--client-secret", "5s33nk3qpvnj2fw3z4mnm2y6fa",
 193                                 u.to_string())
 194         self.failUnless("file renewal secret: arpszxzc2t6kb4okkg7sp765xgkni5z7caavj7lta73vmtymjlxq" in output, output)
 195
 196         output = self._dump_cap("--nodeid", "tqc35esocrvejvg4mablt6aowg6tl43j",
 197                                 u.to_string())
 198         self.failUnless("write_enabler: mgcavriox2wlb5eer26unwy5cw56elh3sjweffckkmivvsxtaknq" in output, output)
 199         self.failIf("file renewal secret:" in output, output)
 200
 201         output = self._dump_cap("--nodeid", "tqc35esocrvejvg4mablt6aowg6tl43j",
 202                                 "--client-secret", "5s33nk3qpvnj2fw3z4mnm2y6fa",
 203                                 u.to_string())
 204         self.failUnless("write_enabler: mgcavriox2wlb5eer26unwy5cw56elh3sjweffckkmivvsxtaknq" in output, output)
 205         self.failUnless("file renewal secret: arpszxzc2t6kb4okkg7sp765xgkni5z7caavj7lta73vmtymjlxq" in output, output)
 206         self.failUnless("lease renewal secret: 7pjtaumrb7znzkkbvekkmuwpqfjyfyamznfz4bwwvmh4nw33lorq" in output, output)
 207
 208         u = u.get_readonly()
 209         output = self._dump_cap(u.to_string())
 210         self.failUnless("Directory Read-only URI:" in output, output)
 211         self.failUnless("readkey: nvgh5vj2ekzzkim5fgtb4gey5y" in output, output)
 212         self.failUnless("storage index: nt4fwemuw7flestsezvo2eveke" in output, output)
 213         self.failUnless("fingerprint: 737p57x6737p57x6737p57x6737p57x6737p57x6737p57x6737a" in output, output)
 214
 215         u = u.get_verifier()
 216         output = self._dump_cap(u.to_string())
 217         self.failUnless("Directory Verifier URI:" in output, output)
 218         self.failUnless("storage index: nt4fwemuw7flestsezvo2eveke" in output, output)
 219         self.failUnless("fingerprint: 737p57x6737p57x6737p57x6737p57x6737p57x6737p57x6737a" in output, output)
 220