Switch to fully oriented-object with DecodedMail object

tracim/tracim/lib/email_fetcher.py

@@ -19,101 +19,102 @@ from tracim.controllers.events import VALID_TOKEN_VALUE
 TRACIM_SPECIAL_KEY_HEADER = "X-Tracim-Key"
 
 
-def str_header(header: Header) -> str:
-    return str(make_header(decode_header(header)))
-
-
-def decode_mail(msg: Message)-> dict:
-    """
-    Get useful header and body content and decode from Message
-    :param msg:
-    :return:
-    """
-    mail_data = {}
-
-    try:
-        mail_data['subject'] = str_header(msg['subject'])
-        mail_data['from'] = parseaddr(msg['From'])[1]
-        # Reply key
-        mail_data['to'] = parseaddr(msg['To'])[1]
-        # INFO - G.M - 2017-11-15
-        #  We only need to save the first/oldest addr of references
-        mail_data['references'] = parseaddr(msg['References'])[1]
-        if TRACIM_SPECIAL_KEY_HEADER in msg:
-            mail_data[TRACIM_SPECIAL_KEY_HEADER] = str_header(msg[TRACIM_SPECIAL_KEY_HEADER])  # nopep8
-
-    except Exception:
-        # FIXME - G.M - 2017-11-15 - handle exceptions correctly
-        return {}
-
-    # TODO - G.M - 2017-11-15 - Parse properly HTML (and text ?) body
-    body = get_body_mime_part(msg)
-    if body:
-        charset = body.get_content_charset('iso-8859-1')
-        ctype = body.get_content_type()
-        if ctype == "text/plain":
-            mail_data['body'] = body.get_payload(decode=True).decode(charset)
-        elif ctype == "text/html":
-            mail_data['body'] = body.get_payload(decode=True).decode(charset)
+class DecodedMail(object):
+
+    def __init__(self, message: Message):
+        self._message = message
+
+    def _decode_header(self, header_title: str) -> typing.Optional[str]:
+        # FIXME : Handle exception
+        if header_title in self._message:
+            return str(make_header(decode_header(header)))
         else:
-            pass
-    else:
-        pass
-
-    return mail_data
-
-
-def get_body_mime_part(msg) -> Message:
-    # FIXME - G.M - 2017-11-16 - Use stdlib msg.get_body feature for py3.6+
-    # FIXME - G.M - 2017-11-16 - Check support for non-multipart mail
-    #assert msg.is_multipart()
-    part = None
-    # Check for html
-    for part in msg.walk():
-        ctype = part.get_content_type()
-        cdispo = str(part.get('Content-Disposition'))
-        if ctype == 'text/html' and 'attachment' not in cdispo:
-            return part
-    # checj fir plain text
-    for part in msg.walk():
-        ctype = part.get_content_type()
-        cdispo = str(part.get('Content-Disposition'))
-        if ctype == 'text/plain' and 'attachment' not in cdispo:
-            return part
-    return part
-
-
-def get_tracim_content_key(mail_data: dict) -> typing.Optional[str]:
-
-    """ Link mail_data dict to tracim content
-    First try checking special header, them check 'to' header
-    and finally check first(oldest) mail-id of 'references' header
-    """
-    key = None
-    if TRACIM_SPECIAL_KEY_HEADER in mail_data:
-        key = mail_data[TRACIM_SPECIAL_KEY_HEADER]
-    if key is None and 'to' in mail_data:
-        key = find_key_from_mail_adress(mail_data['to'])
-    if key is None and 'references' in mail_data:
-        mail_adress = mail_data['references']
-        key = find_key_from_mail_adress(mail_adress)
-    return key
-
-
-def find_key_from_mail_adress(mail_address: str) -> typing.Optional[str]:
-    """ Parse mail_adress-like string
-    to retrieve key.
-
-    :param mail_address: user+key@something like string
-    :return: key
-    """
-    username = mail_address.split('@')[0]
-    username_data = username.split('+')
-    if len(username_data) == 2:
-        key = username_data[1]
-    else:
+            return None
+
+    def get_subject(self) -> typing.Optional[str]:
+        return self._decode_header('subject')
+
+    def get_from_address(self) -> typing.Optional[str]:
+        return parseaddr(self._message['From'])[1]
+
+    def get_to_address(self)-> typing.Optional[str]:
+        return parseaddr(self._message['To'])[1]
+
+    def get_first_ref(self) -> typing.Optional[str]:
+        return parseaddr(self._message['References'])[1]
+
+    def get_special_key(self) -> typing.Optional[str]:
+        return self._decode_header(TRACIM_SPECIAL_KEY_HEADER)
+
+    def get_body(self) -> typing.Optional[str]:
+        body_part = self._get_mime_body_message()
+        body = None
+        if body_part:
+            charset = body_part.get_content_charset('iso-8859-1')
+            ctype = body_part.get_content_type()
+            if ctype == "text/plain":
+                body = body_part.get_payload(decode=True).decode(
+                    charset)
+            elif ctype == "text/html":
+                body = body_part.get_payload(decode=True).decode(
+                    charset)
+        return body
+
+    def _get_mime_body_message(self) -> typing.Optional[Message]:
+        # FIXME - G.M - 2017-11-16 - Use stdlib msg.get_body feature for py3.6+
+        # FIXME - G.M - 2017-11-16 - Check support for non-multipart mail
+        # assert msg.is_multipart()
+        part = None
+        # Check for html
+        for part in self._message.walk():
+            ctype = part.get_content_type()
+            cdispo = str(part.get('Content-Disposition'))
+            if ctype == 'text/html' and 'attachment' not in cdispo:
+                return part
+        # checj fir plain text
+        for part in self._message.walk():
+            ctype = part.get_content_type()
+            cdispo = str(part.get('Content-Disposition'))
+            if ctype == 'text/plain' and 'attachment' not in cdispo:
+                return part
+        return part
+
+    def get_key(self) -> typing.Optional[str]:
+
+        """
+        First try checking special header, them check 'to' header
+        and finally check first(oldest) mail-id of 'references' header
+        """
         key = None
-    return key
+        first_ref = self.get_first_ref()
+        to_address = self.get_to_address()
+        special_key = self.get_special_key()
+
+        if special_key:
+            key = special_key
+        if not key and to_address:
+            key = DecodedMail.find_key_from_mail_address(to_address)
+        if not key and first_ref:
+            key = DecodedMail.find_key_from_mail_address(first_ref)
+
+        return key
+
+    @staticmethod
+    def find_key_from_mail_address(mail_address: str) \
+            -> typing.Optional[str]:
+        """ Parse mail_adress-like string
+        to retrieve key.
+
+        :param mail_address: user+key@something like string
+        :return: key
+        """
+        username = mail_address.split('@')[0]
+        username_data = username.split('+')
+        if len(username_data) == 2:
+            key = username_data[1]
+        else:
+            key = None
+        return key
 
 
 class MailFetcher(object):
@@ -196,7 +197,8 @@ class MailFetcher(object):
                     rv, data = self._connection.fetch(num, '(RFC822)')
                     if rv == 'OK':
                         msg = message_from_bytes(data[0][1])
-                        self._mails.append(msg)
+                        decodedmsg = DecodedMail(msg)
+                        self._mails.append(decodedmsg)
                     else:
                         log = 'IMAP : Unable to get mail : {}'
                         logger.debug(self, log.format(str(rv)))
@@ -210,12 +212,11 @@ class MailFetcher(object):
     def _notify_tracim(self) -> None:
         while self._mails:
             mail = self._mails.pop()
-            decoded_mail = decode_mail(mail)
             msg = {"token": VALID_TOKEN_VALUE,
-                   "user_mail": decoded_mail['from'],
-                   "content_id": get_tracim_content_key(decoded_mail),
+                   "user_mail": mail.get_from_address(),
+                   "content_id": mail.get_key(),
                    "payload": {
-                       "content": decoded_mail['body'],
+                       "content": mail.get_body(),
                    }}
             # FIXME - G.M - 2017-11-15 - Catch exception from http request
             requests.post(self.endpoint, json=msg)

tracim/tracim/lib/test_email_fetcher.py

@@ -1,88 +1,11 @@
-from email_fetcher import decode_mail, get_tracim_content_key,\
-    TRACIM_SPECIAL_KEY_HEADER,find_key_from_mail_adress
-from email.mime.multipart import MIMEMultipart
-from email.utils import parsedate_tz,mktime_tz
-import datetime
-
-# decode_mail
-
-def test_decode_mail_ok():
-    msg = MIMEMultipart()
-    msg['From'] = 'a@home'
-    msg['To'] = 'b@home'
-    msg['Subject'] = "test"
-    #msg.add_header('Reply-To', '<a+key@home>')
-    msg.add_header('References', '<reply+key@home>')
-    msg.add_header(TRACIM_SPECIAL_KEY_HEADER, 'key')
-    maildata=decode_mail(msg)
-
-    assert maildata == {
-        TRACIM_SPECIAL_KEY_HEADER: 'key',
-        'from': 'a@home',
-        'to': 'b@home',
-        'subject':'test',
-        'references':'reply+key@home',
-    }
-# get_tracim_content_key
-
-def test_get_tracim_content_key_empty():
-    mail_data={}
-    assert get_tracim_content_key(mail_data) == None
-
-def test_get_tracim_content_key_no_key():
-    mail_data={
-        'to':'a@b',
-        'references':'<a@b> <b@c>'
-    }
-    assert get_tracim_content_key(mail_data) == None
-
-def test_get_tracim_content_key_special_key():
-    mail_data={
-        'to':'a@b',
-        'references':'<a@b> <b@c>',
-        TRACIM_SPECIAL_KEY_HEADER : 'key'
-    }
-    assert get_tracim_content_key(mail_data) == 'key'
-
-
-def test_get_tracim_content_key_to_key():
-    mail_data={
-        'to':'a+key@b',
-        'references':'<a@b> <b@c>',
-    }
-    assert get_tracim_content_key(mail_data) == 'key'
-
-def test_get_tracim_content_key_references_key():
-    mail_data={
-        'to':'a@b',
-        'references':'<a+key@b> <b@c>',
-    }
-    assert get_tracim_content_key(mail_data) == 'key'
-
-def test_get_tracim_content_key_order():
-    mail_data={
-        'to':'a+2@b',
-        'references':'<a+3@b> <b@c>',
-        TRACIM_SPECIAL_KEY_HEADER: '1'
-    }
-    assert get_tracim_content_key(mail_data) == '1'
-    mail_data={
-        'to':'a+2@b',
-        'references':'<a+3@b> <b@c>',
-    }
-    assert get_tracim_content_key(mail_data) == '2'
-
-    mail_data={
-        'references':'<a+3@b> <b@c>',
-    }
-    assert get_tracim_content_key(mail_data) == '3'
+from email_fetcher import DecodedMail
 
 # find_key_from_mail_address
 
 def test_find_key_from_mail_address_no_key():
-    mail_adress="a@b"
-    assert find_key_from_mail_adress(mail_adress) == None
+    mail_address="a@b"
+    assert DecodedMail.find_key_from_mail_address(mail_address) == None
 
 def test_find_key_from_mail_adress_key():
     mail_address="a+key@b"
-    assert find_key_from_mail_adress(mail_address) == 'key'
+    assert DecodedMail.find_key_from_mail_address(mail_address) == 'key'