import sys
import time
import traceback
+import types
import doctest
import unittest
-from becommands import subscribe
-import libbe.cmdutil, libbe.encoding, libbe.utility, libbe.diff, \
- libbe.bugdir, libbe.bug, libbe.comment
+import libbe.bugdir
+import libbe.bug
+import libbe.comment
+import libbe.diff
+import libbe.command
+import libbe.command.subscribe as subscribe
+import libbe.storage
+import libbe.ui.command_line
+import libbe.util.encoding
+import libbe.util.utility
import send_pgp_mime
-THIS_SERVER = u"thor.physics.drexel.edu"
-THIS_ADDRESS = u"BE Bugs <wking@thor.physics.drexel.edu>"
-
+THIS_SERVER = u'thor.physics.drexel.edu'
+THIS_ADDRESS = u'BE Bugs <wking@thor.physics.drexel.edu>'
+UI = None
_THIS_DIR = os.path.abspath(os.path.dirname(__file__))
-BE_DIR = _THIS_DIR
-LOGPATH = os.path.join(_THIS_DIR, u"be-handle-mail.log")
+LOGPATH = os.path.join(_THIS_DIR, u'be-handle-mail.log')
LOGFILE = None
# Tag strings generated by generate_global_tags()
-SUBJECT_TAG_BASE = u"be-bug"
+SUBJECT_TAG_BASE = u'be-bug'
SUBJECT_TAG_RESPONSE = None
SUBJECT_TAG_START = None
SUBJECT_TAG_NEW = None
SUBJECT_TAG_COMMENT = None
SUBJECT_TAG_CONTROL = None
-SUBJECT_TAG_XML = None
-
-BREAK = u"--"
-NEW_REQUIRED_PSEUDOHEADERS = [u"Version"]
-NEW_OPTIONAL_PSEUDOHEADERS = [u"Reporter", u"Assign", u"Depend", u"Severity",
- u"Status", u"Tag", u"Target",
- u"Confirm", u"Subscribe"]
-CONTROL_COMMENT = u"#"
-ALLOWED_COMMANDS = [u"assign", u"comment", u"commit", u"depend", u"help",
- u"list", u"merge", u"new", u"open", u"severity", u"show",
- u"status", u"subscribe", u"tag", u"target", u"import-xml"]
+
+BREAK = u'--'
+NEW_REQUIRED_PSEUDOHEADERS = [u'Version']
+NEW_OPTIONAL_PSEUDOHEADERS = [u'Reporter', u'Assign', u'Depend', u'Severity',
+ u'Status', u'Tag', u'Target',
+ u'Confirm', u'Subscribe']
+CONTROL_COMMENT = u'#'
+ALLOWED_COMMANDS = [u'assign', u'comment', u'commit', u'depend', u'diff',
+ u'due', u'help', u'list', u'merge', u'new', u'severity',
+ u'show', u'status', u'subscribe', u'tag', u'target']
AUTOCOMMIT = True
-libbe.encoding.ENCODING = u"utf-8" # force default encoding
-ENCODING = libbe.encoding.get_encoding()
+ENCODING = u'utf-8'
+libbe.util.encoding.ENCODING = ENCODING # force default encoding
class InvalidEmail (ValueError):
def __init__(self, msg, message):
self.msg = msg
def response(self):
header = self.msg.response_header
- body = [u"Error processing email:\n",
- self.response_body(), u""]
+ body = [u'Error processing email:\n',
+ self.response_body(), u'']
response_generator = \
- send_pgp_mime.PGPMimeMessageFactory(u"\n".join(body))
+ send_pgp_mime.PGPMimeMessageFactory(u'\n'.join(body))
response = MIMEMultipart()
response.attach(response_generator.plain())
response.attach(self.msg.msg)
return ret
def response_body(self):
err_text = [unicode(self)]
- return u"\n".join(err_text)
+ return u'\n'.join(err_text)
class InvalidSubject (InvalidEmail):
def __init__(self, msg, message=None):
if message == None:
- message = u"Invalid subject"
+ message = u'Invalid subject'
InvalidEmail.__init__(self, msg, message)
def response_body(self):
- err_text = u"\n".join([unicode(self), u"",
- u"full subject was:",
+ err_text = u'\n'.join([unicode(self), u'',
+ u'full subject was:',
self.msg.subject()])
return err_text
class InvalidPseudoHeader (InvalidEmail):
def response_body(self):
- err_text = [u"Invalid pseudo-header:\n",
+ err_text = [u'Invalid pseudo-header:\n',
unicode(self)]
- return u"\n".join(err_text)
+ return u'\n'.join(err_text)
class InvalidCommand (InvalidEmail):
def __init__(self, msg, command, message=None):
- bigmessage = u"Invalid execution command '%s'" % command
+ bigmessage = u'Invalid execution command "%s"' % command
if message != None:
- bigmessage += u"\n%s" % message
+ bigmessage += u'\n%s' % message
InvalidEmail.__init__(self, msg, bigmessage)
self.command = command
class InvalidOption (InvalidCommand):
def __init__(self, msg, option, message=None):
- bigmessage = u"Invalid option '%s'" % (option)
+ bigmessage = u'Invalid option "%s"' % (option)
if message != None:
- bigmessage += u"\n%s" % message
+ bigmessage += u'\n%s' % message
InvalidCommand.__init__(self, msg, info, command, bigmessage)
self.option = option
class NotificationFailed (Exception):
def __init__(self, msg):
- bigmessage = "Notification failed: %s" % msg
+ bigmessage = 'Notification failed: %s' % msg
Exception.__init__(self, bigmessage)
self.short_msg = msg
def __init__(self, command):
self.command = command
def extract_id(self):
- if hasattr(self, "cached_id"):
+ if hasattr(self, 'cached_id'):
return self._cached_id
assert self.command.ret == 0, self.command.ret
- if self.command.command == u"new":
- regexp = re.compile(u"Created bug with ID (.*)")
+ if self.command.command.name == u'new':
+ regexp = re.compile(u'Created bug with ID (.*)')
else:
raise NotImplementedError, self.command.command
match = regexp.match(self.command.stdout)
return self._cached_id
def __str__(self):
if self.command.ret != 0:
- return "<id for %s>" % repr(self.command)
- return "<id %s>" % self.extract_id()
+ return '<id for %s>' % repr(self.command)
+ return '<id %s>' % self.extract_id()
class Command (object):
"""
- A becommands command wrapper.
- Doesn't validate input, so do that before initializing.
+ A libbe.command.Command handler.
Initialize with
Command(msg, command, args=None, stdin=None)
"""
def __init__(self, msg, command, args=None, stdin=None):
self.msg = msg
- self.command = command
if args == None:
self.args = []
else:
self.args = args
- self.stdin = stdin
+ self.command = libbe.command.get_command_class(command_name=command)()
+ self.command._setup_io = lambda i_enc,o_enc : None
self.ret = None
+ self.stdin = stdin
self.stdout = None
- self.stderr = None
- self.err = None
def __str__(self):
- return "<command: %s %s>" % (self.command, " ".join([str(s) for s in self.args]))
+ return '<command: %s %s>' % (self.command, ' '.join([str(s) for s in self.args]))
def normalize_args(self):
"""
Expand any ID placeholders in self.args.
info. Returns the exit code, stdout, and stderr produced by the
command.
"""
- if self.command in [None, u""]: # don't accept blank commands
- raise InvalidCommand(self.msg, self, "Blank")
- elif self.command not in ALLOWED_COMMANDS:
- raise InvalidCommand(self.msg, self, "Not allowed")
- assert self.ret == None, u"running %s twice!" % unicode(self)
+ if self.command.name in [None, u'']: # don't accept blank commands
+ raise InvalidCommand(self.msg, self, 'Blank')
+ elif self.command.name not in ALLOWED_COMMANDS:
+ raise InvalidCommand(self.msg, self, 'Not allowed')
+ assert self.ret == None, u'running %s twice!' % unicode(self)
self.normalize_args()
- # set stdin and catch stdout and stderr
- if self.stdin != None:
- orig_stdin = sys.stdin
- sys.stdin = StringIO.StringIO(self.stdin)
- new_stdout = codecs.getwriter(ENCODING)(StringIO.StringIO())
- new_stderr = codecs.getwriter(ENCODING)(StringIO.StringIO())
- orig_stdout = sys.stdout
- orig_stderr = sys.stderr
- sys.stdout = new_stdout
- sys.stderr = new_stderr
- # run the command
- os.chdir(BE_DIR)
- try:
- self.ret = libbe.cmdutil.execute(self.command, self.args,
- manipulate_encodings=False,
- restrict_file_access=True)
- except libbe.cmdutil.GetHelp:
- print libbe.cmdutil.help(command)
- except libbe.cmdutil.GetCompletions:
- self.err = InvalidOption(self.msg, self.command, u"--complete")
- except libbe.cmdutil.UsageError, e:
- self.err = InvalidCommand(self.msg, self,
- "%s\n%s" % (type(e), unicode(e)))
- except libbe.cmdutil.UserError, e:
- self.err = InvalidCommand(self.msg, self,
- "%s\n%s" % (type(e), unicode(e)))
- # restore stdin, stdout, and stderr
- if self.stdin != None:
- sys.stdin = orig_stdin
- sys.stdout.flush()
- sys.stderr.flush()
- sys.stdout = orig_stdout
- sys.stderr = orig_stderr
- self.stdout = codecs.decode(new_stdout.getvalue(), ENCODING)
- self.stderr = codecs.decode(new_stderr.getvalue(), ENCODING)
- if self.err != None:
- raise self.err
- return (self.ret, self.stdout, self.stderr)
+ UI.io.set_stdin(self.stdin)
+ self.ret = libbe.ui.command_line.dispatch(UI, self.command, self.args)
+ self.stdout = UI.io.get_stdout()
+ return (self.ret, self.stdout)
def response_msg(self):
if self.ret == None: self.ret = -1
- response_body = [u"Results of running: (exit code %d)" % self.ret,
- u" %s %s" % (self.command, u" ".join(self.args))]
+ response_body = [u'Results of running: (exit code %d)' % self.ret,
+ u' %s %s' % (self.command.name,u' '.join(self.args))]
if self.stdout != None and len(self.stdout) > 0:
- response_body.extend([u"", u"stdout:", u"", self.stdout])
- if self.stderr != None and len(self.stderr) > 0:
- response_body.extend([u"", u"stderr:", u"", self.stderr])
- response_body.append(u"") # trailing endline
+ response_body.extend([u'', u'output:', u'', self.stdout])
+ response_body.append(u'') # trailing endline
response_generator = \
- send_pgp_mime.PGPMimeMessageFactory(u"\n".join(response_body))
+ send_pgp_mime.PGPMimeMessageFactory(u'\n'.join(response_body))
return response_generator.plain()
class DiffTree (libbe.diff.DiffTree):
def report_string(self):
report = self.report_or_none()
if report == None:
- return "No changes"
+ return 'No changes'
else:
return send_pgp_mime.flatten(report, to_unicode=True)
def make_root(self):
return MIMEMultipart()
def join(self, root, parent, data_part):
- if hasattr(parent, "attach_child_text"):
+ if hasattr(parent, 'attach_child_text'):
self.attach_child_text = True
if data_part != None:
- send_pgp_mime.append_text(parent.data_mime_part, u"\n\n%s" % (data_part))
+ send_pgp_mime.append_text(parent.data_mime_part, u'\n\n%s' % (data_part))
self.data_mime_part = parent.data_mime_part
else:
self.data_mime_part = None
if data_part != None:
self.data_mime_part = send_pgp_mime.encodedMIMEText(data_part)
- if parent != None and parent.name in [u"new", u"rem", u"mod"]:
+ if parent != None and parent.name in [u'new', u'rem', u'mod']:
self.attach_child_text = True
if data_part == None: # make blank data_mime_part for children's appends
- self.data_mime_part = send_pgp_mime.encodedMIMEText(u"")
+ self.data_mime_part = send_pgp_mime.encodedMIMEText(u'')
if self.data_mime_part != None:
- self.data_mime_part[u"Content-Description"] = self.name
+ self.data_mime_part[u'Content-Description'] = self.name
root.attach(self.data_mime_part)
def data_part(self, depth, indent=False):
return libbe.diff.DiffTree.data_part(self, depth, indent=indent)
p=email.Parser.Parser()
self.msg=p.parsestr(self.text)
if LOGFILE != None:
- LOGFILE.write(u"handling %s\n" % self.author_addr())
- LOGFILE.write(u"\n%s\n\n" % self.text)
+ LOGFILE.write(u'handling %s\n' % self.author_addr())
+ LOGFILE.write(u'\n%s\n\n' % self.text)
self.confirm = True # enable/disable confirmation email
def _yes_no(self, boolean):
if boolean == True:
- return "yes"
- return "no"
+ return 'yes'
+ return 'no'
def author_tuple(self):
"""
Extract and normalize the sender's email address. Returns a
(name, email) tuple.
"""
- if not hasattr(self, "author_tuple_cache"):
+ if not hasattr(self, 'author_tuple_cache'):
self._author_tuple_cache = \
send_pgp_mime.source_email(self.msg, return_realname=True)
return self._author_tuple_cache
return self.msg[attr_name]
return default
def message_id(self, default=None):
- return self.default_msg_attribute_access("message-id", default=default)
+ return self.default_msg_attribute_access('message-id', default=default)
def subject(self):
- if "subject" not in self.msg:
- raise InvalidSubject(self, u"Email must contain a subject")
- return self.msg["subject"]
+ if 'subject' not in self.msg:
+ raise InvalidSubject(self, u'Email must contain a subject')
+ return self.msg['subject']
def _split_subject(self):
"""
Returns (tag, subject), with missing values replaced by None.
"""
- if hasattr(self, "_split_subject_cache"):
+ if hasattr(self, '_split_subject_cache'):
return self._split_subject_cache
- args = self.subject().split(u"]",1)
+ args = self.subject().split(u']',1)
if len(args) < 1:
self._split_subject_cache = (None, None)
elif len(args) < 2:
- self._split_subject_cache = (args[0]+u"]", None)
+ self._split_subject_cache = (args[0]+u']', None)
else:
- self._split_subject_cache = (args[0]+u"]", args[1].strip())
+ self._split_subject_cache = (args[0]+u']', args[1].strip())
return self._split_subject_cache
def _subject_tag_type(self):
"""
type = None
value = None
if tag == SUBJECT_TAG_NEW:
- type = u"new"
+ type = u'new'
elif tag == SUBJECT_TAG_CONTROL:
- type = u"control"
- elif tag == SUBJECT_TAG_XML:
- type = u"xml"
+ type = u'control'
else:
match = SUBJECT_TAG_COMMENT.match(tag)
if len(match.groups()) == 1:
- type = u"comment"
+ type = u'comment'
value = match.group(1)
return (type, value)
def validate_subject(self):
tag,subject = self._split_subject()
if not tag.startswith(SUBJECT_TAG_START):
raise InvalidSubject(
- self, u"Subject must start with '%s'" % SUBJECT_TAG_START)
+ self, u'Subject must start with "%s"' % SUBJECT_TAG_START)
tag_type,value = self._subject_tag_type()
if tag_type == None:
- raise InvalidSubject(self, u"Invalid tag '%s'" % tag)
- elif tag_type == u"new" and len(subject) == 0:
- raise InvalidSubject(self, u"Cannot create a bug with blank title")
- elif tag_type == u"comment" and len(value) == 0:
- raise InvalidSubject(self, u"Must specify a bug ID to comment")
+ raise InvalidSubject(self, u'Invalid tag "%s"' % tag)
+ elif tag_type == u'new' and len(subject) == 0:
+ raise InvalidSubject(self, u'Cannot create a bug with blank title')
+ elif tag_type == u'comment' and len(value) == 0:
+ raise InvalidSubject(self, u'Must specify a bug ID to comment')
def _get_bodies_and_mime_types(self):
"""
Traverse the email message returning (body, mime_type) for
continue
body,mime_type=(part.get_payload(decode=True),part.get_content_type())
charset = part.get_content_charset(msg_charset).lower()
- if mime_type.startswith("text/"):
+ if mime_type.startswith('text/'):
body = unicode(body, charset) # convert text types to unicode
yield (body, mime_type)
def _parse_body_pseudoheaders(self, body, required, optional,
line = line.strip()
if len(line) == 0:
break
- if ":" not in line:
+ if ':' not in line:
raise InvalidPseudoheader(self, line)
- key,value = line.split(":", 1)
+ key,value = line.split(':', 1)
value = value.strip()
if key not in all:
raise InvalidPseudoHeader(self, key)
if len(value) == 0:
raise InvalidEmail(
- self, u"Blank value for: %s" % key)
+ self, u'Blank value for: %s' % key)
dictionary[key] = value
missing = []
for key in required:
missing.append(key)
if len(missing) > 0:
raise InvalidPseudoHeader(self,
- u"Missing required pseudo-headers:\n%s"
- % u", ".join(missing))
- remaining_body = u"\n".join(body_lines[i:]).strip()
+ u'Missing required pseudo-headers:\n%s'
+ % u', '.join(missing))
+ remaining_body = u'\n'.join(body_lines[i:]).strip()
return (remaining_body, dictionary)
def _strip_footer(self, body):
body_lines = body.splitlines()
if line.startswith(BREAK):
break
i += 1 # increment past the current valid line.
- return u"\n".join(body_lines[:i]).strip()
+ return u'\n'.join(body_lines[:i]).strip()
def parse(self):
"""
Parse the commands given in the email. Raises assorted
"""
self.validate_subject()
tag_type,value = self._subject_tag_type()
- if tag_type == u"new":
+ if tag_type == u'new':
commands = self.parse_new()
- elif tag_type == u"comment":
+ elif tag_type == u'comment':
commands = self.parse_comment(value)
- elif tag_type == u"control":
+ elif tag_type == u'control':
commands = self.parse_control()
- elif tag_type == u"xml":
- commands = self.parse_xml()
else:
- raise Exception, u"Unrecognized tag type '%s'" % tag_type
+ raise Exception, u'Unrecognized tag type "%s"' % tag_type
return commands
def parse_new(self):
- command = u"new"
+ command = u'new'
tag,subject = self._split_subject()
summary = subject
- options = {u"Reporter": self.author_addr(),
- u"Confirm": self._yes_no(self.confirm),
- u"Subscribe": "no",
+ options = {u'Reporter': self.author_addr(),
+ u'Confirm': self._yes_no(self.confirm),
+ u'Subscribe': 'no',
}
body,mime_type = list(self._get_bodies_and_mime_types())[0]
comment_body,options = \
NEW_REQUIRED_PSEUDOHEADERS,
NEW_OPTIONAL_PSEUDOHEADERS,
options)
- if options[u"Confirm"].lower() == "no":
+ if options[u'Confirm'].lower() == 'no':
self.confirm = False
- if options[u"Subscribe"].lower() == "yes" and self.confirm == True:
+ if options[u'Subscribe'].lower() == 'yes' and self.confirm == True:
# respond with the subscription format rather than the
# normal command-output format, because the subscription
# format is more user-friendly.
self.confirm = False
- args = [u"--reporter", options[u"Reporter"]]
+ args = [u'--reporter', options[u'Reporter']]
args.append(summary)
commands = [Command(self, command, args)]
id = ID(commands[0])
comment_body = self._strip_footer(comment_body)
if len(comment_body) > 0:
- command = u"comment"
- comment = u"Version: %s\n\n"%options[u"Version"] + comment_body
- args = [u"--author", self.author_addr(),
- u"--alt-id", self.message_id(),
- u"--content-type", mime_type]
+ command = u'comment'
+ comment = u'Version: %s\n\n'%options[u'Version'] + comment_body
+ args = [u'--author', self.author_addr(),
+ u'--alt-id', self.message_id(),
+ u'--content-type', mime_type]
args.append(id)
- args.append(u"-")
- commands.append(Command(self, u"comment", args, stdin=comment))
+ args.append(u'-')
+ commands.append(Command(self, u'comment', args, stdin=comment))
for key,value in options.items():
- if key in [u"Version", u"Reporter", u"Confirm"]:
+ if key in [u'Version', u'Reporter', u'Confirm']:
continue # we've already handled these options
command = key.lower()
- args = [id, value]
- if key == u"Subscribe":
- if value.lower() != "yes":
+ if key in [u'Depend', u'Tag', u'Target', u'Subscribe']:
+ args = [id, value]
+ else:
+ args = [value, id]
+ if key == u'Subscribe':
+ if value.lower() != 'yes':
continue
- args = ["--subscriber", self.author_addr(), id]
+ args = ['--subscriber', self.author_addr(), id]
commands.append(Command(self, command, args))
return commands
def parse_comment(self, bug_uuid):
- command = u"comment"
+ command = u'comment'
bug_id = bug_uuid
author = self.author_addr()
alt_id = self.message_id()
body,mime_type = list(self._get_bodies_and_mime_types())[0]
- if mime_type == "text/plain":
+ if mime_type == 'text/plain':
body = self._strip_footer(body)
content_type = mime_type
- args = [u"--author", author]
+ args = [u'--author', author]
if alt_id != None:
- args.extend([u"--alt-id", alt_id])
- args.extend([u"--content-type", content_type, bug_id, u"-"])
+ args.extend([u'--alt-id', alt_id])
+ args.extend([u'--content-type', content_type, bug_id, u'-'])
commands = [Command(self, command, args, stdin=body)]
return commands
def parse_control(self):
continue
if line.startswith(BREAK):
break
+ if type(line) == types.UnicodeType:
+ # work around http://bugs.python.org/issue1170
+ line = line.encode('unicode escape')
fields = shlex.split(line)
+ if type(line) == types.UnicodeType:
+ # work around http://bugs.python.org/issue1170
+ for field in fields:
+ field = unicode(field, 'unicode escape')
command,args = (fields[0], fields[1:])
commands.append(Command(self, command, args))
if len(commands) == 0:
- raise InvalidEmail(self, u"No commands in control email.")
+ raise InvalidEmail(self, u'No commands in control email.')
return commands
- def parse_xml(self):
- command = u"import-xml"
- body,mime_type = list(self._get_bodies_and_mime_types())[0]
- if mime_type != "text/xml":
- raise InvalidEmail(self,
- u"Emails to %s must have MIME type 'text/xml', not '%s'."
- % (SUBJECT_TAG_XML, mime_type))
- args = [u"--add-only", u"-"]
- commands = [Command(self, command, args, stdin=body)]
- return commands
- def run(self):
+ def run(self, repo='.'):
self._begin_response()
commands = self.parse()
try:
- for command in commands:
+ for i,command in enumerate(commands):
command.run()
self._add_response(command.response_msg())
finally:
if AUTOCOMMIT == True:
tag,subject = self._split_subject()
- self.commit_command = Command(self, "commit", [subject])
+ self.commit_command = Command(self, 'commit', [subject])
self.commit_command.run()
if LOGFILE != None:
- LOGFILE.write(u"Autocommit:\n%s\n\n" %
+ LOGFILE.write(u'Autocommit:\n%s\n\n' %
send_pgp_mime.flatten(self.commit_command.response_msg(),
to_unicode=True))
def _begin_response(self):
tag,subject = self._split_subject()
- response_header = [u"From: %s" % THIS_ADDRESS,
- u"To: %s" % self.author_addr(),
- u"Date: %s" % libbe.utility.time_to_str(time.time()),
- u"Subject: %s Re: %s"%(SUBJECT_TAG_RESPONSE,subject)
+ response_header = [u'From: %s' % THIS_ADDRESS,
+ u'To: %s' % self.author_addr(),
+ u'Date: %s' % libbe.util.utility.time_to_str(time.time()),
+ u'Subject: %s Re: %s'%(SUBJECT_TAG_RESPONSE,subject)
]
if self.message_id() != None:
- response_header.append(u"In-reply-to: %s" % self.message_id())
+ response_header.append(u'In-reply-to: %s' % self.message_id())
self.response_header = \
- send_pgp_mime.header_from_text(text=u"\n".join(response_header))
+ send_pgp_mime.header_from_text(text=u'\n'.join(response_header))
self._response_messages = []
def _add_response(self, response_message):
self._response_messages.append(response_message)
def subscriber_emails(self, previous_revision=None):
if previous_revision == None:
if AUTOCOMMIT != True: # no way to tell what's changed
- raise NotificationFailed("Autocommit dissabled")
+ raise NotificationFailed('Autocommit dissabled')
if len(self._response_messages) == 0:
- raise NotificationFailed("Initial email failed.")
+ raise NotificationFailed('Initial email failed.')
if self.commit_command.ret != 0:
# commit failed. Error already logged.
- raise NotificationFailed("Commit failed")
+ raise NotificationFailed('Commit failed')
- # read only bugdir.
- bd = libbe.bugdir.BugDir(from_disk=True,
- manipulate_encodings=False)
+ bd = UI.storage_callbacks.get_bugdir()
+ writeable = bd.storage.writeable
+ bd.storage.writeable = False
if bd.vcs.versioned == False: # no way to tell what's changed
- raise NotificationFailed("Not versioned")
+ bd.storage.writeable = writeable
+ raise NotificationFailed('Not versioned')
bd.load_all_bugs()
subscribers = subscribe.get_bugdir_subscribers(bd, THIS_SERVER)
if len(subscribers) == 0:
+ bd.storage.writeable = writeable
return []
for subscriber,subscriptions in subscribers.items():
subscribers[subscriber] = []
emails = []
for subscriber,subscriptions in subscribers.items():
- header.replace_header("to", subscriber)
+ header.replace_header('to', subscriber)
report = diff.report_tree(subscriptions, diff_tree=DiffTree)
root = report.report_or_none()
if root != None:
emails.append(send_pgp_mime.attach_root(header, root))
if LOGFILE != None:
- LOGFILE.write(u"Preparing to notify %s of changes\n" % subscriber)
+ LOGFILE.write(u'Preparing to notify %s of changes\n' % subscriber)
+ bd.storage.writeable = writeable
return emails
def _get_before_and_after_bugdirs(self, bd, previous_revision=None):
if previous_revision == None:
commit_msg = self.commit_command.stdout
- assert commit_msg.startswith("Committed "), commit_msg
- after_revision = commit_msg[len("Committed "):]
+ assert commit_msg.startswith('Committed '), commit_msg
+ after_revision = commit_msg[len('Committed '):]
before_revision = bd.vcs.revision_id(-2)
else:
before_revision = previous_revision
def _subscriber_header(self, bd, previous_revision=None):
root_dir = os.path.basename(bd.root)
if previous_revision == None:
- subject = "Changes to %s on %s by %s" \
+ subject = 'Changes to %s on %s by %s' \
% (root_dir, THIS_SERVER, self.author_addr())
else:
- subject = "Changes to %s on %s since revision %s" \
+ subject = 'Changes to %s on %s since revision %s' \
% (root_dir, THIS_SERVER, previous_revision)
- header = [u"From: %s" % THIS_ADDRESS,
- u"To: %s" % u"DUMMY-AUTHOR",
- u"Date: %s" % libbe.utility.time_to_str(time.time()),
- u"Subject: %s Re: %s" % (SUBJECT_TAG_RESPONSE, subject)
+ header = [u'From: %s' % THIS_ADDRESS,
+ u'To: %s' % u'DUMMY-AUTHOR',
+ u'Date: %s' % libbe.util.utility.time_to_str(time.time()),
+ u'Subject: %s Re: %s' % (SUBJECT_TAG_RESPONSE, subject)
]
- return send_pgp_mime.header_from_text(text=u"\n".join(header))
+ return send_pgp_mime.header_from_text(text=u'\n'.join(header))
-def generate_global_tags(tag_base=u"be-bug"):
+def generate_global_tags(tag_base=u'be-bug'):
"""
Generate a series of tags from a base tag string.
"""
global SUBJECT_TAG_BASE, SUBJECT_TAG_START, SUBJECT_TAG_RESPONSE, \
- SUBJECT_TAG_NEW, SUBJECT_TAG_COMMENT, SUBJECT_TAG_CONTROL, \
- SUBJECT_TAG_XML
+ SUBJECT_TAG_NEW, SUBJECT_TAG_COMMENT, SUBJECT_TAG_CONTROL
SUBJECT_TAG_BASE = tag_base
- SUBJECT_TAG_START = u"[%s" % tag_base
- SUBJECT_TAG_RESPONSE = u"[%s]" % tag_base
- SUBJECT_TAG_NEW = u"[%s:submit]" % tag_base
- SUBJECT_TAG_COMMENT = re.compile(u"\[%s:([\-0-9a-z]*)]" % tag_base)
+ SUBJECT_TAG_START = u'[%s' % tag_base
+ SUBJECT_TAG_RESPONSE = u'[%s]' % tag_base
+ SUBJECT_TAG_NEW = u'[%s:submit]' % tag_base
+ SUBJECT_TAG_COMMENT = re.compile(u'\[%s:([\-0-9a-z]*)]' % tag_base)
SUBJECT_TAG_CONTROL = SUBJECT_TAG_RESPONSE
- SUBJECT_TAG_XML = u"[%s:xml]" % tag_base
def open_logfile(logpath=None):
"""
"""
global LOGPATH, LOGFILE
if logpath != None:
- if logpath == u"-":
- LOGPATH = u"stderr"
+ if logpath == u'-':
+ LOGPATH = u'stderr'
LOGFILE = sys.stderr
- elif logpath == u"none":
- LOGPATH = u"none"
+ elif logpath == u'none':
+ LOGPATH = u'none'
LOGFILE = None
elif os.path.isabs(logpath):
LOGPATH = logpath
else:
LOGPATH = os.path.join(_THIS_DIR, logpath)
- if LOGFILE == None and LOGPATH != u"none":
- LOGFILE = codecs.open(LOGPATH, u"a+", ENCODING)
- LOGFILE.write(u"Default encoding: %s\n" % ENCODING)
+ if LOGFILE == None and LOGPATH != u'none':
+ LOGFILE = codecs.open(LOGPATH, u'a+',
+ libbe.utuil.encoding.get_filesystem_encoding())
def close_logfile():
- if LOGFILE != None and LOGPATH not in [u"stderr", u"none"]:
+ if LOGFILE != None and LOGPATH not in [u'stderr', u'none']:
LOGFILE.close()
+unitsuite = unittest.TestLoader().loadTestsFromModule(sys.modules[__name__])
+suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])
+
def test():
- unitsuite = unittest.TestLoader().loadTestsFromModule(sys.modules[__name__])
- suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])
result = unittest.TextTestRunner(verbosity=2).run(suite)
num_errors = len(result.errors)
num_failures = len(result.failures)
def main(args):
from optparse import OptionParser
- global AUTOCOMMIT, BE_DIR
+ global AUTOCOMMIT, UI
- usage="be-handle-mail [options]\n\n%s" % (__doc__)
+ usage='be-handle-mail [options]\n\n%s' % (__doc__)
parser = OptionParser(usage=usage)
- parser.add_option('-b', '--be-dir', dest='be_dir', default=BE_DIR,
- metavar="DIR",
- help='Select the BE directory to serve (%default).')
+ parser.add_option('-r', '--repo', dest='repo', default=_THIS_DIR,
+ metavar='REPO',
+ help='Select the BE repository to serve (%default).')
parser.add_option('-t', '--tag-base', dest='tag_base',
- default=SUBJECT_TAG_BASE, metavar="TAG",
+ default=SUBJECT_TAG_BASE, metavar='TAG',
help='Set the subject tag base (%default).')
parser.add_option('-o', '--output', dest='output', action='store_true',
help="Don't mail the generated message, print it to stdout instead. Useful for testing be-handle-mail functionality without the whole mail transfer agent and procmail setup.")
num_bad = 1
sys.exit(num_bad)
- BE_DIR = options.be_dir
AUTOCOMMIT = options.autocommit
if options.notify_since == None:
msg_text = sys.stdin.read()
- libbe.encoding.set_IO_stream_encodings(ENCODING) # _after_ reading message
open_logfile(options.logfile)
generate_global_tags(options.tag_base)
+ io = libbe.command.StringInputOutput()
+ UI = libbe.command.UserInterface(io, location=options.repo)
+
if options.notify_since != None:
if options.subscribers == True:
if LOGFILE != None:
- LOGFILE.write(u"Checking for subscribers to notify since revision %s\n"
+ LOGFILE.write(u'Checking for subscribers to notify since revision %s\n'
% options.notify_since)
try:
m = Message(disable_parsing=True)
emails = m.subscriber_emails(options.notify_since)
except NotificationFailed, e:
if LOGFILE != None:
- LOGFILE.write(unicode(e) + u"\n")
+ LOGFILE.write(unicode(e) + u'\n')
else:
for msg in emails:
if options.output == True:
print send_pgp_mime.flatten(msg, to_unicode=True)
else:
send_pgp_mime.mail(msg, send_pgp_mime.sendmail)
+ self.commit_command.cleanup()
close_logfile()
+ UI.cleanup()
sys.exit(0)
if len(msg_text.strip()) == 0: # blank email!?
if LOGFILE != None:
- LOGFILE.write(u"Blank email!\n")
+ LOGFILE.write(u'Blank email!\n')
close_logfile()
+ UI.cleanup()
sys.exit(1)
try:
m = Message(msg_text)
response = e.response()
except Exception, e:
if LOGFILE != None:
- LOGFILE.write(u"Uncaught exception:\n%s\n" % (e,))
+ LOGFILE.write(u'Uncaught exception:\n%s\n' % (e,))
traceback.print_tb(sys.exc_traceback, file=LOGFILE)
close_logfile()
+ UI.cleanup()
sys.exit(1)
else:
response = m.response_email()
print send_pgp_mime.flatten(response, to_unicode=True)
elif m.confirm == True:
if LOGFILE != None:
- LOGFILE.write(u"Sending response to %s\n" % m.author_addr())
- LOGFILE.write(u"\n%s\n\n" % send_pgp_mime.flatten(response,
+ LOGFILE.write(u'Sending response to %s\n' % m.author_addr())
+ LOGFILE.write(u'\n%s\n\n' % send_pgp_mime.flatten(response,
to_unicode=True))
send_pgp_mime.mail(response, send_pgp_mime.sendmail)
else:
if LOGFILE != None:
- LOGFILE.write(u"Response declined by %s\n" % m.author_addr())
+ LOGFILE.write(u'Response declined by %s\n' % m.author_addr())
if options.subscribers == True:
if LOGFILE != None:
- LOGFILE.write(u"Checking for subscribers\n")
+ LOGFILE.write(u'Checking for subscribers\n')
try:
emails = m.subscriber_emails()
except NotificationFailed, e:
if LOGFILE != None:
- LOGFILE.write(unicode(e) + u"\n")
+ LOGFILE.write(unicode(e) + u'\n')
else:
for msg in emails:
if options.output == True:
send_pgp_mime.mail(msg, send_pgp_mime.sendmail)
close_logfile()
-
+ UI.cleanup()
class GenerateGlobalTagsTestCase (unittest.TestCase):
def setUp(self):
def test_restore_global_tags(self):
"Test global tag restoration by teardown function."
global SUBJECT_TAG_BASE
- self.failUnlessEqual(SUBJECT_TAG_BASE, u"be-bug")
- SUBJECT_TAG_BASE = "projectX-bug"
- self.failUnlessEqual(SUBJECT_TAG_BASE, u"projectX-bug")
+ self.failUnlessEqual(SUBJECT_TAG_BASE, u'be-bug')
+ SUBJECT_TAG_BASE = 'projectX-bug'
+ self.failUnlessEqual(SUBJECT_TAG_BASE, u'projectX-bug')
self.restore_global_tags()
- self.failUnlessEqual(SUBJECT_TAG_BASE, u"be-bug")
+ self.failUnlessEqual(SUBJECT_TAG_BASE, u'be-bug')
def test_subject_tag_base(self):
"Should set SUBJECT_TAG_BASE global correctly"
- generate_global_tags(u"projectX-bug")
- self.failUnlessEqual(SUBJECT_TAG_BASE, u"projectX-bug")
+ generate_global_tags(u'projectX-bug')
+ self.failUnlessEqual(SUBJECT_TAG_BASE, u'projectX-bug')
def test_subject_tag_start(self):
"Should set SUBJECT_TAG_START global correctly"
- generate_global_tags(u"projectX-bug")
- self.failUnlessEqual(SUBJECT_TAG_START, u"[projectX-bug")
+ generate_global_tags(u'projectX-bug')
+ self.failUnlessEqual(SUBJECT_TAG_START, u'[projectX-bug')
def test_subject_tag_response(self):
"Should set SUBJECT_TAG_RESPONSE global correctly"
- generate_global_tags(u"projectX-bug")
- self.failUnlessEqual(SUBJECT_TAG_RESPONSE, u"[projectX-bug]")
+ generate_global_tags(u'projectX-bug')
+ self.failUnlessEqual(SUBJECT_TAG_RESPONSE, u'[projectX-bug]')
def test_subject_tag_new(self):
"Should set SUBJECT_TAG_NEW global correctly"
- generate_global_tags(u"projectX-bug")
- self.failUnlessEqual(SUBJECT_TAG_NEW, u"[projectX-bug:submit]")
+ generate_global_tags(u'projectX-bug')
+ self.failUnlessEqual(SUBJECT_TAG_NEW, u'[projectX-bug:submit]')
def test_subject_tag_control(self):
"Should set SUBJECT_TAG_CONTROL global correctly"
- generate_global_tags(u"projectX-bug")
- self.failUnlessEqual(SUBJECT_TAG_CONTROL, u"[projectX-bug]")
+ generate_global_tags(u'projectX-bug')
+ self.failUnlessEqual(SUBJECT_TAG_CONTROL, u'[projectX-bug]')
def test_subject_tag_comment(self):
"Should set SUBJECT_TAG_COMMENT global correctly"
- generate_global_tags(u"projectX-bug")
- m = SUBJECT_TAG_COMMENT.match("[projectX-bug:xyz-123]")
+ generate_global_tags(u'projectX-bug')
+ m = SUBJECT_TAG_COMMENT.match('[projectX-bug:xyz-123]')
self.failUnlessEqual(len(m.groups()), 1)
- self.failUnlessEqual(m.group(1), u"xyz-123")
- def test_subject_tag_xml(self):
- "Should set SUBJECT_TAG_XML global correctly"
- generate_global_tags(u"projectX-bug")
- self.failUnlessEqual(SUBJECT_TAG_XML, u"[projectX-bug:xml]")
+ self.failUnlessEqual(m.group(1), u'xyz-123')
if __name__ == "__main__":
main(sys.argv)
projects / be.git / commitdiff