settings_object.SavedSettingsObject.__init__(self)
self.storage = storage
self.id = libbe.util.id.ID(self, 'bugdir')
+ self.uuid = uuid
if from_storage == True:
- self.uuid = [c for c in self.storage.children()
- if c != 'version'][0]
+ if self.uuid == None:
+ self.uuid = [c for c in self.storage.children()
+ if c != 'version'][0]
self.load_settings()
else:
- if uuid == None:
+ if self.uuid == None:
self.uuid = libbe.util.id.uuid_gen()
self.settings = {}
self._setup_saved_settings()
else:
dir = utility.Dir()
self._dir_ref = dir # postpone cleanup since dir.cleanup() removes dir.
- storage = libbe.storage.base.Storage( \
- os.path.join(dir.path, 'repo.pkl'))
+ storage = libbe.storage.base.Storage(dir.path)
storage.init()
storage.connect()
- BugDir.__init__(self, storage=storage)
- bug_a = self.new_bug(summary="Bug A", _uuid="a")
- bug_a.creator = "John Doe <jdoe@example.com>"
+ BugDir.__init__(self, storage=storage, uuid='simple')
+ bug_a = self.new_bug(summary='Bug A', _uuid='a')
+ bug_a.creator = 'John Doe <jdoe@example.com>'
bug_a.time = 0
- bug_b = self.new_bug(summary="Bug B", _uuid="b")
- bug_b.creator = "Jane Doe <jdoe@example.com>"
+ bug_b = self.new_bug(summary='Bug B', _uuid='b')
+ bug_b.creator = 'Jane Doe <jdoe@example.com>'
bug_b.time = 0
- bug_b.status = "closed"
+ bug_b.status = 'closed'
if self.storage != None:
self.storage.disconnect() # flush to storage
self.storage.connect()
+
def cleanup(self):
if self.storage != None:
+ self.storage.writeable = True
self.storage.disconnect()
self.storage.destroy()
- if hasattr(self, "_dir_ref"):
+ if hasattr(self, '_dir_ref'):
self._dir_ref.cleanup()
+ def flush_reload(self):
+ if self.storage != None:
+ self.storage.disconnect()
+ self.storage.connect()
+ self._clear_bugs()
+
# class BugDirTestCase(unittest.TestCase):
# def setUp(self):
# self.dir = utility.Dir()
def setUp(self):
# create a pre-existing bugdir in a temporary directory
self.dir = utility.Dir()
- self.storage = libbe.storage.base.Storage( \
- os.path.join(self.dir.path, 'repo.pkl'))
+ self.storage = libbe.storage.base.Storage(self.dir.path)
self.storage.init()
self.storage.connect()
self.bugdir = BugDir(self.storage)
bugdir.storage.is_writeable())
uuids = sorted([bug.uuid for bug in bugdir])
self.failUnless(uuids == ['a', 'b'], uuids)
- self.storage.disconnect() # flush
- self.storage.connect()
- bugdir._clear_bugs()
+ bugdir.flush_reload()
uuids = sorted(bugdir.uuids())
self.failUnless(uuids == ['a', 'b'], uuids)
uuids = sorted([bug.uuid for bug in bugdir])
# 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.
-"""Assign an individual or group to fix a bug"""
-from libbe import cmdutil, bugdir
-__desc__ = __doc__
-def execute(args, manipulate_encodings=True, restrict_file_access=False,
- dir="."):
- """
- >>> import os
- >>> bd = bugdir.SimpleBugDir()
- >>> os.chdir(bd.root)
- >>> bd.bug_from_shortname("a").assigned is None
- True
+import libbe
+import libbe.command
+import libbe.command.util
+
+class Assign (libbe.command.Command):
+ """Assign an individual or group to fix a bug
+
+ >>> import os, sys
+ >>> import libbe.bugdir
+ >>> bd = libbe.bugdir.SimpleBugDir(memory=False)
+ >>> cmd = Assign()
+ >>> cmd._setup_io = lambda i_enc,o_enc : None
+ >>> cmd.stdout = sys.stdout
- >>> execute(["a"], manipulate_encodings=False)
- >>> bd._clear_bugs()
- >>> bd.bug_from_shortname("a").assigned == bd.user_id
+ >>> bd.bug_from_uuid('a').assigned is None
True
+ >>> cmd.run(bd, {'user-id':u'Fran\xe7ois'}, ['-', 'a'])
+ >>> bd.flush_reload()
+ >>> bd.bug_from_uuid('a').assigned
+ u'Fran\\xe7ois'
- >>> execute(["a", "someone"], manipulate_encodings=False)
- >>> bd._clear_bugs()
- >>> print bd.bug_from_shortname("a").assigned
- someone
+ >>> cmd.run(bd, args=['someone', 'a', 'b'])
+ >>> bd.flush_reload()
+ >>> bd.bug_from_uuid('a').assigned
+ 'someone'
+ >>> bd.bug_from_uuid('b').assigned
+ 'someone'
- >>> execute(["a","none"], manipulate_encodings=False)
- >>> bd._clear_bugs()
- >>> bd.bug_from_shortname("a").assigned is None
+ >>> cmd.run(bd, args=['none', 'a'])
+ >>> bd.flush_reload()
+ >>> bd.bug_from_uuid('a').assigned is None
True
>>> bd.cleanup()
"""
- parser = get_parser()
- options, args = parser.parse_args(args)
- cmdutil.default_complete(options, args, parser,
- bugid_args={0: lambda bug : bug.active==True})
- assert(len(args) in (0, 1, 2))
- if len(args) == 0:
- raise cmdutil.UsageError("Please specify a bug id.")
- if len(args) > 2:
- help()
- raise cmdutil.UsageError("Too many arguments.")
- bd = bugdir.BugDir(from_disk=True,
- manipulate_encodings=manipulate_encodings,
- root=dir)
- bug = cmdutil.bug_from_id(bd, args[0])
- bug = bd.bug_from_shortname(args[0])
- if len(args) == 1:
- bug.assigned = bd.user_id
- elif len(args) == 2:
- if args[1] == "none":
- bug.assigned = None
- else:
- bug.assigned = args[1]
- bd.save()
+
+ name = 'assign'
-def get_parser():
- parser = cmdutil.CmdOptionParser("be assign BUG-ID [ASSIGNEE]")
- return parser
+ def __init__(self, *args, **kwargs):
+ libbe.command.Command.__init__(self, *args, **kwargs)
+ self.requires_bugdir = True
+ self.args.extend([
+ libbe.command.Argument(
+ name='assignee', metavar='ASSIGNEE', default=None,
+ completion_callback=libbe.command.util.complete_assigned),
+ libbe.command.Argument(
+ name='bug-id', metavar='BUG-ID', default=None,
+ repeatable=True,
+ completion_callback=libbe.command.util.complete_bug_id),
+ ])
-longhelp = """
-Assign a person to fix a bug.
+ def _run(self, bugdir, **params):
+ assignee = params['assignee']
+ if assignee == 'none':
+ assignee = None
+ elif assignee == '-':
+ assignee = params['user-id']
+ for bug_id in params['bug-id']:
+ bug = bugdir.bug_from_uuid(bug_id)
+ if bug.assigned != assignee:
+ bug.assigned = assignee
-By default, the bug is self-assigned. If an assignee is specified, the bug
-will be assigned to that person.
+ def _long_help(self):
+ return """
+Assign a person to fix a bug.
-Assignees should be the person's Bugs Everywhere identity, the string that
-appears in Creator fields.
+Assignees should be the person's Bugs Everywhere identity, the same
+string that appears in Creator fields.
-To un-assign a bug, specify "none" for the assignee.
+Special assignee strings:
+ "-" assign the bug to yourself
+ "none" un-assigns the bug
"""
-
-def help():
- return get_parser().help_str() + longhelp
import sys
import libbe
+import libbe.ui.util.user
import libbe.util.encoding
import libbe.util.plugin
+
class UserError(Exception):
pass
args = []
params = {}
for option in self.options:
+ assert option.name not in params, params[option.name]
if option.name in options:
params[option.name] = options.pop(option.name)
elif option.arg != None:
params[option.name] = option.arg.default
else: # non-arg options are flags, set to default flag value
params[option.name] = False
+ assert 'user-id' not in params, params['user-id']
+ if 'user-id' in options:
+ params['user-id'] = options.pop('user-id')
+ else:
+ params['user-id'] = libbe.ui.util.user.get_user_id(bugdir.storage)
if len(options) > 0:
raise UserError, 'Invalid option passed to command %s:\n %s' \
% (self.name, '\n '.join(['%s: %s' % (k,v)
for k,v in options.items()]))
- for arg in self.args:
- pass
+ in_optional_args = False
+ for i,arg in enumerate(self.args):
+ if arg.repeatable == True:
+ assert i == len(self.args)-1, arg.name
+ if in_optional_args == True:
+ assert arg.optional == True, arg.name
+ else:
+ in_optional_args = arg.optional
+ if i < len(args):
+ if arg.repeatable == True:
+ params[arg.name] = [args[i]]
+ else:
+ params[arg.name] = args[i]
+ else: # no value given
+ assert in_optional_args == True, arg.name
+ if arg.repeatable == True:
+ params[arg.name] = [arg.default]
+ else:
+ params[arg.name] = arg.default
+ if len(args) > len(self.args): # add some additional repeats
+ assert self.args[-1].repeatable == True, self.args[-1].name
+ params[self.args[-1].name].extend(args[len(self.args):])
+
if params['help'] == True:
pass
else:
pass
else:
params.pop('complete')
+
self._setup_io(self.input_encoding, self.output_encoding)
self.status = self._run(bugdir, **params)
return self.status
class List (libbe.command.Command):
"""List bugs
+ >>> import sys
>>> import libbe.bugdir
- >>> bd = libbe.bugdir.SimpleBugDir()
- >>> bd.uuid = '1234abcd'
+ >>> bd = libbe.bugdir.SimpleBugDir(memory=False)
>>> cmd = List()
>>> cmd._setup_io = lambda i_enc,o_enc : None
+ >>> cmd.stdout = sys.stdout
>>> cmd.run(bd)
- 123/a:om: Bug A
+ sim/a:om: Bug A
>>> cmd.run(bd, {'status':'closed'})
- 123/b:cm: Bug B
+ sim/b:cm: Bug B
+ >>> bd.storage.writeable
+ True
>>> bd.cleanup()
"""
# ])
def _run(self, bugdir, **params):
+ writeable = bugdir.storage.writeable
bugdir.storage.writeable = False
cmp_list, status, severity, assigned, extra_strings_regexps = \
self._parse_params(params)
print >> self.stdout, bug.uuid
else:
self._list_bugs(bugs, xml=params['xml'])
+ bugdir.storage.writeable = writeable
def _parse_params(self, params):
cmp_list = []
In addition, there are some shortcut options that set boolean flags.
The boolean options are ignored if the matching string option is used.
-""" % (','.join(bug.status_values), ','.join(bug.severity_values))
-
-def complete(options, args, parser):
- for option, value in cmdutil.option_value_pairs(options, parser):
- if value == "--complete":
- if option == "status":
- raise cmdutil.GetCompletions(bug.status_values)
- elif option == "severity":
- raise cmdutil.GetCompletions(bug.severity_values)
- raise cmdutil.GetCompletions()
- if "--complete" in args:
- raise cmdutil.GetCompletions() # no positional arguments for list
+""" % (','.join(libbe.bug.status_values),
+ ','.join(libbe.bug.severity_values))
return [fragment]
def complete_extra_strings(command, argument, fragment=None):
return [fragment]
+def complete_bug_id(command, argument, fragment=None):
+ return [fragment]
def select_values(string, possible_values, name="unkown"):
"""
Note that the Arch VCS backend *enforces* ids with this format.
"""
+import os
import re
from socket import gethostname
import libbe
import libbe.storage.util.config
-def get_fallback_username(self):
+def get_fallback_username():
name = None
for env in ["LOGNAME", "USERNAME"]:
if os.environ.has_key(env):
assert name != None
return name
-def get_fallback_email(self):
+def get_fallback_email():
hostname = gethostname()
name = get_fallback_username()
return "%s@%s" % (name, hostname)
assert len(name) > 0
return (name, email)
-def get_user_id(self, storage=None):
+def get_user_id(storage=None):
"""
Sometimes the storage will also keep track of the user id (e.g. most VCSs).
"""
user = create_user_id(name, email)
return user
-def set_user_id(self, user_id):
+def set_user_id(user_id):
"""
"""
user = libbe.storage.util.config.set_val('user', user_id)
projects / be.git / commitdiff