""" Ported to Python 3. """ from io import StringIO import re from six import ensure_text import os.path from urllib.parse import quote as url_quote from twisted.trial import unittest from twisted.internet.testing import ( MemoryReactor, ) from twisted.internet.test.modulehelpers import ( AlternateReactor, ) import allmydata from allmydata.crypto import ed25519 from allmydata.util import fileutil, hashutil, base32 from allmydata import uri from allmydata.immutable import upload from allmydata.dirnode import normalize from allmydata.scripts.common_http import socket_error import allmydata.scripts.common_http # Test that the scripts can be imported. from allmydata.scripts import create_node, debug, \ tahoe_add_alias, tahoe_backup, tahoe_check, tahoe_cp, tahoe_get, tahoe_ls, \ tahoe_manifest, tahoe_mkdir, tahoe_mv, tahoe_put, tahoe_unlink, tahoe_webopen, \ tahoe_run _hush_pyflakes = [create_node, debug, tahoe_add_alias, tahoe_backup, tahoe_check, tahoe_cp, tahoe_get, tahoe_ls, tahoe_manifest, tahoe_mkdir, tahoe_mv, tahoe_put, tahoe_unlink, tahoe_webopen, tahoe_run] from allmydata.scripts import common from allmydata.scripts.common import DEFAULT_ALIAS, get_aliases, get_alias, \ DefaultAliasMarker from allmydata.scripts import cli, debug, runner from allmydata.test.common_util import (ReallyEqualMixin, skip_if_cannot_represent_filename, run_cli) from allmydata.test.no_network import GridTestMixin from allmydata.test.cli.common import CLITestMixin, parse_options from twisted.python import usage from allmydata.util.encodingutil import listdir_unicode, get_io_encoding class CLI(CLITestMixin, unittest.TestCase): def _dump_cap(self, *args): args = [ensure_text(s) for s in args] config = debug.DumpCapOptions() config.stdout,config.stderr = StringIO(), StringIO() config.parseOptions(args) debug.dump_cap(config) self.failIf(config.stderr.getvalue()) output = config.stdout.getvalue() return output def test_dump_cap_chk(self): key = b"\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f" uri_extension_hash = hashutil.uri_extension_hash(b"stuff") needed_shares = 25 total_shares = 100 size = 1234 u = uri.CHKFileURI(key=key, uri_extension_hash=uri_extension_hash, needed_shares=needed_shares, total_shares=total_shares, size=size) output = self._dump_cap(u.to_string()) self.failUnless("CHK File:" in output, output) self.failUnless("key: aaaqeayeaudaocajbifqydiob4" in output, output) self.failUnless("UEB hash: nf3nimquen7aeqm36ekgxomalstenpkvsdmf6fplj7swdatbv5oa" in output, output) self.failUnless("size: 1234" in output, output) self.failUnless("k/N: 25/100" in output, output) self.failUnless("storage index: hdis5iaveku6lnlaiccydyid7q" in output, output) output = self._dump_cap("--client-secret", "5s33nk3qpvnj2fw3z4mnm2y6fa", u.to_string()) self.failUnless("client renewal secret: znxmki5zdibb5qlt46xbdvk2t55j7hibejq3i5ijyurkr6m6jkhq" in output, output) output = self._dump_cap(str(u.get_verify_cap().to_string(), "ascii")) self.failIf("key: " in output, output) self.failUnless("UEB hash: nf3nimquen7aeqm36ekgxomalstenpkvsdmf6fplj7swdatbv5oa" in output, output) self.failUnless("size: 1234" in output, output) self.failUnless("k/N: 25/100" in output, output) self.failUnless("storage index: hdis5iaveku6lnlaiccydyid7q" in output, output) prefixed_u = "http://127.0.0.1/uri/%s" % url_quote(u.to_string()) output = self._dump_cap(prefixed_u) self.failUnless("CHK File:" in output, output) self.failUnless("key: aaaqeayeaudaocajbifqydiob4" in output, output) self.failUnless("UEB hash: nf3nimquen7aeqm36ekgxomalstenpkvsdmf6fplj7swdatbv5oa" in output, output) self.failUnless("size: 1234" in output, output) self.failUnless("k/N: 25/100" in output, output) self.failUnless("storage index: hdis5iaveku6lnlaiccydyid7q" in output, output) def test_dump_cap_lit(self): u = uri.LiteralFileURI(b"this is some data") output = self._dump_cap(u.to_string()) self.failUnless("Literal File URI:" in output, output) self.failUnless("data: 'this is some data'" in output, output) def test_dump_cap_sdmf(self): writekey = b"\x01" * 16 fingerprint = b"\xfe" * 32 u = uri.WriteableSSKFileURI(writekey, fingerprint) output = self._dump_cap(u.to_string()) self.failUnless("SDMF Writeable URI:" in output, output) self.failUnless("writekey: aeaqcaibaeaqcaibaeaqcaibae" in output, output) self.failUnless("readkey: nvgh5vj2ekzzkim5fgtb4gey5y" in output, output) self.failUnless("storage index: nt4fwemuw7flestsezvo2eveke" in output, output) self.failUnless("fingerprint: 737p57x6737p57x6737p57x6737p57x6737p57x6737p57x6737a" in output, output) output = self._dump_cap("--client-secret", "5s33nk3qpvnj2fw3z4mnm2y6fa", u.to_string()) self.failUnless("file renewal secret: arpszxzc2t6kb4okkg7sp765xgkni5z7caavj7lta73vmtymjlxq" in output, output) fileutil.make_dirs("cli/test_dump_cap/private") fileutil.write("cli/test_dump_cap/private/secret", "5s33nk3qpvnj2fw3z4mnm2y6fa\n") output = self._dump_cap("--client-dir", "cli/test_dump_cap", u.to_string()) self.failUnless("file renewal secret: arpszxzc2t6kb4okkg7sp765xgkni5z7caavj7lta73vmtymjlxq" in output, output) output = self._dump_cap("--client-dir", "cli/test_dump_cap_BOGUS", u.to_string()) self.failIf("file renewal secret:" in output, output) output = self._dump_cap("--nodeid", "tqc35esocrvejvg4mablt6aowg6tl43j", u.to_string()) self.failUnless("write_enabler: mgcavriox2wlb5eer26unwy5cw56elh3sjweffckkmivvsxtaknq" in output, output) self.failIf("file renewal secret:" in output, output) output = self._dump_cap("--nodeid", "tqc35esocrvejvg4mablt6aowg6tl43j", "--client-secret", "5s33nk3qpvnj2fw3z4mnm2y6fa", u.to_string()) self.failUnless("write_enabler: mgcavriox2wlb5eer26unwy5cw56elh3sjweffckkmivvsxtaknq" in output, output) self.failUnless("file renewal secret: arpszxzc2t6kb4okkg7sp765xgkni5z7caavj7lta73vmtymjlxq" in output, output) self.failUnless("lease renewal secret: 7pjtaumrb7znzkkbvekkmuwpqfjyfyamznfz4bwwvmh4nw33lorq" in output, output) u = u.get_readonly() output = self._dump_cap(u.to_string()) self.failUnless("SDMF Read-only URI:" in output, output) self.failUnless("readkey: nvgh5vj2ekzzkim5fgtb4gey5y" in output, output) self.failUnless("storage index: nt4fwemuw7flestsezvo2eveke" in output, output) self.failUnless("fingerprint: 737p57x6737p57x6737p57x6737p57x6737p57x6737p57x6737a" in output, output) u = u.get_verify_cap() output = self._dump_cap(u.to_string()) self.failUnless("SDMF Verifier URI:" in output, output) self.failUnless("storage index: nt4fwemuw7flestsezvo2eveke" in output, output) self.failUnless("fingerprint: 737p57x6737p57x6737p57x6737p57x6737p57x6737p57x6737a" in output, output) def test_dump_cap_mdmf(self): writekey = b"\x01" * 16 fingerprint = b"\xfe" * 32 u = uri.WriteableMDMFFileURI(writekey, fingerprint) output = self._dump_cap(u.to_string()) self.failUnless("MDMF Writeable URI:" in output, output) self.failUnless("writekey: aeaqcaibaeaqcaibaeaqcaibae" in output, output) self.failUnless("readkey: nvgh5vj2ekzzkim5fgtb4gey5y" in output, output) self.failUnless("storage index: nt4fwemuw7flestsezvo2eveke" in output, output) self.failUnless("fingerprint: 737p57x6737p57x6737p57x6737p57x6737p57x6737p57x6737a" in output, output) output = self._dump_cap("--client-secret", "5s33nk3qpvnj2fw3z4mnm2y6fa", u.to_string()) self.failUnless("file renewal secret: arpszxzc2t6kb4okkg7sp765xgkni5z7caavj7lta73vmtymjlxq" in output, output) fileutil.make_dirs("cli/test_dump_cap/private") fileutil.write("cli/test_dump_cap/private/secret", "5s33nk3qpvnj2fw3z4mnm2y6fa\n") output = self._dump_cap("--client-dir", "cli/test_dump_cap", u.to_string()) self.failUnless("file renewal secret: arpszxzc2t6kb4okkg7sp765xgkni5z7caavj7lta73vmtymjlxq" in output, output) output = self._dump_cap("--client-dir", "cli/test_dump_cap_BOGUS", u.to_string()) self.failIf("file renewal secret:" in output, output) output = self._dump_cap("--nodeid", "tqc35esocrvejvg4mablt6aowg6tl43j", u.to_string()) self.failUnless("write_enabler: mgcavriox2wlb5eer26unwy5cw56elh3sjweffckkmivvsxtaknq" in output, output) self.failIf("file renewal secret:" in output, output) output = self._dump_cap("--nodeid", "tqc35esocrvejvg4mablt6aowg6tl43j", "--client-secret", "5s33nk3qpvnj2fw3z4mnm2y6fa", u.to_string()) self.failUnless("write_enabler: mgcavriox2wlb5eer26unwy5cw56elh3sjweffckkmivvsxtaknq" in output, output) self.failUnless("file renewal secret: arpszxzc2t6kb4okkg7sp765xgkni5z7caavj7lta73vmtymjlxq" in output, output) self.failUnless("lease renewal secret: 7pjtaumrb7znzkkbvekkmuwpqfjyfyamznfz4bwwvmh4nw33lorq" in output, output) u = u.get_readonly() output = self._dump_cap(u.to_string()) self.failUnless("MDMF Read-only URI:" in output, output) self.failUnless("readkey: nvgh5vj2ekzzkim5fgtb4gey5y" in output, output) self.failUnless("storage index: nt4fwemuw7flestsezvo2eveke" in output, output) self.failUnless("fingerprint: 737p57x6737p57x6737p57x6737p57x6737p57x6737p57x6737a" in output, output) u = u.get_verify_cap() output = self._dump_cap(u.to_string()) self.failUnless("MDMF Verifier URI:" in output, output) self.failUnless("storage index: nt4fwemuw7flestsezvo2eveke" in output, output) self.failUnless("fingerprint: 737p57x6737p57x6737p57x6737p57x6737p57x6737p57x6737a" in output, output) def test_dump_cap_chk_directory(self): key = b"\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f" uri_extension_hash = hashutil.uri_extension_hash(b"stuff") needed_shares = 25 total_shares = 100 size = 1234 u1 = uri.CHKFileURI(key=key, uri_extension_hash=uri_extension_hash, needed_shares=needed_shares, total_shares=total_shares, size=size) u = uri.ImmutableDirectoryURI(u1) output = self._dump_cap(u.to_string()) self.failUnless("CHK Directory URI:" in output, output) self.failUnless("key: aaaqeayeaudaocajbifqydiob4" in output, output) self.failUnless("UEB hash: nf3nimquen7aeqm36ekgxomalstenpkvsdmf6fplj7swdatbv5oa" in output, output) self.failUnless("size: 1234" in output, output) self.failUnless("k/N: 25/100" in output, output) self.failUnless("storage index: hdis5iaveku6lnlaiccydyid7q" in output, output) output = self._dump_cap("--client-secret", "5s33nk3qpvnj2fw3z4mnm2y6fa", u.to_string()) self.failUnless("file renewal secret: csrvkjgomkyyyil5yo4yk5np37p6oa2ve2hg6xmk2dy7kaxsu6xq" in output, output) u = u.get_verify_cap() output = self._dump_cap(u.to_string()) self.failUnless("CHK Directory Verifier URI:" in output, output) self.failIf("key: " in output, output) self.failUnless("UEB hash: nf3nimquen7aeqm36ekgxomalstenpkvsdmf6fplj7swdatbv5oa" in output, output) self.failUnless("size: 1234" in output, output) self.failUnless("k/N: 25/100" in output, output) self.failUnless("storage index: hdis5iaveku6lnlaiccydyid7q" in output, output) def test_dump_cap_sdmf_directory(self): writekey = b"\x01" * 16 fingerprint = b"\xfe" * 32 u1 = uri.WriteableSSKFileURI(writekey, fingerprint) u = uri.DirectoryURI(u1) output = self._dump_cap(u.to_string()) self.failUnless("Directory Writeable URI:" in output, output) self.failUnless("writekey: aeaqcaibaeaqcaibaeaqcaibae" in output, output) self.failUnless("readkey: nvgh5vj2ekzzkim5fgtb4gey5y" in output, output) self.failUnless("storage index: nt4fwemuw7flestsezvo2eveke" in output, output) self.failUnless("fingerprint: 737p57x6737p57x6737p57x6737p57x6737p57x6737p57x6737a" in output, output) output = self._dump_cap("--client-secret", "5s33nk3qpvnj2fw3z4mnm2y6fa", u.to_string()) self.failUnless("file renewal secret: arpszxzc2t6kb4okkg7sp765xgkni5z7caavj7lta73vmtymjlxq" in output, output) output = self._dump_cap("--nodeid", "tqc35esocrvejvg4mablt6aowg6tl43j", u.to_string()) self.failUnless("write_enabler: mgcavriox2wlb5eer26unwy5cw56elh3sjweffckkmivvsxtaknq" in output, output) self.failIf("file renewal secret:" in output, output) output = self._dump_cap("--nodeid", "tqc35esocrvejvg4mablt6aowg6tl43j", "--client-secret", "5s33nk3qpvnj2fw3z4mnm2y6fa", u.to_string()) self.failUnless("write_enabler: mgcavriox2wlb5eer26unwy5cw56elh3sjweffckkmivvsxtaknq" in output, output) self.failUnless("file renewal secret: arpszxzc2t6kb4okkg7sp765xgkni5z7caavj7lta73vmtymjlxq" in output, output) self.failUnless("lease renewal secret: 7pjtaumrb7znzkkbvekkmuwpqfjyfyamznfz4bwwvmh4nw33lorq" in output, output) u = u.get_readonly() output = self._dump_cap(u.to_string()) self.failUnless("Directory Read-only URI:" in output, output) self.failUnless("readkey: nvgh5vj2ekzzkim5fgtb4gey5y" in output, output) self.failUnless("storage index: nt4fwemuw7flestsezvo2eveke" in output, output) self.failUnless("fingerprint: 737p57x6737p57x6737p57x6737p57x6737p57x6737p57x6737a" in output, output) u = u.get_verify_cap() output = self._dump_cap(u.to_string()) self.failUnless("Directory Verifier URI:" in output, output) self.failUnless("storage index: nt4fwemuw7flestsezvo2eveke" in output, output) self.failUnless("fingerprint: 737p57x6737p57x6737p57x6737p57x6737p57x6737p57x6737a" in output, output) def test_dump_cap_mdmf_directory(self): writekey = b"\x01" * 16 fingerprint = b"\xfe" * 32 u1 = uri.WriteableMDMFFileURI(writekey, fingerprint) u = uri.MDMFDirectoryURI(u1) output = self._dump_cap(u.to_string()) self.failUnless("Directory Writeable URI:" in output, output) self.failUnless("writekey: aeaqcaibaeaqcaibaeaqcaibae" in output, output) self.failUnless("readkey: nvgh5vj2ekzzkim5fgtb4gey5y" in output, output) self.failUnless("storage index: nt4fwemuw7flestsezvo2eveke" in output, output) self.failUnless("fingerprint: 737p57x6737p57x6737p57x6737p57x6737p57x6737p57x6737a" in output, output) output = self._dump_cap("--client-secret", "5s33nk3qpvnj2fw3z4mnm2y6fa", u.to_string()) self.failUnless("file renewal secret: arpszxzc2t6kb4okkg7sp765xgkni5z7caavj7lta73vmtymjlxq" in output, output) output = self._dump_cap("--nodeid", "tqc35esocrvejvg4mablt6aowg6tl43j", u.to_string()) self.failUnless("write_enabler: mgcavriox2wlb5eer26unwy5cw56elh3sjweffckkmivvsxtaknq" in output, output) self.failIf("file renewal secret:" in output, output) output = self._dump_cap("--nodeid", "tqc35esocrvejvg4mablt6aowg6tl43j", "--client-secret", "5s33nk3qpvnj2fw3z4mnm2y6fa", u.to_string()) self.failUnless("write_enabler: mgcavriox2wlb5eer26unwy5cw56elh3sjweffckkmivvsxtaknq" in output, output) self.failUnless("file renewal secret: arpszxzc2t6kb4okkg7sp765xgkni5z7caavj7lta73vmtymjlxq" in output, output) self.failUnless("lease renewal secret: 7pjtaumrb7znzkkbvekkmuwpqfjyfyamznfz4bwwvmh4nw33lorq" in output, output) u = u.get_readonly() output = self._dump_cap(u.to_string()) self.failUnless("Directory Read-only URI:" in output, output) self.failUnless("readkey: nvgh5vj2ekzzkim5fgtb4gey5y" in output, output) self.failUnless("storage index: nt4fwemuw7flestsezvo2eveke" in output, output) self.failUnless("fingerprint: 737p57x6737p57x6737p57x6737p57x6737p57x6737p57x6737a" in output, output) u = u.get_verify_cap() output = self._dump_cap(u.to_string()) self.failUnless("Directory Verifier URI:" in output, output) self.failUnless("storage index: nt4fwemuw7flestsezvo2eveke" in output, output) self.failUnless("fingerprint: 737p57x6737p57x6737p57x6737p57x6737p57x6737p57x6737a" in output, output) def _catalog_shares(self, *basedirs): o = debug.CatalogSharesOptions() o.stdout,o.stderr = StringIO(), StringIO() args = list(basedirs) o.parseOptions(args) debug.catalog_shares(o) out = o.stdout.getvalue() err = o.stderr.getvalue() return out, err def test_catalog_shares_error(self): nodedir1 = "cli/test_catalog_shares/node1" sharedir = os.path.join(nodedir1, "storage", "shares", "mq", "mqfblse6m5a6dh45isu2cg7oji") fileutil.make_dirs(sharedir) fileutil.write("cli/test_catalog_shares/node1/storage/shares/mq/not-a-dir", "") # write a bogus share that looks a little bit like CHK fileutil.write(os.path.join(sharedir, "8"), b"\x00\x00\x00\x01" + b"\xff" * 200) # this triggers an assert nodedir2 = "cli/test_catalog_shares/node2" fileutil.make_dirs(nodedir2) fileutil.write("cli/test_catalog_shares/node1/storage/shares/not-a-dir", "") # now make sure that the 'catalog-shares' commands survives the error out, err = self._catalog_shares(nodedir1, nodedir2) self.assertEqual(out, "") self.failUnless("Error processing " in err, "didn't see 'error processing' in '%s'" % err) #self.failUnless(nodedir1 in err, # "didn't see '%s' in '%s'" % (nodedir1, err)) # windows mangles the path, and os.path.join isn't enough to make # up for it, so just look for individual strings self.failUnless("node1" in err, "didn't see 'node1' in '%s'" % err) self.failUnless("mqfblse6m5a6dh45isu2cg7oji" in err, "didn't see 'mqfblse6m5a6dh45isu2cg7oji' in '%s'" % err) def test_alias(self): def s128(c): return base32.b2a(c*(128//8)) def s256(c): return base32.b2a(c*(256//8)) TA = b"URI:DIR2:%s:%s" % (s128(b"T"), s256(b"T")) WA = b"URI:DIR2:%s:%s" % (s128(b"W"), s256(b"W")) CA = b"URI:DIR2:%s:%s" % (s128(b"C"), s256(b"C")) aliases = {"tahoe": TA, "work": WA, "c": CA} def ga1(path): return get_alias(aliases, path, u"tahoe") uses_lettercolon = common.platform_uses_lettercolon_drivename() self.failUnlessReallyEqual(ga1(u"bare"), (TA, b"bare")) self.failUnlessReallyEqual(ga1(u"baredir/file"), (TA, b"baredir/file")) self.failUnlessReallyEqual(ga1(u"baredir/file:7"), (TA, b"baredir/file:7")) self.failUnlessReallyEqual(ga1(u"tahoe:"), (TA, b"")) self.failUnlessReallyEqual(ga1(u"tahoe:file"), (TA, b"file")) self.failUnlessReallyEqual(ga1(u"tahoe:dir/file"), (TA, b"dir/file")) self.failUnlessReallyEqual(ga1(u"work:"), (WA, b"")) self.failUnlessReallyEqual(ga1(u"work:file"), (WA, b"file")) self.failUnlessReallyEqual(ga1(u"work:dir/file"), (WA, b"dir/file")) # default != None means we really expect a tahoe path, regardless of # whether we're on windows or not. This is what 'tahoe get' uses. self.failUnlessReallyEqual(ga1(u"c:"), (CA, b"")) self.failUnlessReallyEqual(ga1(u"c:file"), (CA, b"file")) self.failUnlessReallyEqual(ga1(u"c:dir/file"), (CA, b"dir/file")) self.failUnlessReallyEqual(ga1(u"URI:stuff"), (b"URI:stuff", b"")) self.failUnlessReallyEqual(ga1(u"URI:stuff/file"), (b"URI:stuff", b"file")) self.failUnlessReallyEqual(ga1(u"URI:stuff:./file"), (b"URI:stuff", b"file")) self.failUnlessReallyEqual(ga1(u"URI:stuff/dir/file"), (b"URI:stuff", b"dir/file")) self.failUnlessReallyEqual(ga1(u"URI:stuff:./dir/file"), (b"URI:stuff", b"dir/file")) self.failUnlessRaises(common.UnknownAliasError, ga1, u"missing:") self.failUnlessRaises(common.UnknownAliasError, ga1, u"missing:dir") self.failUnlessRaises(common.UnknownAliasError, ga1, u"missing:dir/file") def ga2(path): return get_alias(aliases, path, None) self.failUnlessReallyEqual(ga2(u"bare"), (DefaultAliasMarker, b"bare")) self.failUnlessReallyEqual(ga2(u"baredir/file"), (DefaultAliasMarker, b"baredir/file")) self.failUnlessReallyEqual(ga2(u"baredir/file:7"), (DefaultAliasMarker, b"baredir/file:7")) self.failUnlessReallyEqual(ga2(u"baredir/sub:1/file:7"), (DefaultAliasMarker, b"baredir/sub:1/file:7")) self.failUnlessReallyEqual(ga2(u"tahoe:"), (TA, b"")) self.failUnlessReallyEqual(ga2(u"tahoe:file"), (TA, b"file")) self.failUnlessReallyEqual(ga2(u"tahoe:dir/file"), (TA, b"dir/file")) # on windows, we really want c:foo to indicate a local file. # default==None is what 'tahoe cp' uses. if uses_lettercolon: self.failUnlessReallyEqual(ga2(u"c:"), (DefaultAliasMarker, b"c:")) self.failUnlessReallyEqual(ga2(u"c:file"), (DefaultAliasMarker, b"c:file")) self.failUnlessReallyEqual(ga2(u"c:dir/file"), (DefaultAliasMarker, b"c:dir/file")) else: self.failUnlessReallyEqual(ga2(u"c:"), (CA, b"")) self.failUnlessReallyEqual(ga2(u"c:file"), (CA, b"file")) self.failUnlessReallyEqual(ga2(u"c:dir/file"), (CA, b"dir/file")) self.failUnlessReallyEqual(ga2(u"work:"), (WA, b"")) self.failUnlessReallyEqual(ga2(u"work:file"), (WA, b"file")) self.failUnlessReallyEqual(ga2(u"work:dir/file"), (WA, b"dir/file")) self.failUnlessReallyEqual(ga2(u"URI:stuff"), (b"URI:stuff", b"")) self.failUnlessReallyEqual(ga2(u"URI:stuff/file"), (b"URI:stuff", b"file")) self.failUnlessReallyEqual(ga2(u"URI:stuff:./file"), (b"URI:stuff", b"file")) self.failUnlessReallyEqual(ga2(u"URI:stuff/dir/file"), (b"URI:stuff", b"dir/file")) self.failUnlessReallyEqual(ga2(u"URI:stuff:./dir/file"), (b"URI:stuff", b"dir/file")) self.failUnlessRaises(common.UnknownAliasError, ga2, u"missing:") self.failUnlessRaises(common.UnknownAliasError, ga2, u"missing:dir") self.failUnlessRaises(common.UnknownAliasError, ga2, u"missing:dir/file") def ga3(path): old = common.pretend_platform_uses_lettercolon try: common.pretend_platform_uses_lettercolon = True retval = get_alias(aliases, path, None) finally: common.pretend_platform_uses_lettercolon = old return retval self.failUnlessReallyEqual(ga3(u"bare"), (DefaultAliasMarker, b"bare")) self.failUnlessReallyEqual(ga3(u"baredir/file"), (DefaultAliasMarker, b"baredir/file")) self.failUnlessReallyEqual(ga3(u"baredir/file:7"), (DefaultAliasMarker, b"baredir/file:7")) self.failUnlessReallyEqual(ga3(u"baredir/sub:1/file:7"), (DefaultAliasMarker, b"baredir/sub:1/file:7")) self.failUnlessReallyEqual(ga3(u"tahoe:"), (TA, b"")) self.failUnlessReallyEqual(ga3(u"tahoe:file"), (TA, b"file")) self.failUnlessReallyEqual(ga3(u"tahoe:dir/file"), (TA, b"dir/file")) self.failUnlessReallyEqual(ga3(u"c:"), (DefaultAliasMarker, b"c:")) self.failUnlessReallyEqual(ga3(u"c:file"), (DefaultAliasMarker, b"c:file")) self.failUnlessReallyEqual(ga3(u"c:dir/file"), (DefaultAliasMarker, b"c:dir/file")) self.failUnlessReallyEqual(ga3(u"work:"), (WA, b"")) self.failUnlessReallyEqual(ga3(u"work:file"), (WA, b"file")) self.failUnlessReallyEqual(ga3(u"work:dir/file"), (WA, b"dir/file")) self.failUnlessReallyEqual(ga3(u"URI:stuff"), (b"URI:stuff", b"")) self.failUnlessReallyEqual(ga3(u"URI:stuff:./file"), (b"URI:stuff", b"file")) self.failUnlessReallyEqual(ga3(u"URI:stuff:./dir/file"), (b"URI:stuff", b"dir/file")) self.failUnlessRaises(common.UnknownAliasError, ga3, u"missing:") self.failUnlessRaises(common.UnknownAliasError, ga3, u"missing:dir") self.failUnlessRaises(common.UnknownAliasError, ga3, u"missing:dir/file") # calling get_alias with a path that doesn't include an alias and # default set to something that isn't in the aliases argument should # raise an UnknownAliasError. def ga4(path): return get_alias(aliases, path, u"badddefault:") self.failUnlessRaises(common.UnknownAliasError, ga4, u"afile") self.failUnlessRaises(common.UnknownAliasError, ga4, u"a/dir/path/") def ga5(path): old = common.pretend_platform_uses_lettercolon try: common.pretend_platform_uses_lettercolon = True retval = get_alias(aliases, path, u"baddefault:") finally: common.pretend_platform_uses_lettercolon = old return retval self.failUnlessRaises(common.UnknownAliasError, ga5, u"C:\\Windows") def test_alias_tolerance(self): def s128(c): return base32.b2a(c*(128//8)) def s256(c): return base32.b2a(c*(256//8)) TA = b"URI:DIR2:%s:%s" % (s128(b"T"), s256(b"T")) aliases = {"present": TA, "future": b"URI-FROM-FUTURE:ooh:aah"} def ga1(path): return get_alias(aliases, path, u"tahoe") self.failUnlessReallyEqual(ga1(u"present:file"), (TA, b"file")) # this throws, via assert IDirnodeURI.providedBy(), since get_alias() # wants a dirnode, and the future cap gives us UnknownURI instead. self.failUnlessRaises(AssertionError, ga1, u"future:stuff") def test_listdir_unicode_good(self): filenames = [u'L\u00F4zane', u'Bern', u'Gen\u00E8ve'] # must be NFC for name in filenames: skip_if_cannot_represent_filename(name) basedir = "cli/common/listdir_unicode_good" fileutil.make_dirs(basedir) for name in filenames: open(os.path.join(str(basedir), name), "wb").close() for file in listdir_unicode(str(basedir)): self.failUnlessIn(normalize(file), filenames) def test_exception_catcher(self): """ An exception that is otherwise unhandled during argument dispatch is written to stderr and causes the process to exit with code 1. """ self.basedir = "cli/exception_catcher" exc = Exception("canary") class BrokenOptions: def parseOptions(self, argv): raise exc stderr = StringIO() reactor = MemoryReactor() with AlternateReactor(reactor): with self.assertRaises(SystemExit) as ctx: runner.run( configFactory=BrokenOptions, argv=["tahoe"], stderr=stderr, ) self.assertTrue(reactor.hasRun) self.assertFalse(reactor.running) self.failUnlessIn(str(exc), stderr.getvalue()) self.assertEqual(1, ctx.exception.code) class Help(unittest.TestCase): def failUnlessInNormalized(self, x, y): # helper function to deal with the --help output being wrapped to # various widths, depending on the $COLUMNS environment variable self.failUnlessIn(x.replace("\n", " "), y.replace("\n", " ")) def test_get(self): help = str(cli.GetOptions()) self.failUnlessIn("[options] REMOTE_FILE LOCAL_FILE", help) self.failUnlessIn("% tahoe get FOO |less", help) def test_put(self): help = str(cli.PutOptions()) self.failUnlessIn("[options] LOCAL_FILE REMOTE_FILE", help) self.failUnlessIn("% cat FILE | tahoe put", help) def test_ls(self): help = str(cli.ListOptions()) self.failUnlessIn("[options] [PATH]", help) def test_unlink(self): help = str(cli.UnlinkOptions()) self.failUnlessIn("[options] REMOTE_FILE", help) def test_mv(self): help = str(cli.MvOptions()) self.failUnlessIn("[options] FROM TO", help) self.failUnlessInNormalized("Use 'tahoe mv' to move files", help) def test_cp(self): help = str(cli.CpOptions()) self.failUnlessIn("[options] FROM.. TO", help) self.failUnlessInNormalized("Use 'tahoe cp' to copy files", help) def test_ln(self): help = str(cli.LnOptions()) self.failUnlessIn("[options] FROM_LINK TO_LINK", help) self.failUnlessInNormalized("Use 'tahoe ln' to duplicate a link", help) def test_mkdir(self): help = str(cli.MakeDirectoryOptions()) self.failUnlessIn("[options] [REMOTE_DIR]", help) self.failUnlessInNormalized("Create a new directory", help) def test_backup(self): help = str(cli.BackupOptions()) self.failUnlessIn("[options] FROM ALIAS:TO", help) def test_webopen(self): help = str(cli.WebopenOptions()) self.failUnlessIn("[options] [ALIAS:PATH]", help) def test_manifest(self): help = str(cli.ManifestOptions()) self.failUnlessIn("[options] [ALIAS:PATH]", help) def test_stats(self): help = str(cli.StatsOptions()) self.failUnlessIn("[options] [ALIAS:PATH]", help) def test_check(self): help = str(cli.CheckOptions()) self.failUnlessIn("[options] [ALIAS:PATH]", help) def test_deep_check(self): help = str(cli.DeepCheckOptions()) self.failUnlessIn("[options] [ALIAS:PATH]", help) def test_create_alias(self): help = str(cli.CreateAliasOptions()) self.failUnlessIn("[options] ALIAS[:]", help) def test_add_alias(self): help = str(cli.AddAliasOptions()) self.failUnlessIn("[options] ALIAS[:] DIRCAP", help) def test_list_aliases(self): help = str(cli.ListAliasesOptions()) self.failUnlessIn("[options]", help) def test_run(self): help = str(tahoe_run.RunOptions()) self.failUnlessIn("[options] [NODEDIR [twistd-options]]", help) def test_create_client(self): help = str(create_node.CreateClientOptions()) self.failUnlessIn("[options] [NODEDIR]", help) def test_create_node(self): help = str(create_node.CreateNodeOptions()) self.failUnlessIn("[options] [NODEDIR]", help) def test_create_introducer(self): help = str(create_node.CreateIntroducerOptions()) self.failUnlessIn("[options] NODEDIR", help) def test_debug_flogtool(self): options = debug.FlogtoolOptions() help = str(options) self.failUnlessIn(" [global-options] debug flogtool ", help) self.failUnlessInNormalized("The 'tahoe debug flogtool' command uses the correct imports", help) for (option, shortcut, oClass, desc) in options.subCommands: subhelp = str(oClass()) self.failUnlessIn(" [global-options] debug flogtool %s " % (option,), subhelp) class Ln(GridTestMixin, CLITestMixin, unittest.TestCase): def _create_test_file(self): data = "puppies" * 1000 path = os.path.join(self.basedir, "datafile") fileutil.write(path, data) self.datafile = path def test_ln_without_alias(self): # if invoked without an alias when the 'tahoe' alias doesn't # exist, 'tahoe ln' should output a useful error message and not # a stack trace self.basedir = "cli/Ln/ln_without_alias" self.set_up_grid(oneshare=True) d = self.do_cli("ln", "from", "to") def _check(args): (rc, out, err) = args self.failUnlessReallyEqual(rc, 1) self.failUnlessIn("error:", err) self.assertEqual(out, "") d.addCallback(_check) # Make sure that validation extends to the "to" parameter d.addCallback(lambda ign: self.do_cli("create-alias", "havasu")) d.addCallback(lambda ign: self._create_test_file()) d.addCallback(lambda ign: self.do_cli("put", self.datafile, "havasu:from")) d.addCallback(lambda ign: self.do_cli("ln", "havasu:from", "to")) d.addCallback(_check) return d def test_ln_with_nonexistent_alias(self): # If invoked with aliases that don't exist, 'tahoe ln' should # output a useful error message and not a stack trace. self.basedir = "cli/Ln/ln_with_nonexistent_alias" self.set_up_grid(oneshare=True) d = self.do_cli("ln", "havasu:from", "havasu:to") def _check(args): (rc, out, err) = args self.failUnlessReallyEqual(rc, 1) self.failUnlessIn("error:", err) d.addCallback(_check) # Make sure that validation occurs on the to parameter if the # from parameter passes. d.addCallback(lambda ign: self.do_cli("create-alias", "havasu")) d.addCallback(lambda ign: self._create_test_file()) d.addCallback(lambda ign: self.do_cli("put", self.datafile, "havasu:from")) d.addCallback(lambda ign: self.do_cli("ln", "havasu:from", "huron:to")) d.addCallback(_check) return d class Admin(unittest.TestCase): def test_generate_keypair(self): d = run_cli("admin", "generate-keypair") def _done(args): (rc, stdout, stderr) = args lines = [line.strip() for line in stdout.splitlines()] privkey_bits = lines[0].split() pubkey_bits = lines[1].split() sk_header = "private:" vk_header = "public:" self.failUnlessEqual(privkey_bits[0], sk_header, lines[0]) self.failUnlessEqual(pubkey_bits[0], vk_header, lines[1]) self.failUnless(privkey_bits[1].startswith("priv-v0-"), lines[0]) self.failUnless(pubkey_bits[1].startswith("pub-v0-"), lines[1]) sk, pk = ed25519.signing_keypair_from_string( privkey_bits[1].encode("ascii")) vk_bytes = pubkey_bits[1].encode("ascii") self.assertEqual( ed25519.string_from_verifying_key(pk), vk_bytes, ) d.addCallback(_done) return d def test_derive_pubkey(self): priv_key, pub_key = ed25519.create_signing_keypair() priv_key_str = str(ed25519.string_from_signing_key(priv_key), "ascii") pub_key_str = str(ed25519.string_from_verifying_key(pub_key), "ascii") d = run_cli("admin", "derive-pubkey", priv_key_str) def _done(args): (rc, stdout, stderr) = args lines = stdout.split("\n") privkey_line = lines[0].strip() pubkey_line = lines[1].strip() sk_header = "private: priv-v0-" vk_header = "public: pub-v0-" self.failUnless(privkey_line.startswith(sk_header), privkey_line) self.failUnless(pubkey_line.startswith(vk_header), pubkey_line) pub_key_str2 = pubkey_line[len(vk_header):] self.assertEqual("pub-v0-" + pub_key_str2, pub_key_str) d.addCallback(_done) return d class Errors(GridTestMixin, CLITestMixin, unittest.TestCase): def test_get(self): self.basedir = "cli/Errors/get" self.set_up_grid() c0 = self.g.clients[0] self.fileurls = {} DATA = b"data" * 100 d = c0.upload(upload.Data(DATA, convergence=b"")) def _stash_bad(ur): self.uri_1share = ur.get_uri() self.delete_shares_numbered(ur.get_uri(), list(range(1,10))) d.addCallback(_stash_bad) # the download is abandoned as soon as it's clear that we won't get # enough shares. The one remaining share might be in either the # COMPLETE or the PENDING state. in_complete_msg = "ran out of shares: complete=sh0 pending= overdue= unused= need 3" in_pending_msg_regex = r"ran out of shares: complete= pending=Share\(.+\) overdue= unused= need 3" d.addCallback(lambda ign: self.do_cli("get", self.uri_1share)) def _check1(args): (rc, out, err) = args self.failIfEqual(rc, 0) self.failUnless("410 Gone" in err, err) self.failUnlessIn("NotEnoughSharesError: ", err) self.failUnless(in_complete_msg in err or re.search(in_pending_msg_regex, err)) d.addCallback(_check1) targetf = os.path.join(self.basedir, "output") d.addCallback(lambda ign: self.do_cli("get", self.uri_1share, targetf)) def _check2(args): (rc, out, err) = args self.failIfEqual(rc, 0) self.failUnless("410 Gone" in err, err) self.failUnlessIn("NotEnoughSharesError: ", err) self.failUnless(in_complete_msg in err or re.search(in_pending_msg_regex, err)) self.failIf(os.path.exists(targetf)) d.addCallback(_check2) return d def test_broken_socket(self): # When the http connection breaks (such as when node.url is overwritten # by a confused user), a user friendly error message should be printed. self.basedir = "cli/Errors/test_broken_socket" self.set_up_grid(oneshare=True) # Simulate a connection error def _socket_error(*args, **kwargs): raise socket_error('test error') self.patch(allmydata.scripts.common_http.http_client.HTTPConnection, "endheaders", _socket_error) d = self.do_cli("mkdir") def _check_invalid(args): (rc, stdout, stderr) = args self.failIfEqual(rc, 0) self.failUnlessIn("Error trying to connect to http://127.0.0.1", stderr) d.addCallback(_check_invalid) return d class Get(GridTestMixin, CLITestMixin, unittest.TestCase): def test_get_without_alias(self): # 'tahoe get' should output a useful error message when invoked # without an explicit alias and when the default 'tahoe' alias # hasn't been created yet. self.basedir = "cli/Get/get_without_alias" self.set_up_grid(oneshare=True) d = self.do_cli('get', 'file') def _check(args): (rc, out, err) = args self.failUnlessReallyEqual(rc, 1) self.failUnlessIn("error:", err) self.assertEqual(out, "") d.addCallback(_check) return d def test_get_with_nonexistent_alias(self): # 'tahoe get' should output a useful error message when invoked with # an explicit alias that doesn't exist. self.basedir = "cli/Get/get_with_nonexistent_alias" self.set_up_grid(oneshare=True) d = self.do_cli("get", "nonexistent:file") def _check(args): (rc, out, err) = args self.failUnlessReallyEqual(rc, 1) self.failUnlessIn("error:", err) self.failUnlessIn("nonexistent", err) self.assertEqual(out, "") d.addCallback(_check) return d class Manifest(GridTestMixin, CLITestMixin, unittest.TestCase): def test_manifest_without_alias(self): # 'tahoe manifest' should output a useful error message when invoked # without an explicit alias when the default 'tahoe' alias is # missing. self.basedir = "cli/Manifest/manifest_without_alias" self.set_up_grid(oneshare=True) d = self.do_cli("manifest") def _check(args): (rc, out, err) = args self.failUnlessReallyEqual(rc, 1) self.failUnlessIn("error:", err) self.assertEqual(out, "") d.addCallback(_check) return d def test_manifest_with_nonexistent_alias(self): # 'tahoe manifest' should output a useful error message when invoked # with an explicit alias that doesn't exist. self.basedir = "cli/Manifest/manifest_with_nonexistent_alias" self.set_up_grid(oneshare=True) d = self.do_cli("manifest", "nonexistent:") def _check(args): (rc, out, err) = args self.failUnlessReallyEqual(rc, 1) self.failUnlessIn("error:", err) self.failUnlessIn("nonexistent", err) self.assertEqual(out, "") d.addCallback(_check) return d class Mkdir(GridTestMixin, CLITestMixin, unittest.TestCase): def test_mkdir(self): self.basedir = os.path.dirname(self.mktemp()) self.set_up_grid(oneshare=True) d = self.do_cli("create-alias", "tahoe") d.addCallback(lambda res: self.do_cli("mkdir", "test")) def _check(args): (rc, out, err) = args self.failUnlessReallyEqual(rc, 0) self.assertEqual(err, "") self.failUnlessIn("URI:", out) d.addCallback(_check) return d def test_mkdir_mutable_type(self): self.basedir = os.path.dirname(self.mktemp()) self.set_up_grid(oneshare=True) d = self.do_cli("create-alias", "tahoe") def _check(args, st): (rc, out, err) = args self.failUnlessReallyEqual(rc, 0) self.assertEqual(err, "") self.failUnlessIn(st, out) return out def _mkdir(ign, mutable_type, uri_prefix, dirname): """ :param str mutable_type: 'sdmf' or 'mdmf' (or uppercase versions) :param str uri_prefix: kind of URI :param str dirname: the directory alias """ d2 = self.do_cli("mkdir", "--format={}".format(mutable_type), dirname) d2.addCallback(_check, uri_prefix) def _stash_filecap(cap): u = uri.from_string(cap) fn_uri = u.get_filenode_cap() self._filecap = fn_uri.to_string() d2.addCallback(_stash_filecap) d2.addCallback(lambda ign: self.do_cli("ls", "--json", dirname)) d2.addCallback(_check, uri_prefix) d2.addCallback(lambda ign: self.do_cli("ls", "--json", self._filecap)) d2.addCallback(_check, '"format": "%s"' % (mutable_type.upper(),)) return d2 d.addCallback(_mkdir, "sdmf", "URI:DIR2", "tahoe:foo") d.addCallback(_mkdir, "SDMF", "URI:DIR2", "tahoe:foo2") d.addCallback(_mkdir, "mdmf", "URI:DIR2-MDMF", "tahoe:bar") d.addCallback(_mkdir, "MDMF", "URI:DIR2-MDMF", "tahoe:bar2") return d def test_mkdir_mutable_type_unlinked(self): self.basedir = os.path.dirname(self.mktemp()) self.set_up_grid(oneshare=True) d = self.do_cli("mkdir", "--format=SDMF") def _check(args, st): (rc, out, err) = args self.failUnlessReallyEqual(rc, 0) self.assertEqual(err, "") self.failUnlessIn(st, out) return out d.addCallback(_check, "URI:DIR2") def _stash_dircap(cap): self._dircap = cap # Now we're going to feed the cap into uri.from_string... u = uri.from_string(cap) # ...grab the underlying filenode uri. fn_uri = u.get_filenode_cap() # ...and stash that. self._filecap = fn_uri.to_string() d.addCallback(_stash_dircap) d.addCallback(lambda res: self.do_cli("ls", "--json", self._filecap)) d.addCallback(_check, '"format": "SDMF"') d.addCallback(lambda res: self.do_cli("mkdir", "--format=MDMF")) d.addCallback(_check, "URI:DIR2-MDMF") d.addCallback(_stash_dircap) d.addCallback(lambda res: self.do_cli("ls", "--json", self._filecap)) d.addCallback(_check, '"format": "MDMF"') return d def test_mkdir_bad_mutable_type(self): o = cli.MakeDirectoryOptions() self.failUnlessRaises(usage.UsageError, o.parseOptions, ["--format=LDMF"]) def test_mkdir_unicode(self): self.basedir = os.path.dirname(self.mktemp()) self.set_up_grid(oneshare=True) try: motorhead_arg = u"tahoe:Mot\u00F6rhead".encode(get_io_encoding()) except UnicodeEncodeError: raise unittest.SkipTest("A non-ASCII command argument could not be encoded on this platform.") d = self.do_cli("create-alias", "tahoe") d.addCallback(lambda res: self.do_cli("mkdir", motorhead_arg)) def _check(args): (rc, out, err) = args self.failUnlessReallyEqual(rc, 0) self.assertEqual(err, "") self.failUnlessIn("URI:", out) d.addCallback(_check) return d def test_mkdir_with_nonexistent_alias(self): # when invoked with an alias that doesn't exist, 'tahoe mkdir' should # output a sensible error message rather than a stack trace. self.basedir = "cli/Mkdir/mkdir_with_nonexistent_alias" self.set_up_grid(oneshare=True) d = self.do_cli("mkdir", "havasu:") def _check(args): (rc, out, err) = args self.failUnlessReallyEqual(rc, 1) self.failUnlessIn("error:", err) self.assertEqual(out, "") d.addCallback(_check) return d class Unlink(GridTestMixin, CLITestMixin, unittest.TestCase): command = "unlink" def _create_test_file(self): data = "puppies" * 1000 path = os.path.join(self.basedir, "datafile") fileutil.write(path, data) self.datafile = path def test_unlink_without_alias(self): # 'tahoe unlink' should behave sensibly when invoked without an explicit # alias before the default 'tahoe' alias has been created. self.basedir = "cli/Unlink/%s_without_alias" % (self.command,) self.set_up_grid(oneshare=True) d = self.do_cli(self.command, "afile") def _check(args): (rc, out, err) = args self.failUnlessReallyEqual(rc, 1) self.failUnlessIn("error:", err) self.assertEqual(out, "") d.addCallback(_check) d.addCallback(lambda ign: self.do_cli(self.command, "afile")) d.addCallback(_check) return d def test_unlink_with_nonexistent_alias(self): # 'tahoe unlink' should behave sensibly when invoked with an explicit # alias that doesn't exist. self.basedir = "cli/Unlink/%s_with_nonexistent_alias" % (self.command,) self.set_up_grid(oneshare=True) d = self.do_cli(self.command, "nonexistent:afile") def _check(args): (rc, out, err) = args self.failUnlessReallyEqual(rc, 1) self.failUnlessIn("error:", err) self.failUnlessIn("nonexistent", err) self.assertEqual(out, "") d.addCallback(_check) d.addCallback(lambda ign: self.do_cli(self.command, "nonexistent:afile")) d.addCallback(_check) return d def test_unlink_without_path(self): # 'tahoe unlink' should give a sensible error message when invoked without a path. self.basedir = "cli/Unlink/%s_without_path" % (self.command,) self.set_up_grid(oneshare=True) self._create_test_file() d = self.do_cli("create-alias", "tahoe") d.addCallback(lambda ign: self.do_cli("put", self.datafile, "tahoe:test")) def _do_unlink(args): (rc, out, err) = args self.failUnlessReallyEqual(rc, 0) self.failUnless(out.startswith("URI:"), out) return self.do_cli(self.command, out.strip('\n')) d.addCallback(_do_unlink) def _check(args): (rc, out, err) = args self.failUnlessReallyEqual(rc, 1) self.failUnlessIn("'tahoe %s'" % (self.command,), err) self.failUnlessIn("path must be given", err) self.assertEqual(out, "") d.addCallback(_check) return d class Stats(GridTestMixin, CLITestMixin, unittest.TestCase): def test_empty_directory(self): self.basedir = "cli/Stats/empty_directory" self.set_up_grid(oneshare=True) c0 = self.g.clients[0] self.fileurls = {} d = c0.create_dirnode() def _stash_root(n): self.rootnode = n self.rooturi = n.get_uri() d.addCallback(_stash_root) # make sure we can get stats on an empty directory too d.addCallback(lambda ign: self.do_cli("stats", self.rooturi)) def _check_stats(args): (rc, out, err) = args self.assertEqual(err, "") self.failUnlessReallyEqual(rc, 0) lines = out.splitlines() self.failUnlessIn(" count-immutable-files: 0", lines) self.failUnlessIn(" count-mutable-files: 0", lines) self.failUnlessIn(" count-literal-files: 0", lines) self.failUnlessIn(" count-directories: 1", lines) self.failUnlessIn(" size-immutable-files: 0", lines) self.failIfIn("Size Histogram:", lines) d.addCallback(_check_stats) return d def test_stats_without_alias(self): # when invoked with no explicit alias and before the default 'tahoe' # alias is created, 'tahoe stats' should output an informative error # message, not a stack trace. self.basedir = "cli/Stats/stats_without_alias" self.set_up_grid(oneshare=True) d = self.do_cli("stats") def _check(args): (rc, out, err) = args self.failUnlessReallyEqual(rc, 1) self.failUnlessIn("error:", err) self.assertEqual(out, "") d.addCallback(_check) return d def test_stats_with_nonexistent_alias(self): # when invoked with an explicit alias that doesn't exist, # 'tahoe stats' should output a useful error message. self.basedir = "cli/Stats/stats_with_nonexistent_alias" self.set_up_grid(oneshare=True) d = self.do_cli("stats", "havasu:") def _check(args): (rc, out, err) = args self.failUnlessReallyEqual(rc, 1) self.failUnlessIn("error:", err) self.assertEqual(out, "") d.addCallback(_check) return d class Webopen(GridTestMixin, CLITestMixin, unittest.TestCase): def test_webopen_with_nonexistent_alias(self): # when invoked with an alias that doesn't exist, 'tahoe webopen' # should output an informative error message instead of a stack # trace. self.basedir = "cli/Webopen/webopen_with_nonexistent_alias" self.set_up_grid(oneshare=True) d = self.do_cli("webopen", "fake:") def _check(args): (rc, out, err) = args self.failUnlessReallyEqual(rc, 1) self.failUnlessIn("error:", err) self.assertEqual(out, "") d.addCallback(_check) return d def test_webopen(self): # TODO: replace with @patch that supports Deferreds. import webbrowser def call_webbrowser_open(url): self.failUnlessIn(str(self.alias_uri, "ascii").replace(':', '%3A'), url) self.webbrowser_open_called = True def _cleanup(res): webbrowser.open = self.old_webbrowser_open return res self.old_webbrowser_open = webbrowser.open try: webbrowser.open = call_webbrowser_open self.basedir = "cli/Webopen/webopen" self.set_up_grid(oneshare=True) d = self.do_cli("create-alias", "alias:") def _check_alias(args): (rc, out, err) = args self.failUnlessReallyEqual(rc, 0, repr((rc, out, err))) self.failUnlessIn("Alias 'alias' created", out) self.assertEqual(err, "") self.alias_uri = get_aliases(self.get_clientdir())["alias"] d.addCallback(_check_alias) d.addCallback(lambda res: self.do_cli("webopen", "alias:")) def _check_webopen(args): (rc, out, err) = args self.failUnlessReallyEqual(rc, 0, repr((rc, out, err))) self.assertEqual(out, "") self.assertEqual(err, "") self.failUnless(self.webbrowser_open_called) d.addCallback(_check_webopen) d.addBoth(_cleanup) except: _cleanup(None) raise return d class Options(ReallyEqualMixin, unittest.TestCase): # this test case only looks at argument-processing and simple stuff. def parse(self, args, stdout=None): o = runner.Options() if stdout is not None: o.stdout = stdout o.parseOptions(args) while hasattr(o, "subOptions"): o = o.subOptions return o def test_list(self): fileutil.rm_dir("cli/test_options") fileutil.make_dirs("cli/test_options") fileutil.make_dirs("cli/test_options/private") fileutil.write("cli/test_options/node.url", "http://localhost:8080/\n") filenode_uri = uri.WriteableSSKFileURI(writekey=b"\x00"*16, fingerprint=b"\x00"*32) private_uri = uri.DirectoryURI(filenode_uri).to_string() fileutil.write("cli/test_options/private/root_dir.cap", private_uri + b"\n") def parse2(args): return parse_options("cli/test_options", "ls", args) o = parse2([]) self.failUnlessEqual(o['node-url'], "http://localhost:8080/") self.failUnlessEqual(o.aliases[DEFAULT_ALIAS].encode("ascii"), private_uri) self.failUnlessEqual(o.where, u"") o = parse2(["--node-url", "http://example.org:8111/"]) self.failUnlessEqual(o['node-url'], "http://example.org:8111/") self.failUnlessEqual(o.aliases[DEFAULT_ALIAS].encode("ascii"), private_uri) self.failUnlessEqual(o.where, u"") # -u for --node-url used to clash with -u for --uri (tickets #1949 and #2137). o = parse2(["-u", "http://example.org:8111/"]) self.failUnlessEqual(o['node-url'], "http://example.org:8111/") self.failUnlessEqual(o.aliases[DEFAULT_ALIAS].encode("ascii"), private_uri) self.failUnlessEqual(o.where, u"") self.failIf(o["uri"]) o = parse2(["-u", "http://example.org:8111/", "--uri"]) self.failUnlessEqual(o['node-url'], "http://example.org:8111/") self.failUnlessEqual(o.aliases[DEFAULT_ALIAS].encode("ascii"), private_uri) self.failUnlessEqual(o.where, u"") self.failUnless(o["uri"]) o = parse2(["--dir-cap", "root"]) self.failUnlessEqual(o['node-url'], "http://localhost:8080/") self.failUnlessEqual(o.aliases[DEFAULT_ALIAS], "root") self.failUnlessEqual(o.where, u"") other_filenode_uri = uri.WriteableSSKFileURI(writekey=b"\x11"*16, fingerprint=b"\x11"*32) other_uri = uri.DirectoryURI(other_filenode_uri).to_string() o = parse2(["--dir-cap", other_uri]) self.failUnlessEqual(o['node-url'], "http://localhost:8080/") self.failUnlessEqual(o.aliases[DEFAULT_ALIAS].encode("ascii"), other_uri) self.failUnlessEqual(o.where, u"") o = parse2(["--dir-cap", other_uri, "subdir"]) self.failUnlessEqual(o['node-url'], "http://localhost:8080/") self.failUnlessEqual(o.aliases[DEFAULT_ALIAS].encode("ascii"), other_uri) self.failUnlessEqual(o.where, u"subdir") self.failUnlessRaises(usage.UsageError, parse2, ["--node-url", "NOT-A-URL"]) o = parse2(["--node-url", "http://localhost:8080"]) self.failUnlessEqual(o["node-url"], "http://localhost:8080/") o = parse2(["--node-url", "https://localhost/"]) self.failUnlessEqual(o["node-url"], "https://localhost/") def test_version(self): # "tahoe --version" dumps text to stdout and exits stdout = StringIO() self.failUnlessRaises(SystemExit, self.parse, ["--version"], stdout) self.failUnlessIn(allmydata.__full_version__, stdout.getvalue()) # but "tahoe SUBCOMMAND --version" should be rejected self.failUnlessRaises(usage.UsageError, self.parse, ["run", "--version"]) self.failUnlessRaises(usage.UsageError, self.parse, ["run", "--version-and-path"]) def test_quiet(self): # accepted as an overall option, but not on subcommands o = self.parse(["--quiet", "run"]) self.failUnless(o.parent["quiet"]) self.failUnlessRaises(usage.UsageError, self.parse, ["run", "--quiet"]) def test_basedir(self): # accept a --node-directory option before the verb, or a --basedir # option after, or a basedir argument after, but none in the wrong # place, and not more than one of the three. # Here is some option twistd recognizes but we don't. Depending on # where it appears, it should be passed through to twistd. It doesn't # really matter which option it is (it doesn't even have to be a valid # option). This test does not actually run any of the twistd argument # parsing. some_twistd_option = "--spew" o = self.parse(["run"]) self.failUnlessReallyEqual(o["basedir"], os.path.join(fileutil.abspath_expanduser_unicode(u"~"), u".tahoe")) o = self.parse(["run", "here"]) self.failUnlessReallyEqual(o["basedir"], fileutil.abspath_expanduser_unicode(u"here")) o = self.parse(["run", "--basedir", "there"]) self.failUnlessReallyEqual(o["basedir"], fileutil.abspath_expanduser_unicode(u"there")) o = self.parse(["--node-directory", "there", "run"]) self.failUnlessReallyEqual(o["basedir"], fileutil.abspath_expanduser_unicode(u"there")) o = self.parse(["run", "here", some_twistd_option]) self.failUnlessReallyEqual(o["basedir"], fileutil.abspath_expanduser_unicode(u"here")) self.failUnlessRaises(usage.UsageError, self.parse, ["--basedir", "there", "run"]) self.failUnlessRaises(usage.UsageError, self.parse, ["run", "--node-directory", "there"]) self.failUnlessRaises(usage.UsageError, self.parse, ["--node-directory=there", "run", "--basedir=here"]) self.failUnlessRaises(usage.UsageError, self.parse, ["run", "--basedir=here", "anywhere"]) self.failUnlessRaises(usage.UsageError, self.parse, ["--node-directory=there", "run", "anywhere"]) self.failUnlessRaises(usage.UsageError, self.parse, ["--node-directory=there", "run", "--basedir=here", "anywhere"]) self.failUnlessRaises(usage.UsageError, self.parse, ["--node-directory=there", "run", some_twistd_option]) self.failUnlessRaises(usage.UsageError, self.parse, ["run", "--basedir=here", some_twistd_option])