Option = base.Option
Argument = base.Argument
Command = base.Command
+InputOutput = base.InputOutput
+StdInputOutput = base.StdInputOutput
+StringInputOutput = base.StringInputOutput
+UnconnectedStorageGetter = base.UnconnectedStorageGetter
+StorageCallbacks = base.StorageCallbacks
+UserInterface = base.UserInterface
-__all__ = [UserError, UnknownCommand, get_command, get_command_class,
- commands, Option, Argument, Command]
+__all__ = [UserError, UnknownCommand,
+ get_command, get_command_class, commands,
+ Option, Argument, Command,
+ InputOutput, StdInputOutput, StringInputOutput,
+ StorageCallbacks, UnconnectedStorageGetter,
+ UserInterface]
>>> import sys
>>> import libbe.bugdir
>>> bd = libbe.bugdir.SimpleBugDir(memory=False)
- >>> cmd = Assign()
- >>> cmd._storage = bd.storage
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
+ >>> ui.storage_callbacks.set_storage(bd.storage)
+ >>> cmd = Assign(ui=ui)
>>> bd.bug_from_uuid('a').assigned is None
True
- >>> ret = cmd.run({'user-id':u'Fran\xe7ois'}, ['-', '/a'])
+ >>> ui._user_id = u'Fran\xe7ois'
+ >>> ret = ui.run(cmd, args=['-', '/a'])
>>> bd.flush_reload()
>>> bd.bug_from_uuid('a').assigned
u'Fran\\xe7ois'
- >>> ret = cmd.run(args=['someone', '/a', '/b'])
+ >>> ret = ui.run(cmd, args=['someone', '/a', '/b'])
>>> bd.flush_reload()
>>> bd.bug_from_uuid('a').assigned
'someone'
>>> bd.bug_from_uuid('b').assigned
'someone'
- >>> ret = cmd.run(args=['none', '/a'])
+ >>> ret = ui.run(cmd, args=['none', '/a'])
>>> bd.flush_reload()
>>> bd.bug_from_uuid('a').assigned is None
True
+ >>> ui.cleanup()
>>> bd.cleanup()
"""
name = 'assign'
import codecs
import optparse
import os.path
+import StringIO
import sys
import libbe
raise UnknownCommand(command_name)
return cmd
-def get_command_class(module, command_name):
+def get_command_class(module=None, command_name=None):
"""Retrieves a command class from a module.
>>> import_xml_mod = get_command('import-xml')
>>> import_xml = get_command_class(import_xml_mod, 'import-xml')
>>> repr(import_xml)
"<class 'libbe.command.import_xml.Import_XML'>"
+ >>> import_xml = get_command_class(command_name='import-xml')
+ >>> repr(import_xml)
+ "<class 'libbe.command.import_xml.Import_XML'>"
"""
+ if module == None:
+ module = get_command(command_name)
try:
cname = command_name.capitalize().replace('-', '_')
cmd = getattr(module, cname)
return ''.join(ret[:-1])
class Command (object):
- """One-line command description.
+ """One-line command description here.
>>> c = Command()
>>> print c.help()
name = 'command'
- def __init__(self, input_encoding=None, output_encoding=None,
- get_unconnected_storage=None, ui=None):
- self.input_encoding = input_encoding
- self.output_encoding = output_encoding
- self.get_unconnected_storage = get_unconnected_storage
- self.ui = ui # calling user-interface, e.g. for Help()
+ def __init__(self, ui=None):
+ self.ui = ui # calling user-interface
self.status = None
self.result = None
self.restrict_file_access = True
self.args = []
def run(self, options=None, args=None):
+ self.status = 1 # in case we raise an exception
+ params = self._parse_options_args(options, args)
+ if params['help'] == True:
+ pass
+ else:
+ params.pop('help')
+ if params['complete'] != None:
+ pass
+ else:
+ params.pop('complete')
+
+ self.status = self._run(**params)
+ return self.status
+
+ def _parse_options_args(self, options=None, args=None):
if options == None:
options = {}
if args == None:
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:
- params.pop('help')
- if params['complete'] != None:
- pass
- else:
- params.pop('complete')
-
- self._setup_io(self.input_encoding, self.output_encoding)
- self.status = self._run(**params)
- return self.status
+ return params
def _run(self, **kwargs):
raise NotImplementedError
- def _setup_io(self, input_encoding=None, output_encoding=None):
- if input_encoding == None:
- input_encoding = libbe.util.encoding.get_input_encoding()
- if output_encoding == None:
- output_encoding = libbe.util.encoding.get_output_encoding()
- self.stdin = codecs.getwriter(input_encoding)(sys.stdin)
- self.stdin.encoding = input_encoding
- self.stdout = codecs.getwriter(output_encoding)(sys.stdout)
- self.stdout.encoding = output_encoding
-
def help(self, *args):
return '\n\n'.join([self.usage(),
self._option_help(),
raise UserError('file access restricted!\n %s not in %s'
% (path, repo))
- def _get_unconnected_storage(self):
- """Callback for use by commands that need it."""
+ def cleanup(self):
+ pass
+
+class InputOutput (object):
+ def __init__(self, stdin=None, stdout=None):
+ self.stdin = stdin
+ self.stdout = stdout
+
+ def setup_command(self, command):
+ if not hasattr(self.stdin, 'encoding'):
+ self.stdin.encoding = libbe.util.encoding.get_input_encoding()
+ if not hasattr(self.stdout, 'encoding'):
+ self.stdout.encoding = libbe.util.encoding.get_output_encoding()
+ command.stdin = self.stdin
+ command.stdin.encoding = self.stdin.encoding
+ command.stdout = self.stdout
+ command.stdout.encoding = self.stdout.encoding
+
+ def cleanup(self):
+ pass
+
+class StdInputOutput (InputOutput):
+ def __init__(self, input_encoding=None, output_encoding=None):
+ stdin,stdout = self._get_io(input_encoding, output_encoding)
+ InputOutput.__init__(self, stdin, stdout)
+
+ def _get_io(self, input_encoding=None, output_encoding=None):
+ if input_encoding == None:
+ input_encoding = libbe.util.encoding.get_input_encoding()
+ if output_encoding == None:
+ output_encoding = libbe.util.encoding.get_output_encoding()
+ stdin = codecs.getwriter(input_encoding)(sys.stdin)
+ stdin.encoding = input_encoding
+ stdout = codecs.getwriter(output_encoding)(sys.stdout)
+ stdout.encoding = output_encoding
+ return (stdin, stdout)
+
+class StringInputOutput (InputOutput):
+ """
+ >>> s = StringInputOutput()
+ >>> s.set_stdin('hello')
+ >>> s.stdin.read()
+ 'hello'
+ >>> s.stdin.read()
+ >>> print >> s.stdout, 'goodbye'
+ >>> s.get_stdout()
+ 'goodbye\n'
+ >>> s.get_stdout()
+ ''
+
+ Also works with unicode strings
+
+ >>> s.set_stdin(u'hello')
+ >>> s.stdin.read()
+ u'hello'
+ >>> print >> s.stdout, u'goodbye'
+ >>> s.get_stdout()
+ u'goodbye\n'
+ """
+ def __init__(self):
+ stdin = StringIO.StringIO()
+ stdin.encoding = 'utf-8'
+ stdout = StringIO.StringIO()
+ stdout.encoding = 'utf-8'
+ InputOutput.__init__(self, stdin, stdout)
+
+ def set_stdin(self, stdin_string):
+ self.stdin = StringIO.StringIO(stdin_string)
+
+ def get_stdout(self):
+ ret = self.stdout.getvalue()
+ self.stdout = StringIO.StringIO() # clear stdout for next read
+ self.stdin.encoding = 'utf-8'
+ return ret
+
+class UnconnectedStorageGetter (object):
+ def __init__(self, location):
+ self.location = location
+
+ def __call__(self):
+ return libbe.storage.get_storage(self.location)
+
+class StorageCallbacks (object):
+ def __init__(self, location=None):
+ if location == None:
+ location = '.'
+ self.location = location
+ self._get_unconnected_storage = UnconnectedStorageGetter(location)
+
+ def setup_command(self, command):
+ command._get_unconnected_storage = self.get_unconnected_storage
+ command._get_storage = self.get_storage
+ command._get_bugdir = self.get_bugdir
+
+ def get_unconnected_storage(self):
+ """
+ Callback for use by commands that need it.
+
+ The returned Storage instance is may actually be connected,
+ but commands that make use of the returned value should only
+ make use of non-connected Storage methods. This is mainly
+ intended for the init command, which calls Storage.init().
+ """
if not hasattr(self, '_unconnected_storage'):
- if self.get_unconnected_storage == None:
+ if self._get_unconnected_storage == None:
raise NotImplementedError
- self._unconnected_storage = self.get_unconnected_storage()
+ self._unconnected_storage = self._get_unconnected_storage()
return self._unconnected_storage
- def _get_storage(self):
- """
- Callback for use by commands that need it.
+ def set_unconnected_storage(self, unconnected_storage):
+ self._unconnected_storage = unconnected_storage
- Note that with the current implementation,
- _get_unconnected_storage() will not work after this method
- runs, but that shouldn't be an issue for any command I can
- think of...
- """
+ def get_storage(self):
+ """Callback for use by commands that need it."""
if not hasattr(self, '_storage'):
- self._storage = self._get_unconnected_storage()
+ self._storage = self.get_unconnected_storage()
self._storage.connect()
version = self._storage.storage_version()
if version != libbe.storage.STORAGE_VERSION:
raise libbe.storage.InvalidStorageVersion(version)
return self._storage
- def _get_bugdir(self):
+ def set_storage(self, storage):
+ self._storage = storage
+
+ def get_bugdir(self):
"""Callback for use by commands that need it."""
if not hasattr(self, '_bugdir'):
- self._bugdir = libbe.bugdir.BugDir(self._get_storage(), from_storage=True)
+ self._bugdir = libbe.bugdir.BugDir(self.get_storage(),
+ from_storage=True)
return self._bugdir
+ def set_bugdir(self, bugdir):
+ self._bugdir = bugdir
+
+ def cleanup(self):
+ if hasattr(self, '_storage'):
+ self._storage.disconnect()
+
+class UserInterface (object):
+ def __init__(self, io=None, location=None):
+ if io == None:
+ io = StringInputOutput()
+ self.io = io
+ self.storage_callbacks = StorageCallbacks(location)
+ self.restrict_file_access = True
+
+ def help(self):
+ raise NotImplementedError
+
+ def run(self, command, options=None, args=None):
+ command.ui = self
+ self.io.setup_command(command)
+ self.storage_callbacks.setup_command(command)
+ command.restrict_file_access = self.restrict_file_access
+ command._get_user_id = self._get_user_id
+ return command.run(options, args)
+
def _get_user_id(self):
"""Callback for use by commands that need it."""
if not hasattr(self, '_user_id'):
- self._user_id = libbe.ui.util.user.get_user_id(self._get_storage())
+ self._user_id = libbe.ui.util.user.get_user_id(
+ self.storage_callbacks.get_storage())
return self._user_id
def cleanup(self):
- if hasattr(self, '_storage'):
- self._storage.disconnect()
+ self.storage_callbacks.cleanup()
+ self.io.cleanup()
>>> import time
>>> import libbe.bugdir
>>> bd = libbe.bugdir.SimpleBugDir(memory=False)
- >>> cmd = Comment()
- >>> cmd._storage = bd.storage
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
+ >>> ui.storage_callbacks.set_storage(bd.storage)
+ >>> cmd = Comment(ui=ui)
- >>> ret = cmd.run({'user-id':u'Fran\\xe7ois'},
- ... ['/a', 'This is a comment about a'])
+ >>> ui._user_id = u'Fran\\xe7ois'
+ >>> ret = ui.run(cmd, args=['/a', 'This is a comment about a'])
>>> bd.flush_reload()
>>> bug = bd.bug_from_uuid('a')
>>> bug.load_comments(load_full=False)
>>> if 'EDITOR' in os.environ:
... del os.environ['EDITOR']
- >>> ret = cmd.run({'user-id':u'Frank'}, ['/b'])
+ >>> ui._user_id = u'Frank'
+ >>> ret = ui.run(cmd, args=['/b'])
Traceback (most recent call last):
UserError: No comment supplied, and EDITOR not specified.
>>> os.environ['EDITOR'] = "echo 'I like cheese' > "
- >>> ret = cmd.run({'user-id':u'Frank'}, ['/b'])
+ >>> ret = ui.run(cmd, args=['/b'])
>>> bd.flush_reload()
>>> bug = bd.bug_from_uuid('b')
>>> bug.load_comments(load_full=False)
>>> print comment.body
I like cheese
<BLANKLINE>
+ >>> ui.cleanup()
>>> bd.cleanup()
"""
name = 'comment'
>>> import sys
>>> import libbe.bugdir
>>> bd = libbe.bugdir.SimpleBugDir(memory=False, versioned=True)
- >>> cmd = Commit()
- >>> cmd._storage = bd.storage
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
+ >>> ui.storage_callbacks.set_storage(bd.storage)
+ >>> cmd = Commit(ui=ui)
>>> bd.extra_strings = ['hi there']
>>> bd.flush_reload()
- >>> cmd.run({'user-id':'Joe'}, ['Making a commit']) # doctest: +ELLIPSIS
+ >>> ui.run(cmd, {'user-id':'Joe'}, ['Making a commit']) # doctest: +ELLIPSIS
Committed ...
+ >>> ui.cleanup()
>>> bd.cleanup()
"""
name = 'commit'
>>> import sys
>>> import libbe.bugdir
>>> bd = libbe.bugdir.SimpleBugDir(memory=False)
- >>> cmd = Depend()
- >>> cmd._storage = bd.storage
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
+ >>> ui.storage_callbacks.set_storage(bd.storage)
+ >>> cmd = Depend(ui=ui)
- >>> ret = cmd.run(args=['/a', '/b'])
+ >>> ret = ui.run(cmd, args=['/a', '/b'])
a blocked by:
b
- >>> ret = cmd.run(args=['/a'])
+ >>> ret = ui.run(cmd, args=['/a'])
a blocked by:
b
- >>> ret = cmd.run({'show-status':True}, ['/a']) # doctest: +NORMALIZE_WHITESPACE
+ >>> ret = ui.run(cmd, {'show-status':True}, ['/a']) # doctest: +NORMALIZE_WHITESPACE
a blocked by:
b closed
- >>> ret = cmd.run(args=['/b', '/a'])
+ >>> ret = ui.run(cmd, args=['/b', '/a'])
b blocked by:
a
b blocks:
a
- >>> ret = cmd.run({'show-status':True}, ['/a']) # doctest: +NORMALIZE_WHITESPACE
+ >>> ret = ui.run(cmd, {'show-status':True}, ['/a']) # doctest: +NORMALIZE_WHITESPACE
a blocked by:
b closed
a blocks:
b closed
- >>> ret = cmd.run({'repair':True})
- >>> ret = cmd.run({'remove':True}, ['/b', '/a'])
+ >>> ret = ui.run(cmd, {'repair':True})
+ >>> ret = ui.run(cmd, {'remove':True}, ['/b', '/a'])
b blocks:
a
- >>> ret = cmd.run({'remove':True}, ['/a', '/b'])
+ >>> ret = ui.run(cmd, {'remove':True}, ['/a', '/b'])
+ >>> ui.cleanup()
>>> bd.cleanup()
"""
name = 'depend'
>>> import sys
>>> import libbe.bugdir
>>> bd = libbe.bugdir.SimpleBugDir(memory=False, versioned=True)
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
+ >>> ui.storage_callbacks.set_storage(bd.storage)
>>> cmd = Diff()
- >>> cmd._storage = bd.storage
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
>>> original = bd.storage.commit('Original status')
>>> bug = bd.bug_from_uuid('a')
>>> bug.status = 'closed'
>>> changed = bd.storage.commit('Closed bug a')
- >>> ret = cmd.run(args=[original])
+ >>> ret = ui.run(cmd, args=[original])
Modified bugs:
abc/a:cm: Bug A
Changed bug settings:
status: open -> closed
- >>> ret = cmd.run({'subscribe':'%(bugdir_id)s:mod', 'uuids':True}, [original])
+ >>> ret = ui.run(cmd, {'subscribe':'%(bugdir_id)s:mod', 'uuids':True}, [original])
a
>>> bd.storage.versioned = False
- >>> ret = cmd.run(args=[original])
+ >>> ret = ui.run(cmd, args=[original])
Traceback (most recent call last):
...
UserError: This repository is not revision-controlled.
+ >>> ui.cleanup()
>>> bd.cleanup()
""" % {'bugdir_id':libbe.diff.BUGDIR_ID}
name = 'diff'
>>> import sys
>>> import libbe.bugdir
>>> bd = libbe.bugdir.SimpleBugDir(memory=False)
- >>> cmd = Due()
- >>> cmd._storage = bd.storage
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
+ >>> ui.storage_callbacks.set_storage(bd.storage)
+ >>> cmd = Due(ui=ui)
- >>> ret = cmd.run(args=['/a'])
+ >>> ret = ui.run(cmd, args=['/a'])
No due date assigned.
- >>> ret = cmd.run(args=['/a', 'Thu, 01 Jan 1970 00:00:00 +0000'])
- >>> ret = cmd.run(args=['/a'])
+ >>> ret = ui.run(cmd, args=['/a', 'Thu, 01 Jan 1970 00:00:00 +0000'])
+ >>> ret = ui.run(cmd, args=['/a'])
Thu, 01 Jan 1970 00:00:00 +0000
- >>> ret = cmd.run(args=['/a', 'none'])
- >>> ret = cmd.run(args=['/a'])
+ >>> ret = ui.run(cmd, args=['/a', 'none'])
+ >>> ret = ui.run(cmd, args=['/a'])
No due date assigned.
+ >>> ui.cleanup()
>>> bd.cleanup()
"""
name = 'due'
>>> import sys
>>> import libbe.bugdir
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
>>> cmd = Help()
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
- >>> ret = cmd.run(args=['help'])
+ >>> ret = ui.run(cmd, args=['help'])
usage: be help [options] [TOPIC]
<BLANKLINE>
Options:
elif params['topic'] in libbe.command.commands():
module = libbe.command.get_command(params['topic'])
Class = libbe.command.get_command_class(module,params['topic'])
- c = Class(get_unconnected_storage=self.get_unconnected_storage,
- ui=self.ui)
+ c = Class(ui=self.ui)
print >> self.stdout, c.help().rstrip('\n')
elif params['topic'] in TOPICS:
print >> self.stdout, TOPICS[params['topic']].rstrip('\n')
>>> import sys
>>> import libbe.bugdir
>>> bd = libbe.bugdir.SimpleBugDir(memory=False)
- >>> cmd = HTML()
- >>> cmd._storage = bd.storage
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
+ >>> ui.storage_callbacks.set_storage(bd.storage)
+ >>> cmd = HTML(ui=ui)
- >>> ret = cmd.run({'output':os.path.join(bd.storage.repo, 'html_export')})
+ >>> ret = ui.run(cmd, {'output':os.path.join(bd.storage.repo, 'html_export')})
>>> os.path.exists(os.path.join(bd.storage.repo, 'html_export'))
True
>>> os.path.exists(os.path.join(bd.storage.repo, 'html_export', 'index.html'))
True
>>> os.path.exists(os.path.join(bd.storage.repo, 'html_export', 'bugs', 'b.html'))
True
+ >>> ui.cleanup()
>>> bd.cleanup()
"""
name = 'html'
>>> import StringIO
>>> import libbe.bugdir
>>> bd = libbe.bugdir.SimpleBugDir(memory=False)
- >>> cmd = Import_XML()
- >>> cmd._storage = bd.storage
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
+ >>> ui.storage_callbacks.set_storage(bd.storage)
+ >>> cmd = Import_XML(ui=ui)
- >>> cmd.stdin = StringIO.StringIO('<be-xml><comment><uuid>c</uuid><body>This is a comment about a</body></comment></be-xml>')
- >>> cmd.stdin.encoding = 'ascii'
- >>> ret = cmd.run({'comment-root':'/a'}, ['-'])
+ >>> ui.io.set_stdin('<be-xml><comment><uuid>c</uuid><body>This is a comment about a</body></comment></be-xml>')
+ >>> ret = ui.run(cmd, {'comment-root':'/a'}, ['-'])
>>> bd.flush_reload()
>>> bug = bd.bug_from_uuid('a')
>>> bug.load_comments(load_full=False)
True
>>> comment.in_reply_to is None
True
+ >>> ui.cleanup()
>>> bd.cleanup()
"""
name = 'import-xml'
"""
def setUp(self):
self.bugdir = libbe.bugdir.SimpleBugDir(memory=False)
- self.cmd = Import_XML()
+ io = libbe.command.StringInputOutput()
+ self.ui = libbe.command.UserInterface(io=io)
+ self.ui.storage_callbacks.set_storage(self.bugdir.storage)
+ self.cmd = Import_XML(ui=self.ui)
self.cmd._storage = self.bugdir.storage
self.cmd._setup_io = lambda i_enc,o_enc : None
bugA = self.bugdir.bug_from_uuid('a')
</bug>
</be-xml>
"""
- self.cmd.stdin = StringIO.StringIO(self.xml)
- self.cmd.stdin.encoding = 'ascii'
- self.cmd.stdout = StringIO.StringIO()
def tearDown(self):
self.bugdir.cleanup()
+ self.ui.cleanup()
def _execute(self, params={}, args=[]):
- self.cmd.run(params, args)
+ self.ui.io.set_stdin(self.xml)
+ self.ui.run(self.cmd, params, args)
self.bugdir.flush_reload()
def testCleanBugdir(self):
uuids = list(self.bugdir.uuids())
>>> import libbe.storage.vcs
>>> import libbe.storage.vcs.base
>>> import libbe.util.utility
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
>>> cmd = Init()
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
>>> dir = libbe.util.utility.Dir()
>>> vcs = libbe.storage.vcs.vcs_by_name('None')
>>> try:
... vcs.connect()
... except libbe.storage.ConnectionError:
- ... True
- True
- >>> cmd._unconnected_storage = vcs
- >>> cmd.run()
+ ... 'got error'
+ 'got error'
+ >>> ui.storage_callbacks.set_unconnected_storage(vcs)
+ >>> ui.run(cmd)
No revision control detected.
BE repository initialized.
>>> bd = libbe.bugdir.BugDir(vcs)
>>> vcs = libbe.storage.vcs.installed_vcs()
>>> vcs.repo = dir.path
>>> vcs._vcs_init(vcs.repo)
- >>> cmd._unconnected_storage = vcs
+ >>> ui.storage_callbacks.set_unconnected_storage(vcs)
>>> if vcs.name in libbe.storage.vcs.base.VCS_ORDER:
- ... cmd.run() # doctest: +ELLIPSIS
+ ... ui.run(cmd) # doctest: +ELLIPSIS
... else:
... vcs.init()
... vcs.connect()
>>> import sys
>>> import libbe.bugdir
>>> bd = libbe.bugdir.SimpleBugDir(memory=False)
- >>> cmd = List()
- >>> cmd._storage = bd.storage
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
+ >>> ui.storage_callbacks.set_storage(bd.storage)
+ >>> cmd = List(ui=ui)
- >>> ret = cmd.run()
+ >>> ret = ui.run(cmd)
abc/a:om: Bug A
- >>> ret = cmd.run({'status':'closed'})
+ >>> ret = ui.run(cmd, {'status':'closed'})
abc/b:cm: Bug B
>>> bd.storage.writeable
True
+ >>> ui.cleanup()
>>> bd.cleanup()
"""
>>> import libbe.bugdir
>>> import libbe.comment
>>> bd = libbe.bugdir.SimpleBugDir(memory=False)
- >>> cmd = Merge()
- >>> cmd._storage = bd.storage
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
+ >>> ui.storage_callbacks.set_bugdir(bd)
+ >>> cmd = Merge(ui=ui)
>>> a = bd.bug_from_uuid('a')
>>> a.comment_root.time = 0
>>> dummy = dummy.new_reply('1 2 3 4')
>>> dummy.time = 2
- >>> ret = cmd.run(args=['/a', '/b'])
+ >>> ret = ui.run(cmd, args=['/a', '/b'])
Merged bugs #abc/a# and #abc/b#
>>> bd.flush_reload()
>>> a = bd.bug_from_uuid('a')
Merged into bug #abc/a#
>>> print b.status
closed
+ >>> ui.cleanup()
>>> bd.cleanup()
"""
name = 'merge'
>>> import libbe.bugdir
>>> import libbe.util.id
>>> bd = libbe.bugdir.SimpleBugDir(memory=False)
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
+ >>> ui.storage_callbacks.set_storage(bd.storage)
>>> cmd = New()
- >>> cmd._storage = bd.storage
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
>>> uuid_gen = libbe.util.id.uuid_gen
>>> libbe.util.id.uuid_gen = lambda: 'X'
- >>> ret = cmd.run(args=['this is a test',])
+ >>> ret = ui.run(cmd, args=['this is a test',])
Created bug with ID abc/X
>>> libbe.util.id.uuid_gen = uuid_gen
>>> bd.flush_reload()
minor
>>> print bug.status
open
+ >>> ui.cleanup()
>>> bd.cleanup()
"""
name = 'new'
>>> import sys
>>> import libbe.bugdir
>>> bd = libbe.bugdir.SimpleBugDir(memory=False)
- >>> cmd = Remove()
- >>> cmd._storage = bd.storage
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
+ >>> ui.storage_callbacks.set_storage(bd.storage)
+ >>> cmd = Remove(ui=ui)
>>> print bd.bug_from_uuid('b').status
closed
- >>> ret = cmd.run(args=['/b'])
+ >>> ret = ui.run(cmd, args=['/b'])
Removed bug abc/b
>>> bd.flush_reload()
>>> try:
... except libbe.bugdir.NoBugMatches:
... print 'Bug not found'
Bug not found
+ >>> ui.cleanup()
>>> bd.cleanup()
"""
name = 'remove'
>>> import sys
>>> import libbe.bugdir
>>> bd = libbe.bugdir.SimpleBugDir(memory=False)
- >>> cmd = Set()
- >>> cmd._storage = bd.storage
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
+ >>> ui.storage_callbacks.set_storage(bd.storage)
+ >>> cmd = Set(ui=ui)
- >>> ret = cmd.run(args=['target'])
+ >>> ret = ui.run(cmd, args=['target'])
None
- >>> ret = cmd.run(args=['target', 'abcdefg'])
- >>> ret = cmd.run(args=['target'])
+ >>> ret = ui.run(cmd, args=['target', 'abcdefg'])
+ >>> ret = ui.run(cmd, args=['target'])
abcdefg
- >>> ret = cmd.run(args=['target', 'none'])
- >>> ret = cmd.run(args=['target'])
+ >>> ret = ui.run(cmd, args=['target', 'none'])
+ >>> ret = ui.run(cmd, args=['target'])
None
+ >>> ui.cleanup()
>>> bd.cleanup()
"""
name = 'set'
>>> import sys
>>> import libbe.bugdir
>>> bd = libbe.bugdir.SimpleBugDir(memory=False)
- >>> cmd = Severity()
- >>> cmd._storage = bd.storage
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
+ >>> ui.storage_callbacks.set_bugdir(bd)
+ >>> cmd = Severity(ui=ui)
>>> bd.bug_from_uuid('a').severity
'minor'
- >>> ret = cmd.run(args=['wishlist', '/a'])
+ >>> ret = ui.run(cmd, args=['wishlist', '/a'])
>>> bd.flush_reload()
>>> bd.bug_from_uuid('a').severity
'wishlist'
- >>> ret = cmd.run(args=['none', '/a'])
+ >>> ret = ui.run(cmd, args=['none', '/a'])
Traceback (most recent call last):
UserError: Invalid severity level: none
+ >>> ui.cleanup()
>>> bd.cleanup()
"""
name = 'severity'
>>> import sys
>>> import libbe.bugdir
>>> bd = libbe.bugdir.SimpleBugDir(memory=False)
- >>> cmd = Show()
- >>> cmd._storage = bd.storage
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
- >>> cmd.stdout.encoding = 'ascii'
-
- >>> ret = cmd.run(args=['/a',]) # doctest: +ELLIPSIS
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> io.stdout.encoding = 'ascii'
+ >>> ui = libbe.command.UserInterface(io=io)
+ >>> ui.storage_callbacks.set_bugdir(bd)
+ >>> cmd = Show(ui=ui)
+
+ >>> ret = ui.run(cmd, args=['/a',]) # doctest: +ELLIPSIS
ID : a
Short name : abc/a
Severity : minor
Bug A
<BLANKLINE>
- >>> ret = cmd.run({'xml':True}, ['/a']) # doctest: +ELLIPSIS
+ >>> ret = ui.run(cmd, {'xml':True}, ['/a']) # doctest: +ELLIPSIS
<?xml version="1.0" encoding="..." ?>
<be-xml>
<version>
<summary>Bug A</summary>
</bug>
</be-xml>
+ >>> ui.cleanup()
>>> bd.cleanup()
"""
name = 'show'
>>> import sys
>>> import libbe.bugdir
>>> bd = libbe.bugdir.SimpleBugDir(memory=False)
- >>> cmd = Status()
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
+ >>> ui.storage_callbacks.set_bugdir(bd)
+ >>> cmd = Status(ui=ui)
>>> cmd._storage = bd.storage
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
>>> bd.bug_from_uuid('a').status
'open'
- >>> ret = cmd.run(args=['closed', '/a'])
+ >>> ret = ui.run(cmd, args=['closed', '/a'])
>>> bd.flush_reload()
>>> bd.bug_from_uuid('a').status
'closed'
- >>> ret = cmd.run(args=['none', '/a'])
+ >>> ret = ui.run(cmd, args=['none', '/a'])
Traceback (most recent call last):
UserError: Invalid status level: none
+ >>> ui.cleanup()
>>> bd.cleanup()
"""
name = 'status'
>>> import sys
>>> import libbe.bugdir
>>> bd = libbe.bugdir.SimpleBugDir(memory=False)
- >>> cmd = Subscribe()
- >>> cmd._bugdir = bd
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
+ >>> ui.storage_callbacks.set_bugdir(bd)
+ >>> cmd = Subscribe(ui=ui)
>>> a = bd.bug_from_uuid('a')
>>> print a.extra_strings
[]
- >>> ret = cmd.run({'subscriber':'John Doe <j@doe.com>'}, ['/a']) # doctest: +NORMALIZE_WHITESPACE
+ >>> ret = ui.run(cmd, {'subscriber':'John Doe <j@doe.com>'}, ['/a']) # doctest: +NORMALIZE_WHITESPACE
Subscriptions for abc/a:
John Doe <j@doe.com> all *
>>> bd.flush_reload()
>>> a = bd.bug_from_uuid('a')
>>> print a.extra_strings
['SUBSCRIBE:John Doe <j@doe.com>\\tall\\t*']
- >>> ret = cmd.run({'subscriber':'Jane Doe <J@doe.com>', 'servers':'a.com,b.net'}, ['/a']) # doctest: +NORMALIZE_WHITESPACE
+ >>> ret = ui.run(cmd, {'subscriber':'Jane Doe <J@doe.com>', 'servers':'a.com,b.net'}, ['/a']) # doctest: +NORMALIZE_WHITESPACE
Subscriptions for abc/a:
Jane Doe <J@doe.com> all a.com,b.net
John Doe <j@doe.com> all *
- >>> ret = cmd.run({'subscriber':'Jane Doe <J@doe.com>', 'servers':'a.edu'}, ['/a']) # doctest: +NORMALIZE_WHITESPACE
+ >>> ret = ui.run(cmd, {'subscriber':'Jane Doe <J@doe.com>', 'servers':'a.edu'}, ['/a']) # doctest: +NORMALIZE_WHITESPACE
Subscriptions for abc/a:
Jane Doe <J@doe.com> all a.com,a.edu,b.net
John Doe <j@doe.com> all *
- >>> ret = cmd.run({'unsubscribe':True, 'subscriber':'Jane Doe <J@doe.com>', 'servers':'a.com'}, ['/a']) # doctest: +NORMALIZE_WHITESPACE
+ >>> ret = ui.run(cmd, {'unsubscribe':True, 'subscriber':'Jane Doe <J@doe.com>', 'servers':'a.com'}, ['/a']) # doctest: +NORMALIZE_WHITESPACE
Subscriptions for abc/a:
Jane Doe <J@doe.com> all a.edu,b.net
John Doe <j@doe.com> all *
- >>> ret = cmd.run({'subscriber':'Jane Doe <J@doe.com>', 'servers':'*'}, ['/a']) # doctest: +NORMALIZE_WHITESPACE
+ >>> ret = ui.run(cmd, {'subscriber':'Jane Doe <J@doe.com>', 'servers':'*'}, ['/a']) # doctest: +NORMALIZE_WHITESPACE
Subscriptions for abc/a:
Jane Doe <J@doe.com> all *
John Doe <j@doe.com> all *
- >>> ret = cmd.run({'unsubscribe':True, 'subscriber':'Jane Doe <J@doe.com>'}, ['/a']) # doctest: +NORMALIZE_WHITESPACE
+ >>> ret = ui.run(cmd, {'unsubscribe':True, 'subscriber':'Jane Doe <J@doe.com>'}, ['/a']) # doctest: +NORMALIZE_WHITESPACE
Subscriptions for abc/a:
John Doe <j@doe.com> all *
- >>> ret = cmd.run({'unsubscribe':True, 'subscriber':'John Doe <j@doe.com>'}, ['/a'])
- >>> ret = cmd.run({'subscriber':'Jane Doe <J@doe.com>', 'types':'new'}, ['DIR']) # doctest: +NORMALIZE_WHITESPACE
+ >>> ret = ui.run(cmd, {'unsubscribe':True, 'subscriber':'John Doe <j@doe.com>'}, ['/a'])
+ >>> ret = ui.run(cmd, {'subscriber':'Jane Doe <J@doe.com>', 'types':'new'}, ['DIR']) # doctest: +NORMALIZE_WHITESPACE
Subscriptions for bug directory:
Jane Doe <J@doe.com> new *
- >>> ret = cmd.run({'subscriber':'Jane Doe <J@doe.com>'}, ['DIR']) # doctest: +NORMALIZE_WHITESPACE
+ >>> ret = ui.run(cmd, {'subscriber':'Jane Doe <J@doe.com>'}, ['DIR']) # doctest: +NORMALIZE_WHITESPACE
Subscriptions for bug directory:
Jane Doe <J@doe.com> all *
+ >>> ui.cleanup()
>>> bd.cleanup()
"""
name = 'subscribe'
>>> import sys
>>> import libbe.bugdir
>>> bd = libbe.bugdir.SimpleBugDir(memory=False)
- >>> cmd = Tag()
- >>> cmd._bugdir = bd
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
+ >>> ui.storage_callbacks.set_bugdir(bd)
+ >>> cmd = Tag(ui=ui)
>>> a = bd.bug_from_uuid('a')
>>> print a.extra_strings
[]
- >>> ret = cmd.run(args=['/a', 'GUI'])
+ >>> ret = ui.run(cmd, args=['/a', 'GUI'])
Tags for abc/a:
GUI
>>> bd.flush_reload()
>>> a = bd.bug_from_uuid('a')
>>> print a.extra_strings
['%(tag_tag)sGUI']
- >>> ret = cmd.run(args=['/a', 'later'])
+ >>> ret = ui.run(cmd, args=['/a', 'later'])
Tags for abc/a:
GUI
later
- >>> ret = cmd.run(args=['/a'])
+ >>> ret = ui.run(cmd, args=['/a'])
Tags for abc/a:
GUI
later
- >>> ret = cmd.run({'list':True})
+ >>> ret = ui.run(cmd, {'list':True})
GUI
later
- >>> ret = cmd.run(args=['/a', 'Alphabetically first'])
+ >>> ret = ui.run(cmd, args=['/a', 'Alphabetically first'])
Tags for abc/a:
Alphabetically first
GUI
>>> a.extra_strings = []
>>> print a.extra_strings
[]
- >>> ret = cmd.run(args=['/a'])
+ >>> ret = ui.run(cmd, args=['/a'])
>>> bd.flush_reload()
>>> a = bd.bug_from_uuid('a')
>>> print a.extra_strings
[]
- >>> ret = cmd.run(args=['/a', 'Alphabetically first'])
+ >>> ret = ui.run(cmd, args=['/a', 'Alphabetically first'])
Tags for abc/a:
Alphabetically first
- >>> ret = cmd.run({'remove':True}, ['/a', 'Alphabetically first'])
+ >>> ret = ui.run(cmd, {'remove':True}, ['/a', 'Alphabetically first'])
+ >>> ui.cleanup()
>>> bd.cleanup()
""" % {'tag_tag':TAG_TAG}
name = 'tag'
>>> import os, StringIO, sys
>>> import libbe.bugdir
>>> bd = libbe.bugdir.SimpleBugDir(memory=False)
- >>> cmd = Target()
- >>> cmd._storage = bd.storage
- >>> cmd._setup_io = lambda i_enc,o_enc : None
- >>> cmd.stdout = sys.stdout
+ >>> io = libbe.command.StringInputOutput()
+ >>> io.stdout = sys.stdout
+ >>> ui = libbe.command.UserInterface(io=io)
+ >>> ui.storage_callbacks.set_storage(bd.storage)
+ >>> cmd = Target(ui=ui)
- >>> ret = cmd.run(args=['/a'])
+ >>> ret = ui.run(cmd, args=['/a'])
No target assigned.
- >>> ret = cmd.run(args=['/a', 'tomorrow'])
- >>> ret = cmd.run(args=['/a'])
+ >>> ret = ui.run(cmd, args=['/a', 'tomorrow'])
+ >>> ret = ui.run(cmd, args=['/a'])
tomorrow
- >>> cmd.stdout = StringIO.StringIO()
- >>> ret = cmd.run({'resolve':True}, ['tomorrow'])
- >>> output = cmd.stdout.getvalue().strip()
+ >>> ui.io.stdout = StringIO.StringIO()
+ >>> ret = ui.run(cmd, {'resolve':True}, ['tomorrow'])
+ >>> output = ui.io.get_stdout().strip()
>>> target = bd.bug_from_uuid(output)
>>> print target.summary
tomorrow
>>> print target.severity
target
- >>> cmd.stdout = sys.stdout
- >>> ret = cmd.run(args=['/a', 'none'])
- >>> ret = cmd.run(args=['/a'])
+ >>> ui.io.stdout = sys.stdout
+ >>> ret = ui.run(cmd, args=['/a', 'none'])
+ >>> ret = ui.run(cmd, args=['/a'])
No target assigned.
+ >>> ui.cleanup()
>>> bd.cleanup()
"""
name = 'target'
fragment = parser.rargs[0]
self.complete(argument, fragment)
else:
- print command_option.callback(
+ print >> self.command.stdout, command_option.callback(
self.command, command_option, value)
raise CallbackExit
if fragment != None:
comps = [c for c in comps if c.startswith(fragment)]
if len(comps) > 0:
- print '\n'.join(comps)
+ print >> self.command.stdout, '\n'.join(comps)
raise CallbackExit
-
class BE (libbe.command.Command):
"""Class for parsing the command line arguments for `be`.
This class does not contain a useful _run() method. Call this
module's main() function instead.
- >>> be = BE()
+ >>> ui = libbe.command.UserInterface()
+ >>> ui.io.stdout = sys.stdout
+ >>> be = BE(ui=ui)
+ >>> ui.io.setup_command(be)
>>> p = CmdOptionParser(be)
>>> p.exit_after_callback = False
>>> try:
def _long_help(self):
cmdlist = []
for name in libbe.command.commands():
- module = libbe.command.get_command(name)
- Class = libbe.command.get_command_class(module, name)
+ Class = libbe.command.get_command_class(command_name=name)
assert hasattr(Class, '__doc__') and Class.__doc__ != None, \
'Command class %s missing docstring' % Class
cmdlist.append((name, Class.__doc__.splitlines()[0]))
def full_version(self, *args):
return libbe.version.version(verbose=True)
+def dispatch(ui, command, args):
+ parser = CmdOptionParser(command)
+ try:
+ options,args = parser.parse_args(args)
+ ret = ui.run(command, options, args)
+ except CallbackExit:
+ return 0
+ except libbe.command.UserError, e:
+ print >> ui.io.stdout, 'ERROR:\n', e
+ return 1
+ except libbe.storage.ConnectionError, e:
+ print >> ui.io.stdout, 'Connection Error:\n', e
+ return 1
+ except (libbe.util.id.MultipleIDMatches, libbe.util.id.NoIDMatches,
+ libbe.util.id.InvalidIDStructure), e:
+ print >> ui.io.stdout, 'Invalid id:\n', e
+ return 1
+ finally:
+ command.cleanup()
+ return ret
+
def main():
- be = BE()
+ io = libbe.command.StdInputOutput()
+ ui = libbe.command.UserInterface(io)
+ ui.restrict_file_access = False
+ ui.storage_callbacks = None
+ be = BE(ui=ui)
parser = CmdOptionParser(be)
try:
options,args = parser.parse_args()
except CallbackExit:
return 0
except libbe.command.UserError, e:
- print 'ERROR:\n', e
+ print >> be.stdout, 'ERROR:\n', e
return 1
- command_name = args[0]
+ command_name = args.pop(0)
try:
- module = libbe.command.get_command(command_name)
+ Class = libbe.command.get_command_class(command_name=command_name)
except libbe.command.UnknownCommand, e:
- print e
+ print >> be.stdout, e
return 1
- Class = getattr(module, command_name.capitalize())
- class GUCS (object):
- def __init__(self, repo):
- self.repo = repo
- def __call__(self):
- return libbe.storage.get_storage(self.repo)
- command = Class(get_unconnected_storage=GUCS(options['repo']), ui=be)
- parser = CmdOptionParser(command)
+
+ ui.storage_callbacks = libbe.command.StorageCallbacks(options['repo'])
+ command = Class(ui=ui)
if command.name in ['comment']:
paginate = 'never'
paginate = 'never'
libbe.ui.util.pager.run_pager(paginate)
- try:
- options,args = parser.parse_args(args[1:])
- command.run(options, args)
- except CallbackExit:
- command.cleanup()
- return 0
- except libbe.command.UserError, e:
- command.cleanup()
- print 'ERROR:\n', e
- return 1
- except libbe.storage.ConnectionError, e:
- command.cleanup()
- print 'Connection Error:\n', e
- return 1
- command.cleanup()
- return 0
+ ret = dispatch(ui, command, args)
+ ui.cleanup()
+ return ret
if __name__ == '__main__':
sys.exit(main())
projects / be.git / commitdiff