src/allmydata/test/test_cli_create_alias.py

   1 import os.path
   2 from twisted.trial import unittest
   3 import urllib
   4 from allmydata.util import fileutil
   5 from allmydata.scripts.common import get_aliases
   6 from allmydata.scripts import cli, runner
   7 from allmydata.test.no_network import GridTestMixin
   8 from allmydata.util.encodingutil import quote_output, get_io_encoding
   9 from .test_cli import CLITestMixin
  10
  11 timeout = 480 # deep_check takes 360s on Zandr's linksys box, others take > 240s
  12
  13 class CreateAlias(GridTestMixin, CLITestMixin, unittest.TestCase):
  14
  15     def _test_webopen(self, args, expected_url):
  16         o = runner.Options()
  17         o.parseOptions(["--node-directory", self.get_clientdir(), "webopen"]
  18                        + list(args))
  19         urls = []
  20         rc = cli.webopen(o, urls.append)
  21         self.failUnlessReallyEqual(rc, 0)
  22         self.failUnlessReallyEqual(len(urls), 1)
  23         self.failUnlessReallyEqual(urls[0], expected_url)
  24
  25     def test_create(self):
  26         self.basedir = "cli/CreateAlias/create"
  27         self.set_up_grid()
  28         aliasfile = os.path.join(self.get_clientdir(), "private", "aliases")
  29
  30         d = self.do_cli("create-alias", "tahoe")
  31         def _done((rc,stdout,stderr)):
  32             self.failUnless("Alias 'tahoe' created" in stdout)
  33             self.failIf(stderr)
  34             aliases = get_aliases(self.get_clientdir())
  35             self.failUnless("tahoe" in aliases)
  36             self.failUnless(aliases["tahoe"].startswith("URI:DIR2:"))
  37         d.addCallback(_done)
  38         d.addCallback(lambda res: self.do_cli("create-alias", "two:"))
  39
  40         def _stash_urls(res):
  41             aliases = get_aliases(self.get_clientdir())
  42             node_url_file = os.path.join(self.get_clientdir(), "node.url")
  43             nodeurl = fileutil.read(node_url_file).strip()
  44             self.welcome_url = nodeurl
  45             uribase = nodeurl + "uri/"
  46             self.tahoe_url = uribase + urllib.quote(aliases["tahoe"])
  47             self.tahoe_subdir_url = self.tahoe_url + "/subdir"
  48             self.two_url = uribase + urllib.quote(aliases["two"])
  49             self.two_uri = aliases["two"]
  50         d.addCallback(_stash_urls)
  51
  52         d.addCallback(lambda res: self.do_cli("create-alias", "two")) # dup
  53         def _check_create_duplicate((rc,stdout,stderr)):
  54             self.failIfEqual(rc, 0)
  55             self.failUnless("Alias 'two' already exists!" in stderr)
  56             aliases = get_aliases(self.get_clientdir())
  57             self.failUnlessReallyEqual(aliases["two"], self.two_uri)
  58         d.addCallback(_check_create_duplicate)
  59
  60         d.addCallback(lambda res: self.do_cli("add-alias", "added", self.two_uri))
  61         def _check_add((rc,stdout,stderr)):
  62             self.failUnlessReallyEqual(rc, 0)
  63             self.failUnless("Alias 'added' added" in stdout)
  64         d.addCallback(_check_add)
  65
  66         # check add-alias with a duplicate
  67         d.addCallback(lambda res: self.do_cli("add-alias", "two", self.two_uri))
  68         def _check_add_duplicate((rc,stdout,stderr)):
  69             self.failIfEqual(rc, 0)
  70             self.failUnless("Alias 'two' already exists!" in stderr)
  71             aliases = get_aliases(self.get_clientdir())
  72             self.failUnlessReallyEqual(aliases["two"], self.two_uri)
  73         d.addCallback(_check_add_duplicate)
  74
  75         # check create-alias and add-alias with invalid aliases
  76         def _check_invalid((rc,stdout,stderr)):
  77             self.failIfEqual(rc, 0)
  78             self.failUnlessIn("cannot contain", stderr)
  79
  80         for invalid in ['foo:bar', 'foo bar', 'foobar::']:
  81             d.addCallback(lambda res, invalid=invalid: self.do_cli("create-alias", invalid))
  82             d.addCallback(_check_invalid)
  83             d.addCallback(lambda res, invalid=invalid: self.do_cli("add-alias", invalid, self.two_uri))
  84             d.addCallback(_check_invalid)
  85
  86         def _test_urls(junk):
  87             self._test_webopen([], self.welcome_url)
  88             self._test_webopen(["/"], self.tahoe_url)
  89             self._test_webopen(["tahoe:"], self.tahoe_url)
  90             self._test_webopen(["tahoe:/"], self.tahoe_url)
  91             self._test_webopen(["tahoe:subdir"], self.tahoe_subdir_url)
  92             self._test_webopen(["-i", "tahoe:subdir"],
  93                                self.tahoe_subdir_url+"?t=info")
  94             self._test_webopen(["tahoe:subdir/"], self.tahoe_subdir_url + '/')
  95             self._test_webopen(["tahoe:subdir/file"],
  96                                self.tahoe_subdir_url + '/file')
  97             self._test_webopen(["--info", "tahoe:subdir/file"],
  98                                self.tahoe_subdir_url + '/file?t=info')
  99             # if "file" is indeed a file, then the url produced by webopen in
 100             # this case is disallowed by the webui. but by design, webopen
 101             # passes through the mistake from the user to the resultant
 102             # webopened url
 103             self._test_webopen(["tahoe:subdir/file/"], self.tahoe_subdir_url + '/file/')
 104             self._test_webopen(["two:"], self.two_url)
 105         d.addCallback(_test_urls)
 106
 107         def _remove_trailing_newline_and_create_alias(ign):
 108             # ticket #741 is about a manually-edited alias file (which
 109             # doesn't end in a newline) being corrupted by a subsequent
 110             # "tahoe create-alias"
 111             old = fileutil.read(aliasfile)
 112             fileutil.write(aliasfile, old.rstrip())
 113             return self.do_cli("create-alias", "un-corrupted1")
 114         d.addCallback(_remove_trailing_newline_and_create_alias)
 115         def _check_not_corrupted1((rc,stdout,stderr)):
 116             self.failUnless("Alias 'un-corrupted1' created" in stdout, stdout)
 117             self.failIf(stderr)
 118             # the old behavior was to simply append the new record, causing a
 119             # line that looked like "NAME1: CAP1NAME2: CAP2". This won't look
 120             # like a valid dircap, so get_aliases() will raise an exception.
 121             aliases = get_aliases(self.get_clientdir())
 122             self.failUnless("added" in aliases)
 123             self.failUnless(aliases["added"].startswith("URI:DIR2:"))
 124             # to be safe, let's confirm that we don't see "NAME2:" in CAP1.
 125             # No chance of a false-negative, because the hyphen in
 126             # "un-corrupted1" is not a valid base32 character.
 127             self.failIfIn("un-corrupted1:", aliases["added"])
 128             self.failUnless("un-corrupted1" in aliases)
 129             self.failUnless(aliases["un-corrupted1"].startswith("URI:DIR2:"))
 130         d.addCallback(_check_not_corrupted1)
 131
 132         def _remove_trailing_newline_and_add_alias(ign):
 133             # same thing, but for "tahoe add-alias"
 134             old = fileutil.read(aliasfile)
 135             fileutil.write(aliasfile, old.rstrip())
 136             return self.do_cli("add-alias", "un-corrupted2", self.two_uri)
 137         d.addCallback(_remove_trailing_newline_and_add_alias)
 138         def _check_not_corrupted((rc,stdout,stderr)):
 139             self.failUnless("Alias 'un-corrupted2' added" in stdout, stdout)
 140             self.failIf(stderr)
 141             aliases = get_aliases(self.get_clientdir())
 142             self.failUnless("un-corrupted1" in aliases)
 143             self.failUnless(aliases["un-corrupted1"].startswith("URI:DIR2:"))
 144             self.failIfIn("un-corrupted2:", aliases["un-corrupted1"])
 145             self.failUnless("un-corrupted2" in aliases)
 146             self.failUnless(aliases["un-corrupted2"].startswith("URI:DIR2:"))
 147         d.addCallback(_check_not_corrupted)
 148
 149     def test_create_unicode(self):
 150         self.basedir = "cli/CreateAlias/create_unicode"
 151         self.set_up_grid()
 152
 153         try:
 154             etudes_arg = u"\u00E9tudes".encode(get_io_encoding())
 155             lumiere_arg = u"lumi\u00E8re.txt".encode(get_io_encoding())
 156         except UnicodeEncodeError:
 157             raise unittest.SkipTest("A non-ASCII command argument could not be encoded on this platform.")
 158
 159         d = self.do_cli("create-alias", etudes_arg)
 160         def _check_create_unicode((rc, out, err)):
 161             self.failUnlessReallyEqual(rc, 0)
 162             self.failUnlessReallyEqual(err, "")
 163             self.failUnlessIn("Alias %s created" % quote_output(u"\u00E9tudes"), out)
 164
 165             aliases = get_aliases(self.get_clientdir())
 166             self.failUnless(aliases[u"\u00E9tudes"].startswith("URI:DIR2:"))
 167         d.addCallback(_check_create_unicode)
 168
 169         d.addCallback(lambda res: self.do_cli("ls", etudes_arg + ":"))
 170         def _check_ls1((rc, out, err)):
 171             self.failUnlessReallyEqual(rc, 0)
 172             self.failUnlessReallyEqual(err, "")
 173             self.failUnlessReallyEqual(out, "")
 174         d.addCallback(_check_ls1)
 175
 176         d.addCallback(lambda res: self.do_cli("put", "-", etudes_arg + ":uploaded.txt",
 177                                               stdin="Blah blah blah"))
 178
 179         d.addCallback(lambda res: self.do_cli("ls", etudes_arg + ":"))
 180         def _check_ls2((rc, out, err)):
 181             self.failUnlessReallyEqual(rc, 0)
 182             self.failUnlessReallyEqual(err, "")
 183             self.failUnlessReallyEqual(out, "uploaded.txt\n")
 184         d.addCallback(_check_ls2)
 185
 186         d.addCallback(lambda res: self.do_cli("get", etudes_arg + ":uploaded.txt"))
 187         def _check_get((rc, out, err)):
 188             self.failUnlessReallyEqual(rc, 0)
 189             self.failUnlessReallyEqual(err, "")
 190             self.failUnlessReallyEqual(out, "Blah blah blah")
 191         d.addCallback(_check_get)
 192
 193         # Ensure that an Unicode filename in an Unicode alias works as expected
 194         d.addCallback(lambda res: self.do_cli("put", "-", etudes_arg + ":" + lumiere_arg,
 195                                               stdin="Let the sunshine In!"))
 196
 197         d.addCallback(lambda res: self.do_cli("get",
 198                                               get_aliases(self.get_clientdir())[u"\u00E9tudes"] + "/" + lumiere_arg))
 199         def _check_get2((rc, out, err)):
 200             self.failUnlessReallyEqual(rc, 0)
 201             self.failUnlessReallyEqual(err, "")
 202             self.failUnlessReallyEqual(out, "Let the sunshine In!")
 203         d.addCallback(_check_get2)
 204
 205         return d
 206
 207     # TODO: test list-aliases, including Unicode
 208
 209