-import os, sys, urllib
+import os, sys, urllib, textwrap
+import codecs
from twisted.python import usage
+from allmydata.util.assertutil import precondition
+from allmydata.util.encodingutil import unicode_to_url, quote_output, \
+ quote_local_unicode_path, argv_to_abspath
+from allmydata.scripts.default_nodedir import _default_nodedir
+
+def get_default_nodedir():
+ return _default_nodedir
+
+def wrap_paragraphs(text, width):
+ # like textwrap.wrap(), but preserve paragraphs (delimited by double
+ # newlines) and leading whitespace, and remove internal whitespace.
+ text = textwrap.dedent(text)
+ if text.startswith("\n"):
+ text = text[1:]
+ return "\n\n".join([textwrap.fill(paragraph, width=width)
+ for paragraph in text.split("\n\n")])
+
+class BaseOptions(usage.Options):
+ def __init__(self):
+ super(BaseOptions, self).__init__()
+ self.command_name = os.path.basename(sys.argv[0])
+ if self.command_name == 'trial':
+ self.command_name = 'tahoe'
+
+ # Only allow "tahoe --version", not e.g. "tahoe start --version"
+ def opt_version(self):
+ raise usage.UsageError("--version not allowed on subcommands")
+
+ description = None
+ description_unwrapped = None
+
+ def __str__(self):
+ width = int(os.environ.get('COLUMNS', '80'))
+ s = (self.getSynopsis() + '\n' +
+ "(use 'tahoe --help' to view global options)\n" +
+ '\n' +
+ self.getUsage())
+ if self.description:
+ s += '\n' + wrap_paragraphs(self.description, width) + '\n'
+ if self.description_unwrapped:
+ du = textwrap.dedent(self.description_unwrapped)
+ if du.startswith("\n"):
+ du = du[1:]
+ s += '\n' + du + '\n'
+ return s
+
+class BasedirOptions(BaseOptions):
+ default_nodedir = _default_nodedir
+
+ optParameters = [
+ ["basedir", "C", None, "Specify which Tahoe base directory should be used. [default: %s]"
+ % quote_local_unicode_path(_default_nodedir)],
+ ]
+
+ def parseArgs(self, basedir=None):
+ # This finds the node-directory option correctly even if we are in a subcommand.
+ root = self.parent
+ while root.parent is not None:
+ root = root.parent
+
+ if root['node-directory'] and self['basedir']:
+ raise usage.UsageError("The --node-directory (or -d) and --basedir (or -C) options cannot both be used.")
+ if root['node-directory'] and basedir:
+ raise usage.UsageError("The --node-directory (or -d) option and a basedir argument cannot both be used.")
+ if self['basedir'] and basedir:
+ raise usage.UsageError("The --basedir (or -C) option and a basedir argument cannot both be used.")
+
+ if basedir:
+ b = argv_to_abspath(basedir)
+ elif self['basedir']:
+ b = argv_to_abspath(self['basedir'])
+ elif root['node-directory']:
+ b = argv_to_abspath(root['node-directory'])
+ elif self.default_nodedir:
+ b = self.default_nodedir
+ else:
+ raise usage.UsageError("No default basedir available, you must provide one with --node-directory, --basedir, or a basedir argument")
+ self['basedir'] = b
+ self['node-directory'] = b
+ def postOptions(self):
+ if not self['basedir']:
+ raise usage.UsageError("A base directory for the node must be provided.")
-class BaseOptions:
- # unit tests can override these to point at StringIO instances
- stdin = sys.stdin
- stdout = sys.stdout
- stderr = sys.stderr
-
- optFlags = [
- ["quiet", "q", "Operate silently."],
- ["version", "V", "Display version numbers and exit."],
- ["version-and-path", None, "Display version numbers and paths to their locations and exit."],
- ]
-
- def opt_version(self):
- import allmydata
- print allmydata.get_package_versions_string()
- sys.exit(0)
+class NoDefaultBasedirOptions(BasedirOptions):
+ default_nodedir = None
- def opt_version_and_path(self):
- import allmydata
- print allmydata.get_package_versions_string(show_paths=True)
- sys.exit(0)
+ optParameters = [
+ ["basedir", "C", None, "Specify which Tahoe base directory should be used."],
+ ]
+ # This is overridden in order to ensure we get a "Wrong number of arguments."
+ # error when more than one argument is given.
+ def parseArgs(self, basedir=None):
+ BasedirOptions.parseArgs(self, basedir)
-class BasedirMixin:
- optFlags = [
- ["multiple", "m", "allow multiple basedirs to be specified at once"],
- ]
+ def getSynopsis(self):
+ return "Usage: %s [global-options] %s [options] NODEDIR" % (self.command_name, self.subcommand_name)
- def postOptions(self):
- if not self.basedirs:
- raise usage.UsageError("<basedir> parameter is required")
- if self['basedir']:
- del self['basedir']
- self['basedirs'] = [os.path.abspath(os.path.expanduser(b)) for b in self.basedirs]
-
- def parseArgs(self, *args):
- from allmydata.util.assertutil import precondition
- self.basedirs = []
- if self['basedir']:
- precondition(isinstance(self['basedir'], (str, unicode)), self['basedir'])
- self.basedirs.append(self['basedir'])
- if self['multiple']:
- precondition(not [x for x in args if not isinstance(x, (str, unicode))], args)
- self.basedirs.extend(args)
- else:
- if len(args) == 0 and not self.basedirs:
- if sys.platform == 'win32':
- from allmydata.windows import registry
- rbdp = registry.get_base_dir_path()
- if rbdp:
- precondition(isinstance(registry.get_base_dir_path(), (str, unicode)), registry.get_base_dir_path())
- self.basedirs.append(rbdp)
- else:
- precondition(isinstance(os.path.expanduser("~/.tahoe"), (str, unicode)), os.path.expanduser("~/.tahoe"))
- self.basedirs.append(os.path.expanduser("~/.tahoe"))
- if len(args) > 0:
- precondition(isinstance(args[0], (str, unicode)), args[0])
- self.basedirs.append(args[0])
- if len(args) > 1:
- raise usage.UsageError("I wasn't expecting so many arguments")
-
-class NoDefaultBasedirMixin(BasedirMixin):
- def parseArgs(self, *args):
- from allmydata.util.assertutil import precondition
- # create-client won't default to --basedir=~/.tahoe
- self.basedirs = []
- if self['basedir']:
- precondition(isinstance(self['basedir'], (str, unicode)), self['basedir'])
- self.basedirs.append(self['basedir'])
- if self['multiple']:
- precondition(not [x for x in args if not isinstance(x, (str, unicode))], args)
- self.basedirs.extend(args)
- else:
- if len(args) > 0:
- precondition(isinstance(args[0], (str, unicode)), args[0])
- self.basedirs.append(args[0])
- if len(args) > 1:
- raise usage.UsageError("I wasn't expecting so many arguments")
- if not self.basedirs:
- raise usage.UsageError("--basedir must be provided")
-DEFAULT_ALIAS = "tahoe"
+DEFAULT_ALIAS = u"tahoe"
def get_aliases(nodedir):
- from allmydata import uri
aliases = {}
aliasfile = os.path.join(nodedir, "private", "aliases")
rootfile = os.path.join(nodedir, "private", "root_dir.cap")
f = open(rootfile, "r")
rootcap = f.read().strip()
if rootcap:
- aliases["tahoe"] = uri.from_string_dirnode(rootcap).to_string()
+ aliases[DEFAULT_ALIAS] = rootcap
except EnvironmentError:
pass
try:
- f = open(aliasfile, "r")
+ f = codecs.open(aliasfile, "r", "utf-8")
for line in f.readlines():
line = line.strip()
if line.startswith("#") or not line:
continue
- name, cap = line.split(":", 1)
+ name, cap = line.split(u":", 1)
# normalize it: remove http: prefix, urldecode
- cap = cap.strip()
- aliases[name] = uri.from_string_dirnode(cap).to_string()
+ cap = cap.strip().encode('utf-8')
+ aliases[name] = cap
except EnvironmentError:
pass
return aliases
return True
return False
-def get_alias(aliases, path, default):
- # transform "work:path/filename" into (aliases["work"], "path/filename").
- # If default=None, then an empty alias is indicated by returning
- # DefaultAliasMarker. We special-case "URI:" to make it easy to access
- # specific files/directories by their read-cap.
- path = path.strip()
- if path.startswith("URI:"):
- # The only way to get a sub-path is to use URI:blah:./foo, and we
- # strip out the :./ sequence.
+
+class TahoeError(Exception):
+ def __init__(self, msg):
+ Exception.__init__(self, msg)
+ self.msg = msg
+
+ def display(self, err):
+ print >>err, self.msg
+
+
+class UnknownAliasError(TahoeError):
+ def __init__(self, msg):
+ TahoeError.__init__(self, "error: " + msg)
+
+
+def get_alias(aliases, path_unicode, default):
+ """
+ Transform u"work:path/filename" into (aliases[u"work"], u"path/filename".encode('utf-8')).
+ If default=None, then an empty alias is indicated by returning
+ DefaultAliasMarker. We special-case strings with a recognized cap URI
+ prefix, to make it easy to access specific files/directories by their
+ caps.
+ If the transformed alias is either not found in aliases, or is blank
+ and default is not found in aliases, an UnknownAliasError is
+ raised.
+ """
+ precondition(isinstance(path_unicode, unicode), path_unicode)
+
+ from allmydata import uri
+ path = path_unicode.encode('utf-8').strip(" ")
+ if uri.has_uri_prefix(path):
+ # We used to require "URI:blah:./foo" in order to get a subpath,
+ # stripping out the ":./" sequence. We still allow that for compatibility,
+ # but now also allow just "URI:blah/foo".
sep = path.find(":./")
if sep != -1:
return path[:sep], path[sep+3:]
+ sep = path.find("/")
+ if sep != -1:
+ return path[:sep], path[sep+1:]
return path, ""
colon = path.find(":")
if colon == -1:
# no alias
if default == None:
return DefaultAliasMarker, path
- return aliases[default], path
- if colon == 1 and platform_uses_lettercolon_drivename():
+ if default not in aliases:
+ raise UnknownAliasError("No alias specified, and the default %s alias doesn't exist. "
+ "To create it, use 'tahoe create-alias %s'."
+ % (quote_output(default), quote_output(default, quotemarks=False)))
+ return uri.from_string_dirnode(aliases[default]).to_string(), path
+ if colon == 1 and default is None and platform_uses_lettercolon_drivename():
# treat C:\why\must\windows\be\so\weird as a local path, not a tahoe
# file in the "C:" alias
return DefaultAliasMarker, path
- alias = path[:colon]
- if "/" in alias:
+
+ # decoding must succeed because path is valid UTF-8 and colon & space are ASCII
+ alias = path[:colon].decode('utf-8')
+ if u"/" in alias:
# no alias, but there's a colon in a dirname/filename, like
# "foo/bar:7"
if default == None:
return DefaultAliasMarker, path
- return aliases[default], path
- return aliases[alias], path[colon+1:]
+ if default not in aliases:
+ raise UnknownAliasError("No alias specified, and the default %s alias doesn't exist. "
+ "To create it, use 'tahoe create-alias %s'."
+ % (quote_output(default), quote_output(default, quotemarks=False)))
+ return uri.from_string_dirnode(aliases[default]).to_string(), path
+ if alias not in aliases:
+ raise UnknownAliasError("Unknown alias %s, please create it with 'tahoe add-alias' or 'tahoe create-alias'." %
+ quote_output(alias))
+ return uri.from_string_dirnode(aliases[alias]).to_string(), path[colon+1:]
def escape_path(path):
+ # this always returns bytes, specifically US-ASCII, valid URL characters
segments = path.split("/")
- return "/".join([urllib.quote(s) for s in segments])
+ return "/".join([urllib.quote(unicode_to_url(s)) for s in segments])
projects / tahoe-lafs / tahoe-lafs.git / blobdiff