diff options
Diffstat (limited to 'libbe/command')
| -rw-r--r-- | libbe/command/assign.py | 11 | ||||
| -rw-r--r-- | libbe/command/base.py | 105 | ||||
| -rw-r--r-- | libbe/command/comment.py | 205 | ||||
| -rw-r--r-- | libbe/command/commit.py | 134 | ||||
| -rw-r--r-- | libbe/command/html.py | 12 | ||||
| -rw-r--r-- | libbe/command/init.py | 8 | ||||
| -rw-r--r-- | libbe/command/util.py | 18 |
7 files changed, 283 insertions, 210 deletions
diff --git a/libbe/command/assign.py b/libbe/command/assign.py index 6d1233c..f5d2df4 100644 --- a/libbe/command/assign.py +++ b/libbe/command/assign.py @@ -75,15 +75,8 @@ class Assign (libbe.command.Command): elif assignee == '-': assignee = params['user-id'] for bug_id in params['bug-id']: - p = libbe.util.id.parse_user(bugdir, bug_id) - if p['type'] != 'bug': - raise libbe.command.UserError( - '%s is a %s id, not a bug id' % (bug_id, p['type'])) - if p['bugdir'] != bugdir.uuid: - raise libbe.command.UserError( - "%s doesn't belong to this bugdir (%s)" - % (bug_id, bugdir.uuid)) - bug = bugdir.bug_from_uuid(p['bug']) + bug,dummy_comment = \ + libbe.command.util.bug_comment_from_user_id(bugdir, bug_id) if bug.assigned != assignee: bug.assigned = assignee diff --git a/libbe/command/base.py b/libbe/command/base.py index b27e188..e28cf33 100644 --- a/libbe/command/base.py +++ b/libbe/command/base.py @@ -2,6 +2,7 @@ import codecs import optparse +import os.path import sys import libbe @@ -9,7 +10,6 @@ import libbe.ui.util.user import libbe.util.encoding import libbe.util.plugin - class UserError(Exception): pass @@ -90,34 +90,55 @@ class Option (CommandInput): def __repr__(self): return '<Option %s>' % self.__str__() -class _DummyParser (object): - def __init__(self, options): - self.option_list = options - self.option_groups = [] - for option in self.option_list: # add required methods and attributes - option.dest = option.name - option._short_opts = [] - if option.short_name != None: - option._short_opts.append('-' + option.short_name) - option._long_opts = ['--' + option.name] - option.takes_value = lambda : option.arg != None - if option.takes_value(): - option.metavar = option.arg.metavar - else: - option.metavar = None +class _DummyParser (optparse.OptionParser): + def __init__(self, command): + optparse.OptionParser.__init__(self) + self.remove_option('-h') + self.command = command + self._command_opts = [] + for option in self.command.options: + self._add_option(option) + + def _add_option(self, option): + # from libbe.ui.command_line.CmdOptionParser._add_option + option.validate() + long_opt = '--%s' % option.name + if option.short_name != None: + short_opt = '-%s' % option.short_name + assert '_' not in option.name, \ + 'Non-reconstructable option name %s' % option.name + kwargs = {'dest':option.name.replace('-', '_'), + 'help':option.help} + if option.arg == None or option.arg.type == 'bool': + kwargs['action'] = 'store_true' + kwargs['metavar'] = None + kwargs['default'] = False + else: + kwargs['type'] = option.arg.type + kwargs['action'] = 'store' + kwargs['metavar'] = option.arg.metavar + kwargs['default'] = option.arg.default + if option.short_name != None: + opt = optparse.Option(short_opt, long_opt, **kwargs) + else: + opt = optparse.Option(long_opt, **kwargs) + #option.takes_value = lambda : option.arg != None + opt._option = option + self._command_opts.append(opt) + self.add_option(opt) class OptionFormatter (optparse.IndentedHelpFormatter): - def __init__(self, options): + def __init__(self, command): optparse.IndentedHelpFormatter.__init__(self) - self.options = options + self.command = command def option_help(self): # based on optparse.OptionParser.format_option_help() - parser = _DummyParser(self.options) + parser = _DummyParser(self.command) self.store_option_strings(parser) ret = [] ret.append(self.format_heading('Options')) self.indent() - for option in self.options: + for option in parser._command_opts: ret.append(self.format_option(option)) ret.append('\n') self.dedent() @@ -132,11 +153,11 @@ class Command (object): usage: be command [options] <BLANKLINE> Options: - -h HELP, --help=HELP Print a help message. + -h, --help Print a help message. <BLANKLINE> - --complete=STRING Print a list of possible completions. + --complete Print a list of possible completions. <BLANKLINE> - A detailed help message. + A detailed help message. """ name = 'command' @@ -145,7 +166,9 @@ class Command (object): self.status = None self.result = None self.requires_bugdir = False + self.requires_storage = False self.requires_unconnected_storage = False + self.restrict_file_access = True self.input_encoding = None self.output_encoding = None self.options = [ @@ -250,7 +273,7 @@ class Command (object): return usage def _option_help(self): - o = OptionFormatter(self.options) + o = OptionFormatter(self) return o.option_help().strip('\n') def _long_help(self): @@ -266,3 +289,37 @@ class Command (object): # finish a particular argument return argument.completion_callback(self, argument, fragment) return [] # the particular argument doesn't supply completion info + + def check_restricted_access(self, storage, path): + """ + Check that the file at path is inside bugdir.root. This is + important if you allow other users to execute becommands with + your username (e.g. if you're running be-handle-mail through + your ~/.procmailrc). If this check wasn't made, a user could + e.g. run + be commit -b ~/.ssh/id_rsa "Hack to expose ssh key" + which would expose your ssh key to anyone who could read the + VCS log. + + >>> class DummyStorage (object): pass + >>> s = DummyStorage() + >>> s.repo = os.path.expanduser('~/x/') + >>> c = Command() + >>> try: + ... c.check_restricted_access(s, os.path.expanduser('~/.ssh/id_rsa')) + ... except UserError, e: + ... assert str(e).startswith('file access restricted!'), str(e) + ... print 'we got the expected error' + we got the expected error + >>> c.check_restricted_access(s, os.path.expanduser('~/x')) + >>> c.check_restricted_access(s, os.path.expanduser('~/x/y')) + >>> c.restrict_file_access = False + >>> c.check_restricted_access(s, os.path.expanduser('~/.ssh/id_rsa')) + """ + if self.restrict_file_access == True: + path = os.path.abspath(path) + repo = os.path.abspath(storage.repo).rstrip(os.path.sep) + if path == repo or path.startswith(repo+os.path.sep): + return + raise UserError('file access restricted!\n %s not in %s' + % (path, repo)) diff --git a/libbe/command/comment.py b/libbe/command/comment.py index 9919d1d..7a8f5f9 100644 --- a/libbe/command/comment.py +++ b/libbe/command/comment.py @@ -15,43 +15,56 @@ # You should have received a copy of the GNU General Public License along # with this program; if not, write to the Free Software Foundation, Inc., # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. -"""Add a comment to a bug""" -from libbe import cmdutil, bugdir, comment, editor + import os import sys -__desc__ = __doc__ -def execute(args, manipulate_encodings=True, restrict_file_access=False, - dir="."): - """ +import libbe +import libbe.command +import libbe.command.util +import libbe.comment +import libbe.ui.util.editor +import libbe.util.id + + +class Comment (libbe.command.Command): + """Add a comment to a bug + >>> import time - >>> bd = bugdir.SimpleBugDir() - >>> os.chdir(bd.root) - >>> execute(["a", "This is a comment about a"], manipulate_encodings=False) - >>> bd._clear_bugs() - >>> bug = cmdutil.bug_from_id(bd, "a") + >>> import libbe.bugdir + >>> bd = libbe.bugdir.SimpleBugDir(memory=False) + >>> cmd = Comment() + >>> cmd._setup_io = lambda i_enc,o_enc : None + >>> cmd.stdout = sys.stdout + + >>> cmd.run(bd.storage, bd, {'user-id':u'Fran\\xe7ois'}, + ... ['/a', 'This is a comment about a']) + >>> bd.flush_reload() + >>> bug = bd.bug_from_uuid('a') >>> bug.load_comments(load_full=False) >>> comment = bug.comment_root[0] + >>> comment.id.storage() == comment.uuid + True >>> print comment.body This is a comment about a <BLANKLINE> - >>> comment.author == bd.user_id - True + >>> comment.author + u'Fran\\xe7ois' >>> comment.time <= int(time.time()) True >>> comment.in_reply_to is None True >>> if 'EDITOR' in os.environ: - ... del os.environ["EDITOR"] - >>> execute(["b"], manipulate_encodings=False) + ... del os.environ['EDITOR'] + >>> cmd.run(bd.storage, bd, {'user-id':u'Frank'}, ['/b']) Traceback (most recent call last): UserError: No comment supplied, and EDITOR not specified. - >>> os.environ["EDITOR"] = "echo 'I like cheese' > " - >>> execute(["b"], manipulate_encodings=False) - >>> bd._clear_bugs() - >>> bug = cmdutil.bug_from_id(bd, "b") + >>> os.environ['EDITOR'] = "echo 'I like cheese' > " + >>> cmd.run(bd.storage, bd, {'user-id':u'Frank'}, ['/b']) + >>> bd.flush_reload() + >>> bug = bd.bug_from_uuid('b') >>> bug.load_comments(load_full=False) >>> comment = bug.comment_root[0] >>> print comment.body @@ -59,65 +72,75 @@ def execute(args, manipulate_encodings=True, restrict_file_access=False, <BLANKLINE> >>> bd.cleanup() """ - parser = get_parser() - options, args = parser.parse_args(args) - complete(options, args, parser) - if len(args) == 0: - raise cmdutil.UsageError("Please specify a bug or comment id.") - if len(args) > 2: - raise cmdutil.UsageError("Too many arguments.") + name = 'comment' - shortname = args[0] - - bd = bugdir.BugDir(from_disk=True, - manipulate_encodings=manipulate_encodings, - root=dir) - bug, parent = cmdutil.bug_comment_from_id(bd, shortname) - - if len(args) == 1: # try to launch an editor for comment-body entry - try: - if parent == bug.comment_root: - parent_body = bug.summary+"\n" - else: - parent_body = parent.body - estr = "Please enter your comment above\n\n> %s\n" \ - % ("\n> ".join(parent_body.splitlines())) - body = editor.editor_string(estr) - except editor.CantFindEditor, e: - raise cmdutil.UserError, "No comment supplied, and EDITOR not specified." - if body is None: - raise cmdutil.UserError("No comment entered.") - elif args[1] == '-': # read body from stdin - binary = not (options.content_type == None - or options.content_type.startswith("text/")) - if not binary: - body = sys.stdin.read() + def __init__(self, *args, **kwargs): + libbe.command.Command.__init__(self, *args, **kwargs) + self.requires_bugdir = True + self.options.extend([ + libbe.command.Option(name='author', short_name='a', + help='Set the comment author', + arg=libbe.command.Argument( + name='author', metavar='AUTHOR')), + libbe.command.Option(name='alt-id', + help='Set an alternate comment ID', + arg=libbe.command.Argument( + name='alt-id', metavar='ID')), + libbe.command.Option(name='content-type', short_name='c', + help='Set comment content-type (e.g. text/plain)', + arg=libbe.command.Argument(name='content-type', + metavar='MIME')), + ]) + self.args.extend([ + libbe.command.Argument( + name='id', metavar='ID', default=None, + completion_callback=libbe.command.util.complete_bug_comment_id), + libbe.command.Argument( + name='comment', metavar='COMMENT', default=None, + optional=True, + completion_callback=libbe.command.util.complete_assigned), + ]) + def _run(self, storage, bugdir, **params): + bug,parent = \ + libbe.command.util.bug_comment_from_user_id(bugdir, params['id']) + if params['comment'] == None: + # try to launch an editor for comment-body entry + try: + if parent == bug.comment_root: + parent_body = bug.summary+'\n' + else: + parent_body = parent.body + estr = 'Please enter your comment above\n\n> %s\n' \ + % ('\n> '.join(parent_body.splitlines())) + body = libbe.ui.util.editor.editor_string(estr) + except libbe.ui.util.editor.CantFindEditor, e: + raise libbe.command.UserError( + 'No comment supplied, and EDITOR not specified.') + if body is None: + raise libbe.command.UserError('No comment entered.') + elif params['comment'] == '-': # read body from stdin + binary = not (params['content-type'] == None + or params['content-type'].startswith("text/")) + if not binary: + body = self.stdin.read() + if not body.endswith('\n'): + body += '\n' + else: # read-in without decoding + body = sys.stdin.read() + else: # body given on command line + body = params['comment'] if not body.endswith('\n'): body+='\n' - else: # read-in without decoding - body = sys.__stdin__.read() - else: # body = arg[1] - body = args[1] - if not body.endswith('\n'): - body+='\n' - - new = parent.new_reply(body=body, content_type=options.content_type) - if options.author != None: - new.author = options.author - if options.alt_id != None: - new.alt_id = options.alt_id + if params['author'] == None: + params['author'] = params['user-id'] -def get_parser(): - parser = cmdutil.CmdOptionParser("be comment ID [COMMENT]") - parser.add_option("-a", "--author", metavar="AUTHOR", dest="author", - help="Set the comment author", default=None) - parser.add_option("--alt-id", metavar="ID", dest="alt_id", - help="Set an alternate comment ID", default=None) - parser.add_option("-c", "--content-type", metavar="MIME", dest="content_type", - help="Set comment content-type (e.g. text/plain)", default=None) - return parser + new = parent.new_reply(body=body) + for key in ['alt-id', 'author', 'content-type']: + if params[key] != None: + setattr(new, key, params[key]) -longhelp=""" + def _long_help(self): + return """ To add a comment to a bug, use the bug ID as the argument. To reply to another comment, specify the comment name (as shown in "be show" output). COMMENT, if specified, should be either the text of your @@ -126,39 +149,3 @@ you do not specify a COMMENT, $EDITOR is used to launch an editor. If COMMENT is unspecified and EDITOR is not set, no comment will be created. """ - -def help(): - return get_parser().help_str() + longhelp - -def complete(options, args, parser): - for option,value in cmdutil.option_value_pairs(options, parser): - if value == "--complete": - # no argument-options at the moment, so this is future-proofing - raise cmdutil.GetCompletions() - for pos,value in enumerate(args): - if value == "--complete": - if pos == 0: # fist positional argument is a bug or comment id - if len(args) >= 2: - partial = args[1].split(':')[0] # take only bugid portion - else: - partial = "" - ids = [] - try: - bd = bugdir.BugDir(from_disk=True, - manipulate_encodings=False) - bugs = [] - for uuid in bd.uuids(): - if uuid.startswith(partial): - bug = bd.bug_from_uuid(uuid) - if bug.active == True: - bugs.append(bug) - for bug in bugs: - shortname = bd.bug_shortname(bug) - ids.append(shortname) - bug.load_comments(load_full=False) - for id,comment in bug.comment_shortnames(shortname): - ids.append(id) - except bugdir.NoBugDir: - pass - raise cmdutil.GetCompletions(ids) - raise cmdutil.GetCompletions() diff --git a/libbe/command/commit.py b/libbe/command/commit.py index cade355..f795e80 100644 --- a/libbe/command/commit.py +++ b/libbe/command/commit.py @@ -13,70 +13,90 @@ # You should have received a copy of the GNU General Public License along # with this program; if not, write to the Free Software Foundation, Inc., # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. -"""Commit the currently pending changes to the repository""" -from libbe import cmdutil, bugdir, editor, vcs + import sys -__desc__ = __doc__ -def execute(args, manipulate_encodings=True, restrict_file_access=False, - dir="."): - """ - >>> import os - >>> from libbe import bug - >>> bd = bugdir.SimpleBugDir() - >>> os.chdir(bd.root) - >>> full_path = "testfile" - >>> test_contents = "A test file" - >>> bd.vcs.set_file_contents(full_path, test_contents) - >>> execute(["Added %s." % (full_path)], manipulate_encodings=False) # doctest: +ELLIPSIS +import libbe +import libbe.bugdir +import libbe.command +import libbe.command.util +import libbe.storage +import libbe.ui.util.editor + + +class Commit (libbe.command.Command): + """Commit the currently pending changes to the repository + + >>> import os, sys + >>> import libbe.storage.vcs + >>> import libbe.storage.vcs.base + >>> import libbe.util.utility + >>> cmd = Commit() + >>> cmd._setup_io = lambda i_enc,o_enc : None + >>> cmd.stdout = sys.stdout + + >>> dir = libbe.util.utility.Dir() + >>> vcs = libbe.storage.vcs.installed_vcs() + >>> vcs.repo = dir.path + >>> vcs.init() + >>> vcs.connect() + >>> if vcs.name in libbe.storage.vcs.base.VCS_ORDER: + ... bd = libbe.bugdir.BugDir(vcs, from_storage=False) + ... bd.extra_strings = ['hi there'] + ... cmd.run(vcs, None, {'user-id':'Joe'}, + ... ['Making a commit']) # doctest: +ELLIPSIS + ... else: + ... print 'Committed ...' Committed ... - >>> bd.cleanup() + >>> vcs.disconnect() + >>> vcs.destroy() + >>> dir.cleanup() """ - parser = get_parser() - options, args = parser.parse_args(args) - cmdutil.default_complete(options, args, parser) - if len(args) != 1: - raise cmdutil.UsageError("Please supply a commit message") - bd = bugdir.BugDir(from_disk=True, - manipulate_encodings=manipulate_encodings, - root=dir) - if args[0] == '-': # read summary from stdin - assert options.body != "EDITOR", \ - "Cannot spawn and editor when the summary is using stdin." - summary = sys.stdin.readline() - else: - summary = args[0] - if options.body == None: - body = None - elif options.body == "EDITOR": - body = editor.editor_string("Please enter your commit message above") - else: - if restrict_file_access == True: - cmdutil.restrict_file_access(bd, options.body) - body = bd.vcs.get_file_contents(options.body, allow_no_vcs=True) - try: - revision = bd.vcs.commit(summary, body=body, - allow_empty=options.allow_empty) - except vcs.EmptyCommit, e: - print e - return 1 - else: - print "Committed %s" % revision + name = 'commit' -def get_parser(): - parser = cmdutil.CmdOptionParser("be commit COMMENT") - parser.add_option("-b", "--body", metavar="FILE", dest="body", - help='Provide a detailed body for the commit message. In the special case that FILE == "EDITOR", spawn an editor to enter the body text (in which case you cannot use stdin for the summary)', default=None) - parser.add_option("-a", "--allow-empty", dest="allow_empty", - help="Allow empty commits", - default=False, action="store_true") - return parser + def __init__(self, *args, **kwargs): + libbe.command.Command.__init__(self, *args, **kwargs) + self.requires_storage = True + self.options.extend([ + libbe.command.Option(name='body', short_name='b', + help='Provide the detailed body for the commit message. In the special case that FILE == "EDITOR", spawn an editor to enter the body text (in which case you cannot use stdin for the summary)', + arg=libbe.command.Argument(name='body', metavar='FILE', + completion_callback=libbe.command.util.complete_path)), + libbe.command.Option(name='allow-empty', short_name='a', + help='Allow empty commits'), + ]) + self.args.extend([ + libbe.command.Argument( + name='comment', metavar='COMMENT', default=None), + ]) -longhelp=""" + def _run(self, storage, bugdir=None, **params): + if params['comment'] == '-': # read summary from stdin + assert params['body'] != 'EDITOR', \ + 'Cannot spawn and editor when the summary is using stdin.' + summary = sys.stdin.readline() + else: + summary = params['comment'] + if params['body'] == None: + body = None + elif params['body'] == 'EDITOR': + body = libbe.ui.util.editor.editor_string( + 'Please enter your commit message above') + else: + self.check_restricted_access(storage, params['body']) + body = libbe.util.encoding.get_file_contents( + params['body'], decode=True) + try: + revision = storage.commit(summary, body=body, + allow_empty=params['allow-empty']) + print >> self.stdout, 'Committed %s' % revision + except libbe.storage.EmptyCommit, e: + print >> self.stdout, e + return 1 + + def _long_help(self): + return """ Commit the current repository status. The summary specified on the commandline is a string (only one line) that describes the commit briefly or "-", in which case the string will be read from stdin. """ - -def help(): - return get_parser().help_str() + longhelp diff --git a/libbe/command/html.py b/libbe/command/html.py index d9e0d73..487e97c 100644 --- a/libbe/command/html.py +++ b/libbe/command/html.py @@ -16,6 +16,7 @@ # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. """Generate a static HTML dump of the current repository status""" from libbe import cmdutil, bugdir, bug +import libbe.util.encoding import codecs, os, os.path, re, string, time import xml.sax.saxutils, htmlentitydefs @@ -312,15 +313,12 @@ class HTMLGen (object): return dir_path def _write_file(self, content, path_array, mode='w'): - f = codecs.open(os.path.join(*path_array), mode, self.encoding) - f.write(content) - f.close() + return libbe.util.encoding.set_file_contents( + os.path.join(*path_array), content, mode, self.encoding) def _read_file(self, path_array, mode='r'): - f = codecs.open(os.path.join(*path_array), mode, self.encoding) - content = f.read() - f.close() - return content + return libbe.util.encoding.get_file_contents( + os.path.join(*path_array), mode, self.encoding, decode=True) def write_default_template(self, out_dir): if self.verbose: diff --git a/libbe/command/init.py b/libbe/command/init.py index cdaa149..017cdc3 100644 --- a/libbe/command/init.py +++ b/libbe/command/init.py @@ -21,7 +21,6 @@ import os.path import libbe import libbe.bugdir import libbe.command -import libbe.command.util import libbe.storage class Init (libbe.command.Command): @@ -53,11 +52,13 @@ class Init (libbe.command.Command): >>> dir = libbe.util.utility.Dir() >>> vcs = libbe.storage.vcs.installed_vcs() + >>> vcs.repo = dir.path + >>> vcs._vcs_init(vcs.repo) >>> if vcs.name in libbe.storage.vcs.base.VCS_ORDER: - ... vcs.repo = dir.path - ... vcs._vcs_init(vcs.repo) ... cmd.run(vcs) # doctest: +ELLIPSIS ... else: + ... vcs.init() + ... vcs.connect() ... print 'Using ... for revision control.\\nDirectory initialized.' Using ... for revision control. BE repository initialized. @@ -65,7 +66,6 @@ class Init (libbe.command.Command): >>> vcs.destroy() >>> dir.cleanup() """ - name = 'init' def __init__(self, *args, **kwargs): diff --git a/libbe/command/util.py b/libbe/command/util.py index 98b2081..4395592 100644 --- a/libbe/command/util.py +++ b/libbe/command/util.py @@ -43,6 +43,8 @@ def complete_extra_strings(command, argument, fragment=None): return [fragment] def complete_bug_id(command, argument, fragment=None): return [fragment] +def complete_bug_comment_id(command, argument, fragment=None): + return [fragment] def select_values(string, possible_values, name="unkown"): """ @@ -96,3 +98,19 @@ def select_values(string, possible_values, name="unkown"): % (name, value, possible_values)) possible_values = whitelisted_values return possible_values + +def bug_comment_from_user_id(bugdir, id): + p = libbe.util.id.parse_user(bugdir, id) + if not p['type'] in ['bug', 'comment']: + raise libbe.command.UserError( + '%s is a %s id, not a bug or comment id' % (id, p['type'])) + if p['bugdir'] != bugdir.uuid: + raise libbe.command.UserError( + "%s doesn't belong to this bugdir (%s)" + % (id, bugdir.uuid)) + bug = bugdir.bug_from_uuid(p['bug']) + if 'comment' in p: + comment = bug.comment_from_uuid(p['comment']) + else: + comment = bug.comment_root + return (bug, comment) |