queue/queue_job_subscribe/tests/test_job_subscribe.py at 5243fcbdffcd99adc2020cd72947a953dc06b980 · OCA/queue · 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
# Copyright 2016 Cédric Pigeon
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).

import odoo.tests.common as common

from odoo.addons.queue_job.job import Job


class TestJobSubscribe(common.TransactionCase):
    def setUp(self):
        super().setUp()
        self.env = self.env(context=dict(self.env.context, tracking_disable=True))
        grp_queue_job_manager = self.ref("queue_job.group_queue_job_manager")
        self.other_partner_a = self.env["res.partner"].create(
            {"name": "My Company a", "is_company": True, "email": "test@tes.ttest"}
        )
        self.other_user_a = (
            self.env["res.users"]
            .with_context(no_reset_password=True)
            .create(
                {
                    "partner_id": self.other_partner_a.id,
                    "login": "my_login a",
                    "name": "my user",
                    "group_ids": [(4, grp_queue_job_manager)],
                }
            )
        )
        self.other_partner_b = self.env["res.partner"].create(
            {"name": "My Company b", "is_company": True, "email": "test@tes.ttest"}
        )
        self.other_user_b = self.env["res.users"].create(
            {
                "partner_id": self.other_partner_b.id,
                "login": "my_login_b",
                "name": "my user 1",
                "group_ids": [(4, grp_queue_job_manager)],
            }
        )

    def _create_failed_job(self):
        method = self.env["res.users"].with_user(self.other_user_a).mapped
        test_job = Job(method)
        test_job.store()
        test_job_record = self.env["queue.job"].search([("uuid", "=", test_job.uuid)])
        test_job_record.write({"state": "failed"})
        return test_job_record

    def test_job_subscription_all_follow(self):
        """
        When a job is created, all user of group
        queue_job.group_queue_job_manager are automatically set as
        follower
        """

        #################################
        # Test 1: All users are followers
        #################################
        stored = self._create_failed_job()
        users = self.env["res.users"].search(
            [("group_ids", "=", self.ref("queue_job.group_queue_job_manager"))]
        )
        self.assertEqual(len(stored.message_follower_ids), len(users))
        expected_partners = [u.partner_id for u in users]
        self.assertSetEqual(
            set(stored.mapped("message_follower_ids.partner_id")),
            set(expected_partners),
        )
        followers_id = [f.id for f in stored.mapped("message_follower_ids.partner_id")]
        self.assertIn(self.other_partner_a.id, followers_id)
        self.assertIn(self.other_partner_b.id, followers_id)

    def test_job_subscription_not_follow(self):
        """
        If the flag subscribe_job is not set as follower
        """
        ###########################################
        # Test 2: User b request to not be follower
        ###########################################
        self.other_user_b.write({"subscribe_job": False})
        stored = self._create_failed_job()
        users = self.env["res.users"].search(
            [
                ("group_ids", "=", self.ref("queue_job.group_queue_job_manager")),
                ("subscribe_job", "=", True),
            ]
        )
        self.assertEqual(len(stored.message_follower_ids), len(users))
        expected_partners = [u.partner_id for u in users]
        self.assertSetEqual(
            set(stored.mapped("message_follower_ids.partner_id")),
            set(expected_partners),
        )
        followers_id = [f.id for f in stored.mapped("message_follower_ids.partner_id")]
        self.assertIn(self.other_partner_a.id, followers_id)
        self.assertNotIn(self.other_partner_b.id, followers_id)