Preliminary maildir implementation. Folder support is still incomplete

Mailnag/backends/__init__.py

@@ -28,7 +28,7 @@ import re
 
 from Mailnag.backends.imap import IMAPMailboxBackend
 from Mailnag.backends.pop3 import POP3MailboxBackend
-from Mailnag.backends.local import MBoxBackend
+from Mailnag.backends.local import MBoxBackend, MaildirBackend
 from Mailnag.common.utils import splitstr
 
 
@@ -80,6 +80,8 @@ _backends = {
 	'mbox' : Backend(MBoxBackend, [
 				Param('path', 'path', str, str, ''),
 			 ]),
+	'maildir' : Backend(MaildirBackend, [
+			]),
 }
 
 

Mailnag/backends/local.py

@@ -33,7 +33,7 @@ class MBoxBackend(MailboxBackend):
 	"""Implementation of mbox mail boxes."""
 	
 	def __init__(self, name = '', path=None, **kw):
-		"""Initialize mbox mailbox backens with a name and path."""
+		"""Initialize mbox mailbox backend with a name and path."""
 		self.name = name
 		self.path = path
 		self.opened = False
@@ -82,3 +82,66 @@ class MBoxBackend(MailboxBackend):
 	def cancel_notifications(self):
 		raise NotImplementedError("mbox does not support notifications")
 
+
+class MaildirBackend(MailboxBackend):
+	"""Implementation of maildir mail boxes."""
+
+	def __init__(self, name = '', path=None, **kw):
+		"""Initialize maildir mailbox backend with a name and path."""
+		self.name = name
+		self.path = path
+		self.opened = False
+
+
+	def open(self, reopen=False):
+		"""'Open' mailbox. (Actually just checks that maildir directory exists.)"""
+		if not os.path.isdir(self.path):
+			raise IOError('Mailbox {} does not exist.'.format(self.path))
+		self.opened = True
+
+
+	def close(self):
+		"""Close mailbox."""
+		self.opened = False
+
+
+	def is_open(self):
+		"""Return True if mailbox is opened."""
+		return self.opened
+
+
+	def list_messages(self):
+		"""List unread messages from the mailbox.
+		Yields pairs (folder, message) where folder is always ''.
+		"""
+		maildir = mailbox.Maildir(self.path, factory=None, create=False)
+		folder = ''
+		try:
+			for msg in maildir:
+				if 'S' not in msg.get_flags():
+					yield folder, msg
+		finally:
+			maildir.close()
+
+
+	def request_folders(self):
+		"""Lists folder from maildir recursively."""
+
+		def list_folders(maildir, parent):
+			for folder in  maildir.list_folders():
+				this_folder = parent + folder
+				yield this_folder
+				for subfolder in list_folders(maildir.get_folder(folder), this_folder + '/'):
+					yield subfolder
+
+		maildir = mailbox.Maildir(self.path, factory=None, create=False)
+		return [''] + list(list_folders(maildir, ''))
+
+
+	def notify_next_change(self, callback=None, timeout=None):
+		raise NotImplementedError("mbox does not support notifications")
+
+
+	def cancel_notifications(self):
+		raise NotImplementedError("mbox does not support notifications")
+

tests/test_backend_local.py

@@ -28,6 +28,8 @@ import pytest
 from Mailnag.backends import create_backend
 
 
+# mbox tests
+
 def test_create_mbox_backend():
 	be = create_backend('mbox')
 	assert be is not None
@@ -117,6 +119,103 @@ def test_mbox_open_should_fail_if_mailbox_does_not_exist(tmpdir):
 		be.open()
 
 
+# maildir tests
+
+
+def test_create_maildir_backend():
+	be = create_backend('maildir')
+	assert be is not None
+
+
+def test_initially_maildir_should_be_closed():
+	be = create_backend('maildir')
+	assert not be.is_open()
+
+
+def test_when_opened_maildir_should_be_open(tmpdir):
+	path = str(tmpdir.join('sample'))
+	sample_maildir = mailbox.Maildir(path, create=True)
+	be = create_backend('maildir', path=path)
+	be.open()
+	assert be.is_open()
+
+
+def test_closed_maildir_should_be_closed(tmpdir):
+	path = str(tmpdir.join('sample'))
+	sample_maildir = mailbox.Maildir(path, create=True)
+	be = create_backend('maildir', path=path)
+	be.open()
+	be.close()
+	assert not be.is_open()
+
+
+def test_maildir_lists_no_messages_from_empty_mailbox(tmpdir):
+	path = str(tmpdir.join('sample'))
+	sample_maildir = mailbox.Maildir(path, create=True)
+
+	be = create_backend('maildir', name='sample', path=path)
+	be.open()
+	try:
+		msgs = list(be.list_messages())
+		assert len(msgs) == 0
+	finally:
+		be.close()
+
+
+def test_maildir_lists_two_messages_from_mailbox(tmpdir):
+	path = str(tmpdir.join('sample'))
+	sample_maildir = mailbox.Maildir(path, create=True)
+	add_maildir_message(sample_maildir, 'blaa-blaa-1', '', 'new')
+	add_maildir_message(sample_maildir, 'blaa-blaa-2', '', 'cur')
+	add_maildir_message(sample_maildir, 'blaa-blaa-3', 'S', 'cur')
+	sample_maildir.close()
+
+	be = create_backend('maildir', name='sample', path=path)
+	be.open()
+	try:
+		msgs = list(be.list_messages())
+		folders = [folder for folder, msg in msgs]
+		msg_ids = set(msg.get('message-id') for folder, msg in msgs)
+	finally:
+		be.close()
+	assert len(msgs) == 2
+	assert all(folder == '' for folder in folders)
+	assert msg_ids == set(['blaa-blaa-1', 'blaa-blaa-2'])
+
+
+def test_maildir_should_have_folders(tmpdir):
+	path = str(tmpdir.join('sample'))
+	sample_maildir = mailbox.Maildir(path, create=True)
+	f = sample_maildir.add_folder('folder1')
+	sample_maildir.add_folder('folder2')
+	f.add_folder('subfolder')
+
+	be = create_backend('maildir', path=path)
+	be.open()
+	folders = be.request_folders()
+	assert set(['', 'folder1', 'folder2', 'folder1/subfolder']) == set(folders)
+
+
+def test_maildir_does_not_support_notifications(tmpdir): # for now
+	path = str(tmpdir.join('sample'))
+	sample_maildir = mailbox.Maildir(path, create=True)
+	be = create_backend('maildir', path=path)
+	be.open()
+	with pytest.raises(NotImplementedError):
+		be.notify_next_change()
+	with pytest.raises(NotImplementedError):
+		be.cancel_notifications()
+
+
+def test_maildir_open_should_fail_if_mailbox_does_not_exist(tmpdir):
+	path = str(tmpdir.join('not-exist'))
+
+	be = create_backend('maildir', path=path)
+	with pytest.raises(IOError):
+		be.open()
+
+
+
 # Helper fuctions
 
 def add_mbox_message(mbox, msg_id, flags):
@@ -133,3 +232,18 @@ def add_mbox_message(mbox, msg_id, flags):
 	finally:
 		mbox.unlock()
 
+def add_maildir_message(maildir, msg_id, flags, subdir):
+	m = mailbox.MaildirMessage()
+	m.set_payload('Hello world!', 'ascii')
+	m.add_header('from', 'me@example.org')
+	m.add_header('to', 'you@example.org')
+	m.add_header('subject', 'Hi!')
+	m.add_header('message-id', msg_id)
+	m.set_flags(flags)
+	m.set_subdir(subdir)
+	maildir.lock()
+	try:
+		maildir.add(m)
+	finally:
+		maildir.unlock()
+