Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if self.delim == None:
listres = imapobj.list(self.reference, '""')[1]
if listres == [None] or listres == None:
# Some buggy IMAP servers do not respond well to LIST "" ""
# Work around them.
listres = imapobj.list(self.reference, '"*"')[1]
if listres == [None] or listres == None:
# No Folders were returned. This occurs, e.g. if the
# 'reference' prefix does not exist on the mail
# server. Raise exception.
err = "Server '%s' returned no folders in '%s'"% \
(self.repos.getname(), self.reference)
self.ui.warn(err)
raise Exception(err)
self.delim, self.root = \
imaputil.imapsplit(listres[0])[1:]
self.delim = imaputil.dequote(self.delim)
self.root = imaputil.dequote(self.root)
with self.connectionlock:
self.assignedconnections.append(imapobj)
self.lastowner[imapobj] = curThread.ident
return imapobj
except Exception as e:
"""If we are here then we did not succeed in getting a
connection - we should clean up and then re-raise the
error..."""
self.semaphore.release()
severity = OfflineImapError.ERROR.REPO
if type(e) == gaierror:
(probably severity MESSAGE) if e.g. no message with
this UID could be found.
"""
data = self._fetch_from_imap(str(uid), self.retrycount)
# data looks now e.g.
#[('320 (X-GM-LABELS (...) UID 17061 BODY[] {2565}','msgbody....')]
# we only asked for one message, and that msg is in data[0].
# msbody is in [0][1].
body = data[0][1].replace("\r\n", "\n")
# Embed the labels into the message headers
if self.synclabels:
m = re.search('X-GM-LABELS\s*[(](.*)[)]', data[0][0])
if m:
labels = set([imaputil.dequote(lb) for lb in imaputil.imapsplit(m.group(1))])
else:
labels = set()
labels = labels - self.ignorelabels
labels_str = imaputil.format_labels_string(self.labelsheader, sorted(labels))
# First remove old label headers that may be in the message content retrieved
# from gmail Then add a labels header with current gmail labels.
body = self.deletemessageheaders(body, self.labelsheader)
body = self.addmessageheader(body, '\n', self.labelsheader, labels_str)
if len(body)>200:
dbg_output = "%s...%s"% (str(body)[:150], str(body)[-50:])
else:
dbg_output = body
self.ui.debug('imap', "Returned object from fetching %d: '%s'"%
def _messagelabels_aux(self, arg, uidlist, labels):
"""Common code to savemessagelabels and addmessagelabels"""
labels = labels - self.ignorelabels
uidlist = [uid for uid in uidlist if uid > 0]
if len(uidlist) > 0:
imapobj = self.imapserver.acquireconnection()
try:
labels_str = '(' + ' '.join([imaputil.quote(lb) for lb in labels]) + ')'
# Coalesce uid's into ranges
uid_str = imaputil.uid_sequence(uidlist)
result = self._store_to_imap(imapobj, uid_str, arg, labels_str)
except imapobj.readonly:
self.ui.labelstoreadonly(self, uidlist, labels)
return None
finally:
self.imapserver.releaseconnection(imapobj)
if result:
retlabels = imaputil.flags2hash(imaputil.imapsplit(result)[1])['X-GM-LABELS']
retlabels = set([imaputil.dequote(lb) for lb in imaputil.imapsplit(retlabels)])
return retlabels
return None
def savemessagelabels(self, uid, labels, ignorelabels=set()):
"""Change a message's labels to `labels`.
Note that this function does not check against dryrun settings,
so you need to ensure that it is never called in a dryrun mode."""
filename = self.messagelist[uid]['filename']
filepath = os.path.join(self.getfullname(), filename)
file = open(filepath, 'rt')
content = file.read()
file.close()
oldlabels = set()
for hstr in self.getmessageheaderlist(content, self.labelsheader):
oldlabels.update(imaputil.labels_from_header(self.labelsheader, hstr))
labels = labels - ignorelabels
ignoredlabels = oldlabels & ignorelabels
oldlabels = oldlabels - ignorelabels
# Nothing to change
if labels == oldlabels:
return
# Change labels into content
labels_str = imaputil.format_labels_string(self.labelsheader,
sorted(labels | ignoredlabels))
# First remove old labels header, and then add the new one
content = self.deletemessageheaders(content, self.labelsheader)
content = self.addmessageheader(content, '\n', self.labelsheader, labels_str)
def makefolder_single(self, foldername):
self.ui.makefolder(self, foldername)
if self.account.dryrun:
return
imapobj = self.imapserver.acquireconnection()
try:
if self.account.utf_8_support:
foldername = imaputil.utf8_IMAP(foldername)
result = imapobj.create(foldername)
if result[0] != 'OK':
raise OfflineImapError("Folder '%s'[%s] could not be created. "
"Server responded: %s"% (foldername, self, str(result)),
OfflineImapError.ERROR.FOLDER)
finally:
self.imapserver.releaseconnection(imapobj)