
"""
Tests for L{xquotient.mimepart}.
"""

import os
import email.quopriMIME
from StringIO import StringIO

from twisted.trial import unittest
from twisted.python import filepath

from epsilon import extime
from axiom.store import Store, AtomicFile

from xquotient import mimepart, smtpout
from xquotient.mimestorage import Part, ExistingMessageMIMEStorer
from xquotient.test import test_grabber
from xquotient.test.util import MIMEReceiverMixin, PartMaker
from xquotient.exmess import (
    SENDER_RELATION, RECIPIENT_RELATION, COPY_RELATION, BLIND_COPY_RELATION,
    RESENT_TO_RELATION, RESENT_FROM_RELATION)

from xquotient.test.historic.stub_message5to6 import (
    msg as multipartMessageWithEmbeddedMultipartMessage)
from xmantissa import ixmantissa

def msg(s):
    return '\r\n'.join(s.splitlines())


class UtilTest(unittest.TestCase):
    """
    Test utilities associated with MIME parsing.
    """
    def test_safelyDecode(self):
        """
        Verify that _safelyDecode will properly decode a valid encoding.
        """
        self.assertEquals(u"\u1234",
                          mimepart._safelyDecode(u"\u1234".encode("utf8"),
                                                 "utf8"))

    def test_safelyDecodeUnknownCodec(self):
        """
        Verify that _safelyDecode will default to ASCII decoding with replacement
        characters in the case where the provided encoding is unknown.
        """
        self.assertEquals(u'\N{REPLACEMENT CHARACTER}' * 3 + u"hello world",
                          mimepart._safelyDecode(u"\u1234hello world".encode("utf8"),
                                                 "ascii"))

    def test_safelyDecodeMismatch(self):
        """
        Verify that _safelyDecode will not raise an exception when the given string
        does not match the given encoding.
        """
        self.assertEquals(u'\N{REPLACEMENT CHARACTER}',
                          mimepart._safelyDecode('\xe1', "utf8"))



class TestHeaderBodyParse(unittest.TestCase):
    to = 'bob@example.com'
    subject = 'a test message, comma separated'

    # An email address with a bad header. The 'to' header is bad because it
    # does not have a space after its colon.
    badHeader = msg("""\
From: alice@example.com
To:%s
Subject: %s
Date: Tue, 11 Oct 2005 14:25:12 GMT
Junk\xa0Header: value
References: <one@domain>\t<two@domain>\x20
\t<three@domain>

Hello Bob,
  How are you?
-A
""" % (to, subject))


    # An email with a couple of abnormal headers. These have spaces and colons
    # in possibly confusing arrangments.
    confusingSubject = "bar:baz"
    confusingFoo = "bar: baz"
    confusingHeader = msg("""\
From: alice@example.com
To: bob@example.com
Subject:%s
Date: Tue, 11 Oct 2005 14:25:12 GMT
Foo:%s
Junk\xa0Header: value

Hello Bob,
  How are you?
-A
""" % (confusingSubject, confusingFoo))

    brokenEncodingSubject = 'this subject is in ascii, except for this: \xe1\x88\xb4'

    brokenEncodingHeader = msg("""\
From: alice@example.com
To: bob@example.com
Subject: =?completely_invalid_encoding?q?%s?=
Date: Tue, 11 Oct 2005 14:25:12 GMT

Hello Bob,
  I am enjoying my vacation in Encodistan.  The local customs are intriguing.
-A
""" % (brokenEncodingSubject,))


    def setUp(self):
        self.part = mimepart.MIMEPart()
        # This *can't* be HeaderBodyParser, because the class is abstract.
        # It calls but does not define a parse_body method on itself.
        self.parser = mimepart.MIMEMessageParser(self.part, None)


    def parse(self, message):
        """
        Given a raw message, parse it using C{self.parser}, which should be a
        L{HeaderBodyParser}.
        """
        begin = end = 0
        for line in message.splitlines():
            end += len(line) + 1
            self.parser.lineReceived(line, begin, end)
            begin = end
        return self.part


    def test_badHeader(self):
        """
        Check that the parser keeps going after encountering bad headers. In
        particular, check that the bad header and any headers which appear
        after the bad header are parsed.
        """
        part = self.parse(self.badHeader)
        self.assertEquals(part.getHeader(u'to'), self.to)
        self.assertEquals(part.getHeader(u'subject'), self.subject)


    def test_confusingHeaders(self):
        """
        Check that confusing headers like 'foo:bar:baz' and 'foo:bar: baz' are
        parsed correctly.
        """
        part = self.parse(self.confusingHeader)
        self.assertEquals(part.getHeader(u'subject'), self.confusingSubject)
        self.assertEquals(part.getHeader(u'foo'), self.confusingFoo)


    def test_trivialMessage(self):
        """
        Test that a trivial message can be parsed and its headers retrieved.
        """
        # Control test to make sure L{TestHeaderBodyParse.parse} works.
        # Other tests in this module check that trivialMessage can be parsed
        # properly.
        part = self.parse(MessageTestMixin.trivialMessage)
        self.assertEquals(part.getHeader(u'from'), 'alice@example.com')


    def test_invalidEncoding(self):
        """
        Verify that an incorrectly incoded header will be decoded as ASCII.
        """
        part = self.parse(self.brokenEncodingHeader)
        self.assertEquals(part.getHeader(u'subject'),
                          self.brokenEncodingSubject.decode('ascii', 'replace'))




class MessageTestMixin:
    trivialMessage = msg("""\
From: alice@example.com
To: bob@example.com
Subject: a test message, comma separated
Date: Tue, 11 Oct 2005 14:25:12 GMT
Junk\xa0Header: value
References: <one@domain>\t<two@domain>\x20
\t<three@domain>

Hello Bob,
  How are you?
-A
""")

    def assertHeadersEqual(self, a, b):
        self.assertEquals(a.name, b.name)
        self.assertEquals(a.value, b.value)

    def assertTrivialMessageStructure(self, msg):
        map(self.assertHeadersEqual,
            list(msg.getAllHeaders())[:-1],
            [mimepart.Header(u"from", "alice@example.com"),
             mimepart.Header(u"to", "bob@example.com"),
             mimepart.Header(u"subject", "a test message, comma separated"),
             mimepart.Header(u"date", "Tue, 11 Oct 2005 14:25:12 GMT"),
             mimepart.Header(u"junkheader", "value"),
             mimepart.Header(u"references", "<one@domain>"),
             mimepart.Header(u"references", "<two@domain>"),
             mimepart.Header(u"references", "<three@domain>")])

        self.assertEquals(msg.getHeader(u"from"), "alice@example.com")
        self.assertEquals(msg.getHeader(u"to"), "bob@example.com")
        self.assertEquals(msg.getHeader(u"subject"), "a test message, comma separated")
        self.assertEquals(msg.getHeader(u"date"), 'Tue, 11 Oct 2005 14:25:12 GMT')
        self.assertEquals(msg.getHeader(u"junkheader"), "value")
        self.assertEquals([hdr.value for hdr in msg.getHeaders(u"references")],
                          [u"<one@domain>", u"<two@domain>", u"<three@domain>"])

    def testTrivialMessage(self):
        self._messageTest(self.trivialMessage, self.assertTrivialMessageStructure)



    messageWithUnicode = msg("""\
From: =?utf-8?b?VMOpc3Qgx5NzZXIgPHRlc3R1c2VyQGV4YW1wbGUuY29tPg==?=

Body.
""")


    def assertUnicodeHeaderValues(self, msg):
        self.assertEquals(
            msg.getHeader(u"from"),
            u"T\N{LATIN SMALL LETTER E WITH ACUTE}st "
            u"\N{LATIN CAPITAL LETTER U WITH CARON}ser "
            u"<testuser@example.com>")


    def testUnicodeHeaderValues(self):
        """
        MIME Headers may be encoded in various ways.  Assert that none of these
        encoding details make it into the resulting Header objects and that the
        non-ASCII payload is correctly interpreted.
        """
        self._messageTest(self.messageWithUnicode, self.assertUnicodeHeaderValues)


    multipartMessage = msg("""\
Envelope-to: test@domain.tld
Received: from pool-138-88-80-171.res.east.verizon.net
	([138.88.80.171] helo=meson.dyndns.org ident=69gnV1Y3MozcsVOT)
	by pyramid.twistedmatrix.com with esmtp (Exim 3.35 #1 (Debian))
	id 181WHR-0002Rq-00
	for <test@domain.tld>; Tue, 15 Oct 2002 13:18:57 -0500
Received: by meson.dyndns.org (Postfix, from userid 1000)
	id 34DEB13E95; Tue, 15 Oct 2002 14:20:18 -0400 (EDT)
Date: Tue, 15 Oct 2002 14:20:18 -0400
From: Jp Calderone <exarkun@blargchoo.choo.choo.dyndns.org>
To: test@domain.tld
Subject: [user@address: My Cool Apartment!]
Message-ID: <20021015182018.GA11673@unique.oh.yea>
Mime-Version: 1.0
Content-Type: multipart/signed; micalg=x-unknown;
	protocol="application/pgp-signature"; boundary="24zk1gE8NUlDmwG9"
Content-Disposition: inline
User-Agent: Mutt/1.3.25i
Content-Length: 2003
Lines: 72
Status: RO
x-pop3-uid: 3dbf299600000017 twistedmatrix.com


--24zk1gE8NUlDmwG9
Content-Type: multipart/mixed; boundary="h31gzZEtNLTqOjlF"
Content-Disposition: inline


--h31gzZEtNLTqOjlF
Content-Type: text/plain; charset=us-ascii
Content-Disposition: inline
Content-Transfer-Encoding: quoted-printable


--=20
   Know what I pray for? The strength to change what I can, the inability to
accept what I can't and the incapacity to tell the difference.    -- Calvin
--
 2:00pm up 147 days, 14:57, 4 users, load average: 0.00, 0.02, 0.00

--h31gzZEtNLTqOjlF
Content-Type: message/rfc822
Content-Disposition: inline

Return-Path: <dynamox@springstreet.com>
Delivered-To: thisuser@guy.wherever
Received: from mx1.springstreet.com (unknown [206.131.172.28])
	by meson.dyndns.org (Postfix) with ESMTP id E027314187
	for <exarkun@choo.choo.choo.dyndns.org>; Wed, 18 Sep 2002 00:27:14 -0400 (EDT)
Received: from app2.springstreet.com (app2.admin.springstreet.com
	[10.0.149.2])
	by mx1.springstreet.com (8.11.6/8.11.6) with ESMTP id g8I4RUe47820
	for <exarkun@xxxx.org>; Tue, 17 Sep 2002 21:27:30 -0700 (PDT)
	(envelope-from dynamox@springstreet.com)
Received: (from dynamox@localhost)
	by app2.springstreet.com (8.10.2+Sun/8.10.2) id g8I4RUb22565;
	Tue, 17 Sep 2002 21:27:30 -0700 (PDT)
Date: Tue, 17 Sep 2002 21:27:30 -0700 (PDT)
Message-Id: <200209180427.g8I4RUb22565@app2.springstreet.com>
To: this.field.is.for@the.user.it.is.for
From: someuser@anotherdomain.com
SUBJECT: My Cool Apartment!

jp,

Hey! Check out this great apartment I found at Homestore.com apartments &
rentals! You can visit it here:
http://www.springstreet.com/propid/246774?source=1xxctb079


cathedral ceilings



<message id:1032323249933719>

--h31gzZEtNLTqOjlF--

--24zk1gE8NUlDmwG9
Content-Type: application/pgp-signature
Content-Disposition: inline

-----BEGIN PGP SIGNED MESSAGE-----
Hash: SHA1


-----BEGIN PGP SIGNATURE-----
Version: GnuPG v1.0.7 (GNU/Linux)

iD8DBQE9rFxhedcO2BJA+4YRAjZqAKC6jZcmEZu0tInRreBjTbFcIh7rfACdEDhZ
oTZw+Ovl1BvLcE+pK9VFxxY=
=Uds2
-----END PGP SIGNATURE-----

--24zk1gE8NUlDmwG9--
""")

    def assertMultipartMessageStructure(self, msg):
        pass

    def testMultipartMessage(self):
        self._messageTest(self.multipartMessage, self.assertMultipartMessageStructure)


    def test_embeddedMultipartMessage(self):
        """
        Test the structure of a message with an attached multipart message.
        """

        self._messageTest(multipartMessageWithEmbeddedMultipartMessage,
                          self.assertEmbeddedMultipartMessageStructure)

    def assertEmbeddedMultipartMessageStructure(self, msg):
        """
        Ensure that all the parts of a message in an message/rfc822 part get
        parsed.
        """
        parts = list(msg.walk())
        self.assertEqual(len(parts), 6)
        types = [(p.getHeader("content-type").split(';', 1)[0]
                  .lower().strip().encode('ascii'))
                 for p in parts]
        self.assertEqual(types,
                         ['multipart/mixed', 'text/plain',
                          'message/rfc822', 'multipart/mixed',
                          'text/plain', 'image/jpeg'])



class ParsingTestCase(unittest.TestCase, MessageTestMixin):
    def _messageTest(self, source, assertMethod):
        deliveryDir = self.mktemp()
        os.makedirs(deliveryDir)
        f = AtomicFile(
            filepath.FilePath(deliveryDir).child('tmp.eml').path,
            filepath.FilePath(deliveryDir).child('message.eml'))
        mr = mimepart.MIMEMessageReceiver(f)
        msg = mr.feedStringNow(source)
        assertMethod(msg)



class PersistenceMixin(MIMEReceiverMixin):
    def _messageTest(self, source, assertMethod):
        mr = self.setUpMailStuff()
        msg = mr.feedStringNow(source)
        assertMethod(msg)



class PersistenceTestCase(unittest.TestCase, MessageTestMixin, PersistenceMixin):
    alternative = PartMaker('multipart/alternative', 'alt',
                    PartMaker('text/plain', 'plain-1'),
                    PartMaker('text/html', 'html-1'),
                    PartMaker('image/jpeg', 'hi')).make()

    def test_readableParts(self):
        """
        Test that L{xquotient.mimestorage.Part.readableParts} returns the
        text/plain and text/html parts inside a multipart/alternative
        """
        def checkReadable(part):
            self.assertEquals(
                list(p.getContentType() for p in part.readableParts()),
                ['text/plain', 'text/html'])
        self._messageTest(self.alternative, checkReadable)



    def assertIndexability(self, msg):
        fi = ixmantissa.IFulltextIndexable(msg.message)
        self.assertEquals(fi.uniqueIdentifier(), unicode(msg.message.storeID))
        # XXX: the following success fixtures are sketchy.  the
        # 'alice@example.com' token appears twice because it is both the
        # 'senderDisplay' and the 'sender' of this message.  That strikes me as
        # wrong; it should eliminate the duplicate, and not rely on the details
        # of the attributes of Message, since ideally those attributes would be
        # accessed through the Correspondent item anyway. --glyph
        self.assertEquals(sorted(fi.textParts()), [
            u'Hello Bob,\n  How are you?\n-A\n',
            u'a test message, comma separated',
            u'alice@example.com alice example com',
            u'bob@example.com bob example com'])
        self.assertEquals(fi.keywordParts(), {
            u'subject': u'a test message, comma separated',
            u'from': u'alice@example.com alice example com',
            u'to': u'bob@example.com bob example com'})
        self.assertEquals(fi.documentType(), msg.message.typeName)


    def test_unknownMultipartTypeAsMixed(self):
        """
        Test that unknown multipart mime types are rendered with 'multipart/mixed'
        content types (with their original content).

          Any "multipart" subtypes that an implementation does not recognize
          must be treated as being of subtype "mixed".

        RFC 2046, Section 5.1.3
        """
        content = PartMaker('multipart/appledouble', 'appledouble',
                            PartMaker('text/applefile', 'applefile'),
                            PartMaker('image/jpeg', 'jpeg')).make()
        part = self.setUpMailStuff().feedStringNow(content)
        walkedPart = part.walkMessage('image/jpeg').next()
        self.assertEquals(walkedPart.type, 'multipart/mixed')
        self.assertEquals(walkedPart.part, part)


    def test_unknownTypeAsOctetStream(self):
        """
        Test that unknown non-multipart mime types are rendered as
        'application/octet-stream'.

        See RFC 2046, Section 4 for more details.
        """
        content = PartMaker('image/applefile', 'applefile').make()
        part = self.setUpMailStuff().feedStringNow(content)
        walkedPart = part.walkMessage(None).next()
        self.assertEquals(walkedPart.type, 'application/octet-stream')
        self.assertEquals(walkedPart.part, part)


    def testIndexability(self):
        """
        Test that the result is adaptable to IFulltextIndexable and the
        resulting object spits out the right data.
        """
        self._messageTest(self.trivialMessage, self.assertIndexability)


    def testAttachmentsAllHaveLength(self):
        def checkAttachments(msgitem):
            for att in msgitem.walkAttachments():
                self.assertNotEquals(att.part.bodyLength, None)
                self.assertNotEquals(att.part.bodyLength, 0)
        self._messageTest(messageWithEmbeddedMessage, checkAttachments)
        self._messageTest(truncatedMultipartMessage, checkAttachments)



    def testRFC822PartLength(self):
        """
        Ensure that message/rfc822 parts have the correct bodyLength.
        """

        def checkAttachments(msgitem):
            for att in msgitem.walkAttachments():