mail/fetchmail_mail_activity_team_activity/tests/test_activity.py at dea97b31f803b4dd42522617accf4a61af59204f · OCA/mail · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).

import logging
import uuid

from odoo.fields import Command
from odoo.tests import TransactionCase

_logger = logging.getLogger(__name__)


class TestActivity(TransactionCase):
    def setUp(self):
        super().setUp()
        # Clean up any existing teams before each test
        self.env["mail.activity.team"].search([]).unlink()

    def generate_reply(self, message):
        return (
            "From: ext.partner@example.org\r\n"
            "To: odoo.test@local\r\n"
            f"Message-ID: {uuid.uuid4()}\r\n"
            f"In-Reply-To: {message['message_id']}\r\n"
            "Subject: Test Email Reply\r\n\r\n"
        )

    def test_no_team(self):
        """Test that no activity is created when no team exists"""
        # Ensure no teams exist
        teams = self.env["mail.activity.team"].search([])
        self.assertEqual(len(teams), 0, "Should have no teams before test")

        thread = self.env.user.partner_id
        msg = thread.message_notify(
            partner_ids=self.env.user.partner_id.ids,
            body="Testing",
        )

        # Record initial state
        initial_activities = thread.activity_ids
        initial_message_count = len(thread.message_ids)

        # Process email reply
        reply = self.generate_reply(msg)
        thread_id = thread.message_process(thread._name, reply)

        # Assertions
        self.assertEqual(thread_id, thread.id)
        self.assertTrue(len(thread.message_ids) > initial_message_count)

        # CRITICAL: No new activity should be created
        self.assertEqual(
            thread.activity_ids.ids,
            initial_activities.ids,
            "No activity should be created when no team exists",
        )

    def test_with_team(self):
        thread = self.env.user.partner_id
        model = self.env.ref("base.model_res_partner")

        self.env["mail.activity.team"].create(
            {
                "name": "Testing Team",
                "res_model_ids": [Command.set(model.ids)],
                "user_id": self.env.user.id,
            }
        )
        msg = thread.message_notify(
            partner_ids=self.env.user.partner_id.ids,
            body="Testing",
        )

        messages = thread.message_ids
        activities = thread.activity_ids
        reply = self.generate_reply(msg)

        thread_id = thread.message_process(thread._name, reply)
        self.assertTrue(len(messages) < len(thread.message_ids))
        self.assertEqual(thread_id, thread.id)

        activity = thread.activity_ids - activities
        self.assertTrue(activity)
        self.assertEqual(
            activity.activity_type_id,
            self.env.ref("mail.mail_activity_data_email"),
        )

        # Schedule it only once
        activities = thread.activity_ids
        reply = self.generate_reply(msg)
        thread.message_process(thread._name, reply)
        self.assertEqual(activities, thread.activity_ids)

    def test_with_team_no_activity_mixin(self):
        thread = self.env["calendar.event"].create({"name": "Testing"})
        model = self.env.ref("calendar.model_calendar_event")
        self.env["mail.activity.team"].create(
            {
                "name": "Testing Team",
                "res_model_ids": [Command.set(model.ids)],
                "user_id": self.env.user.id,
            }
        )
        msg = thread.message_notify(
            partner_ids=self.env.user.partner_id.ids,
            body="Testing",
        )

        activities = thread.activity_ids
        reply = self.generate_reply(msg)
        thread_id = thread.message_process(thread._name, reply)
        self.assertEqual(activities, thread.activity_ids)
        self.assertEqual(thread_id, thread.id)