Commit 53503e7e authored by Sylvain Thénault's avatar Sylvain Thénault
Browse files

[mboximport] UnixMailbox is deprecated in python 2.6. Closes #2836573

though no class is introduced to read an mbox from stream in remplacement, so
create our own one.

MBOXImporter now have import_mbox(path) in addition to
import_mbox_stream(stream), the former using the std lib mbox class and the
later our own custom class.
parent 3a4ed4aa6831
......@@ -3,7 +3,7 @@ Section: web
Priority: optional
Maintainer: Logilab Packaging Team <>
Uploaders: Sylvain Thenault <>, David Douard <>
Build-Depends: debhelper (>=, python (>=2.5), python-dev (>=2.5)
Build-Depends: debhelper (>=, python (>=2.6), python-dev (>=2.6)
Standards-Version: 3.8.0
"""import an mbox or a single email into an cubicweb application
:organization: Logilab
:copyright: 2007-2010 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
:copyright: 2007-2013 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
:contact: --
__docformat__ = "restructuredtext en"
import re
import mailbox
from rfc822 import parsedate
from logilab.common.umessage import message_from_file
from cubicweb import Binary
......@@ -33,6 +37,23 @@ def combinaison(list):
for i in list:
yield (first, i)
class StreamMailbox(mailbox.mbox):
"""A read-only mbox format mailbox from stream."""
_mangle_from_ = True
def __init__(self, stream, factory=None, create=True):
"""Initialize a stream mailbox."""
self._message_factory = mailbox.mboxMessage
mailbox.Mailbox.__init__(self, '', factory, create)
self._file = stream
self._toc = None
self._next_key = 0
self._pending = False # No changes require rewriting the file.
self._locked = False
self._file_length = None # Used to record mailbox size
class MBOXImporter(object):
"""import content of a Unix mailbox into cubicweb as Email (and related) objects"""
......@@ -64,11 +85,13 @@ class MBOXImporter(object):
def import_mbox_stream(self, stream):
from mailbox import UnixMailbox as Mailbox
from rfc822 import parsedate
from logilab.common.umessage import message_from_file
for message in sorted(Mailbox(stream, message_from_file),
key=lambda x:parsedate(x['Date'])):
self._import(StreamMailbox(stream, message_from_file, create=False))
def import_mbox(self, path):
self._import(mailbox.mbox(path, message_from_file, create=False))
def _import(self, mailbox):
for message in sorted(mailbox, key=lambda x:parsedate(x['Date'])):
if self.autocommit:
"""unit tests for email mbox import functionnality"""
from StringIO import StringIO
from os.path import join
from logilab.common.testlib import TestCase, unittest_main
......@@ -12,9 +11,12 @@ from import MBOXImporter
class MBOXImporterTC(CubicWebTC):
def test_all(self):
def test_mbox_format(self):
mi = MBOXImporter(self.cnx)
mi.import_mbox_stream(open(join(self.datadir, 'mbox')))
def _test_base(self, mi):
self.assertEqual(sorted([(x, len(y)) for x, y in mi.created.items()]),
[('email', 2), ('emailaddress', 4),
('emailpart', 5), ('emailthread', 2), ('file', 2)])
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment