source: trunk/src/allmydata/test/cli/test_create_alias.py

Last change on this file was 53084f7, checked in by Alexandre Detiste <alexandre.detiste@…>, at 2024-02-27T23:49:07Z

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