Twistedでテスト用SMTPサーバーを書く

メールのテストとかってめんどい。
Mockとかそーいうのすらもめんどい。
もうサーバーを書いてしまった方がいいんじゃない？ってひと向け。
Twisted本持ってるひとには無用のエントリだね、そのまんまだから。

mailserver.py

from twisted.mail import smtp, maildir
from zope.interface import implements
from twisted.internet import protocol, reactor, defer

import os
from email.Header import Header

class MaildirMessageWriter(object):
    implements(smtp.IMessage)

    def __init__(self, user_dir):
        print "mailbox %s " % user_dir 
        if not os.path.exists(user_dir):
            os.mkdir(user_dir)
        inbox_dir = os.path.join(user_dir, 'Inbox')
        self.mailbox = maildir.MaildirMailbox(inbox_dir)
        self.lines = []

    def lineReceived(self, line):
        self.lines.append(line)

    def eomReceived(self):
        self.lines.append('')
        messge_data = '\n'.join(self.lines)
        return self.mailbox.appendMessage(messge_data)

    def connectionLost(self):
        del(self.lines)
       
class LocalDelivary(object):
    implements(smtp.IMessageDelivery)

    def __init__(self, base_dir, valid_domain):
        if not os.path.isdir(base_dir):
            raise ValueError, "%s is not dirctory"
        self.base_dir = base_dir
        self.valid_domain = valid_domain
    
    def receivedHeader(self, helo, origin, recipients):
        host_name, client_ip = helo
        header_value = "by %s from %s with ESMTP ; %s" % (host_name, 
                client_ip,
                smtp.rfc822date())
        return "Recived: %s " % Header(header_value) 

    def validateTo(self, user):
        if not user.dest.domain in self.valid_domain:
            raise smtp.SMTPBadRcpt()
        print "Accepting mail for %s " % user.dest
        return lambda:MaildirMessageWriter(
                self._get_addr_writer(str(user.dest)))

    def _get_addr_writer(self, address):
        return os.path.join(self.base_dir, "%s" % address)
    
    def validateFrom(self, helo, origin_address):
        return origin_address

class SMTPFactory(protocol.ServerFactory):
    
    def __init__(self, base_dir, valid_domain):
        self.base_dir = base_dir
        self.valid_domain = valid_domain

    def buildProtocol(self, addr):
        delivery = LocalDelivary(self.base_dir, self.valid_domain)
        smtp_protocol = smtp.SMTP(delivery)
        smtp_protocol.factory = self
        return smtp_protocol

if __name__ == '__main__':
    reactor.listenTCP(2555, SMTPFactory('/tmp', ['localhost']))
    from twisted.internet import ssl
    reactor.run()

このコードだけでもなんとなく流れがわかる。
ToをvalidateしてMessageを処理するってイメージ。

最初Growlに通知しようかと思ってたんだけどそんなに良くもなかったのでそのままファイルに書き出し。
newってフォルダ中になんとかIDみたいな（なんだっけ）のが付与されてるファイルが吐かれる。
こんな感じ

Recived: by [127.0.0.1] from 127.0.0.1 with ESMTP ; Thu, 16 Aug 2007 08:28:04 +0900 
Mime-Version: 1.0 (Apple Message framework v752.3)
Content-Transfer-Encoding: 7bit
Message-Id: <34D621AF-F079-49D3-A0E1-DBAEE64B61CA@gmail.com>
Content-Type: text/plain; charset=US-ASCII; format=flowed
To: test@localhost
From: =?ISO-2022-JP?B?GyRCPj44NhsoQiAbJEJLLRsoQg==?= <xxxxxxxxxx@gmail.com>
Subject: TEST
Date: Thu, 16 Aug 2007 08:28:04 +0900
X-Mailer: Apple Mail (2.752.3)

test mail

まあ見たらわかるけどeomReceivedのところをいじればいろいろできる。
今回はMaildirMailBoxに任せてるけど。
（こいつ自体も内部でFactoryから返ってくるクラスに委譲してるけど）
smtp、popとかその辺は全然見てなかったんだけどなかなか簡単にいろんなことできそう。
もっとカスタマイズしたいひとはsmtp以下のソース見てみるといいよ。

うくく。