[TST] fetchmail_from_imap_folder: more tests

Author: NL66278

fetchmail_from_imap_folder/match_algorithm/email_exact.py

@@ -19,7 +19,7 @@ def _get_mailaddress_search_domain(
         self, folder, message_dict, operator="=", values=None
     ):
         mailaddresses = values or self._get_mailaddresses(folder, message_dict)
-        if not mailaddresses:
+        if not mailaddresses:  # pragma: no cover
             return [(0, "=", 1)]
         search_domain = (
             (["|"] * (len(mailaddresses) - 1))

fetchmail_from_imap_folder/models/fetchmail_server.py

@@ -70,6 +70,14 @@ def parse_list_response(line):
     object_id = fields.Many2one(required=False)  # comodel_name='ir.model'
     server_type = fields.Selection(default="imap")
 
+    def write(self, vals):
+        """Should reset to draft if server type changes and state not specified."""
+        if "state" not in vals and (
+            "server_type" in vals or "is_ssl" in vals or "object_id" in vals
+        ):
+            vals["state"] = "draft"
+        return super().write(vals)
+
     @api.onchange("server_type", "is_ssl", "object_id")
     def onchange_server_type(self):
         result = super().onchange_server_type()
@@ -79,7 +87,7 @@ def onchange_server_type(self):
     def fetch_mail(self, **kwargs):
         result = True
         for this in self:
-            if not this.folders_only:
+            if not this.folders_only:  # pragma: no cover
                 result = result and super(FetchmailServer, this).fetch_mail(**kwargs)
             this.folder_ids.fetch_mail()
         return result

fetchmail_from_imap_folder/models/fetchmail_server_folder.py

@@ -175,7 +175,7 @@ def fetch_mail(self):
                 this.check_imap_archive_folder(connection)
                 this.retrieve_imap_folder(connection)
                 connection.close()
-            except Exception:
+            except Exception:  # pragma: no cover
                 _logger.error(
                     (
                         "General failure when trying to connect to"
@@ -228,7 +228,7 @@ def retrieve_imap_folder(self, connection):
                 self.env.cr.execute("savepoint apply_matching")
                 self.apply_matching(connection, message_uid)
                 self.env.cr.execute("release savepoint apply_matching")
-            except Exception:
+            except Exception:  # pragma: no cover
                 self.env.cr.execute("rollback to savepoint apply_matching")
                 _logger.exception(
                     "Failed to fetch mail %(message_uid)s from server %(server)s",
@@ -260,9 +260,11 @@ def get_message_uids(self, connection, criteria):
         if result != "OK":
             raise UserError(
                 self.env._(
-                    "Could not search folder %(folder)s on server %(server)s",
+                    "Could not search folder %(folder)s on server %(server)s"
+                    " with criteria %(criteria)s",
                     folder=self.path,
                     server=server.name,
+                    criteria=criteria,
                 )
             )
         _logger.info(
@@ -305,7 +307,8 @@ def run_server_action(self, matched_object_ids):
             return
         records = self.env[self.model_id.model].browse(matched_object_ids)
         for record in records:
-            if not record.exists():
+            if not record.exists():  # pragma: no cover
+                # We have encountered cases where record not existed IRL.
                 continue
             action.with_context(
                 **{
@@ -396,8 +399,6 @@ def _find_match(self, message_dict):
         """Try to find existing object to link mail to."""
         self.ensure_one()
         matcher = self._get_algorithm()
-        if not matcher:
-            return None
         matches = matcher.search_matches(self, message_dict)
         if not matches:
             _logger.info(
@@ -435,12 +436,7 @@ def _get_algorithm(self):
         self.ensure_one()
         if self.match_algorithm == "email_domain":
             return match_algorithm.email_domain.EmailDomain()
-        if self.match_algorithm == "email_exact":
-            return match_algorithm.email_exact.EmailExact()
-        _logger.error(
-            "Unknown algorithm %(algorithm)s", {"algorithm": self.match_algorithm}
-        )
-        return None
+        return match_algorithm.email_exact.EmailExact()
 
     @api.model
     def attach_mail(self, match_object, message_dict):

fetchmail_from_imap_folder/tests/test_match_algorithms.py

@@ -5,8 +5,10 @@
 from unittest.mock import patch
 
 from odoo import fields
+from odoo.exceptions import UserError, ValidationError
 from odoo.fields import Command
 from odoo.tests.common import TransactionCase
+from odoo.tools.mail import email_normalize
 
 from ..match_algorithm import email_domain
 from .common import get_message_body
@@ -17,15 +19,30 @@
 
 
 class MockConnection:
+    _msg_store = {}
+
     def select(self, path=None):
-        return ("OK",)
+        if path in ("INBOX", "customers", "archived_messages"):
+            return ("OK",)
+        return ("NO",)
 
     def create(self, path):
         """Mock creating a folder."""
         return ("OK",)
 
     def store(self, message_uid, msg_item, value):
-        """Mock store command."""
+        """Mock store command.
+
+        Actually simplified. We only store te latest flag set (or unset), in the
+        real IMAP folder there can be multiple flags on a message.
+        """
+        if message_uid not in self._msg_store:
+            self._msg_store[message_uid] = {}
+        msg_key = msg_item[1:]  # Remove + or -
+        if msg_item == "-FLAGS":  # This is to remove flags
+            self._msg_store[message_uid][msg_key] = ""
+        else:
+            self._msg_store[message_uid][msg_key] = value
         return "OK"
 
     def copy(self, message_uid, folder_path):
@@ -38,6 +55,8 @@ def fetch(self, message_uid, parts):
         By passing special values in message_uid, we can manipulate
         the body returned.
         """
+        if message_uid == "deleted_uid":
+            return ("BAD", "")
         email = (
             "the_smart_red_one@reynaerde.waesland"
             if message_uid == "test no match"
@@ -47,11 +66,16 @@ def fetch(self, message_uid, parts):
 
     def search(self, charset, criteria):
         """Return some message uid's."""
+        if criteria == "test_invalid":
+            return ("BAD", [])
         return ("OK", ["123 456"])
 
     def close(self):
         pass
 
+    def logout(self):
+        pass
+
     def expunge(self):
         """Mock an IMAP4.expunge action"""
         return ("OK", None)
@@ -77,18 +101,19 @@ def setUpClass(cls):
                 "category_id": [Command.clear()],
             }
         )
-        cls.server_model = cls.env["fetchmail.server"]
-        cls.folder_model = cls.env["fetchmail.server.folder"]
-        cls.server = cls.server_model.create(
+        cls.FetchmailServer = cls.env["fetchmail.server"]
+        cls.FetchmailServerFolder = cls.env["fetchmail.server.folder"]
+        cls.server = cls.FetchmailServer.create(
             {
                 "name": "Test Fetchmail Server",
                 "server": "imap.example.com",
                 "server_type": "imap",
+                "folders_only": True,  # Not going to test default fetchmail
                 "active": True,
                 "state": "done",
             }
         )
-        cls.folder = cls.folder_model.create(
+        cls.folder = cls.FetchmailServerFolder.create(
             {
                 "server_id": cls.server.id,
                 "sequence": 5,
@@ -100,7 +125,7 @@ def setUpClass(cls):
                 "mail_field": "from",
             }
         )
-        cls.partner_ir_model = cls.env["ir.model"].search(
+        partner_ir_model = cls.env["ir.model"].search(
             [
                 ("model", "=", cls.Partner._name),
             ],
@@ -113,7 +138,30 @@ def setUpClass(cls):
                 "update_path": "category_id",
                 "evaluation_type": "value",
                 "value": str(cls.partner_category.id),
-                "model_id": cls.partner_ir_model.id,
+                "model_id": partner_ir_model.id,
+            }
+        )
+
+    def test_server(self):
+        """Test the server model."""
+        self.server.write({"server_type": "pop"})
+        self.assertEqual(self.server.state, "draft")
+        self._reactivate_server()
+        self.assertEqual(self.server.state, "done")
+        with patch.object(
+            self.server.__class__, "_connect__", return_value=MockConnection()
+        ):
+            self.server.fetch_mail()
+        self.server.onchange_server_type()
+        self.assertEqual(self.server.state, "draft")
+
+    def _reactivate_server(self):
+        """Set type to imap and state to "done"."""
+        self.server.write(
+            {
+                "server_type": "imap",
+                "active": True,
+                "state": "done",
             }
         )
 
@@ -159,6 +207,19 @@ def test_apply_matching_exact_no_match(self):
         message_uid = "test no match"
         folder.apply_matching(connection, message_uid)
 
+    def test_apply_matching_odoo_standard(self):
+        folder = self.folder
+        folder.match_algorithm = "odoo_standard"
+        connection = MockConnection()
+        message_uid = "test no match"
+        thread_id = folder.apply_matching(connection, message_uid)
+        self.assertTrue(thread_id)  # Odoo standard either updates of creates record.
+        partner = self.Partner.browse(thread_id)
+        self.assertTrue(partner)
+        self.assertEqual(
+            email_normalize(partner.email), "the_smart_red_one@reynaerde.waesland"
+        )
+
     def test_retrieve_imap_folder_domain(self):
         folder = self.folder
         folder.match_algorithm = "email_domain"
@@ -169,7 +230,20 @@ def test_archive_messages(self):
         folder = self.folder
         folder.archive_path = "archived_messages"
         connection = MockConnection()
-        folder.retrieve_imap_folder(connection)
+        folder.check_imap_archive_folder(connection)
+        # Apply matching should succeed with no error.
+        folder.match_algorithm = "email_exact"
+        message_uid = "<485a8041-d560-a981-5afc-d31c1f136748@acme.com>"
+        folder.apply_matching(connection, message_uid)
+        # Should fail on not existing folder that is not autocreated.
+        with self.assertRaises(UserError):
+            folder.archive_path = "will_not_be_created"
+            folder.check_imap_archive_folder(connection)
+        # Should not fail if no archive path.
+        folder.archive_path = False
+        folder.check_imap_archive_folder(connection)
+        # Archiving should simply not fail.
+        folder._archive_msg(connection, "does not matter")
 
     def test_non_action(self):
         connection = MockConnection()
@@ -204,13 +278,86 @@ def test_button_confirm_folder(self):
         with patch.object(
             self.server.__class__, "_connect__", return_value=MockConnection()
         ):
+            # Inactive folders are not set to done
             folder.active = False
             folder.button_confirm_folder()
             self.assertEqual(folder.state, "draft")
+            # Active folders are set to done
+            folder.active = True
+            folder.button_confirm_folder()
+            self.assertEqual(folder.state, "done")
+            # Should fail on unknown folder.
+            with self.assertRaises(ValidationError):
+                folder.path = "invalid_folder"
+                folder.button_confirm_folder()
 
+    def test_fetchmail(self):
+        """Test the overall process."""
+        folder = self.folder
+        with patch.object(
+            self.server.__class__, "_connect__", return_value=MockConnection()
+        ):
+            # Should not result in error if not active or in draft.
+            folder.active = False
+            folder.button_confirm_folder()
+            self.assertEqual(folder.state, "draft")
+            folder.fetch_mail()
+            # Should also not result in error if active.
             folder.active = True
             folder.button_confirm_folder()
             self.assertEqual(folder.state, "done")
+            folder.fetch_mail()
+
+    def test_fetch_msg_exception(self):
+        """There can be an exception, for example if message no longer exists."""
+        folder = self.folder
+        connection = MockConnection()
+        with self.assertRaises(UserError):
+            folder.fetch_msg(connection, "deleted_uid")
+
+    def test_get_message_uids(self):
+        """Test failure on invalid criteria."""
+        folder = self.folder
+        connection = MockConnection()
+        criteria = folder.get_criteria()
+        message_uids = folder.get_message_uids(connection, criteria)
+        self.assertEqual(message_uids, ["123 456"])
+        folder.path = "invalid_folder"
+        with self.assertRaises(UserError):
+            folder.get_message_uids(connection, criteria)
+        folder.path = "INBOX"  # Restore valid folder path.
+        criteria = "test_invalid"
+        with self.assertRaises(UserError):
+            folder.get_message_uids(connection, criteria)
+
+    def test_update_msg(self):
+        """Message update depends on folder configuration."""
+        folder = self.folder
+        connection = MockConnection()
+        folder.seen = "matching"
+        folder.update_msg(connection, "123", matched=True)
+        self._check_flags(connection, "123", "\\seen")
+        folder.seen = "on_fetch"  # back to default
+        folder.delete_matching = True
+        folder.update_msg(connection, "456", matched=True)
+        self._check_flags(connection, "456", "\\DELETED")
+        folder.delete_matching = False  # back to default
+        self.flag_nonmatching = True
+        folder.update_msg(connection, "789", matched=True, flagged=True)
+        self._check_flags(connection, "789", "")
+
+    def _check_flags(self, connection, uid, expected_flags):
+        """Check whether expected value stored."""
+        self.assertIn(uid, connection._msg_store)
+        self.assertIn("FLAGS", connection._msg_store[uid])
+        self.assertEqual(connection._msg_store[uid]["FLAGS"], expected_flags)
+
+    def test_button_attach_mail_manually(self):
+        """The button should return the an action dictionary."""
+        folder = self.folder
+        action_dict = folder.button_attach_mail_manually()
+        Wizard = self.env[action_dict["res_model"]]
+        self.assertEqual(Wizard._name, "fetchmail.attach.mail.manually")
 
     def test_set_draft(self):
         """Test the set_draft method."""