test_queue.py

# © 2022 Le Filament (<http://www.le-filament.com>)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).

import logging

from odoo.tests import common, tagged

_logger = logging.getLogger(__name__)


@tagged("post_install", "-at_install")
class TestDrupalQueue(common.TransactionCase):
    def test_contact(self):
        Queue = self.env["drupal.job.queue"]

        _logger.info("\nCréation Contact \n----")
        contact = self.env["res.partner"].create(
            {
                "name": "Contact Test",
                "is_company": False,
                "is_cooperative": False,
            }
        )
        _logger.info("Contact " + contact.name + " créée, id = " + str(contact.id))
        queue_item = Queue.search([("partner_id", "=", contact.id)])
        self.assertEqual(len(queue_item), 0)

        _logger.info("\nAjout Tel \n----")
        contact.write({"phone": "00 00 00 00 00"})
        queue_item = Queue.search([("partner_id", "=", contact.id)])
        self.assertEqual(len(queue_item), 0)

        _logger.info("\nActivation extranet \n----")
        contact.write({"extranet_access": True})
        queue_item = Queue.search([("partner_id", "=", contact.id)])
        for i in queue_item:
            _logger.info(i.partner_id.name + " | " + i.action + " | " + i.status)
        self.assertEqual(len(queue_item), 1)

        _logger.info("\nModification email \n----")
        contact.write({"email": "contacttest@coop.com"})
        queue_item = Queue.search([("partner_id", "=", contact.id)])
        for i in queue_item:
            _logger.info(i.partner_id.name + " | " + i.action + " | " + i.status)
        self.assertEqual(len(Queue.search([("partner_id", "=", contact.id)])), 1)

        _logger.info("\nModification statut Queue Item puis modification email\n----")
        queue_item = Queue.search([("partner_id", "=", contact.id)])
        queue_item.write({"status": "done"})
        contact.write({"email": "contacttest2@coop.com"})
        queue_item = Queue.search([("partner_id", "=", contact.id)])
        for i in queue_item:
            _logger.info(i.partner_id.name + " | " + i.action + " | " + i.status)
        self.assertEqual(len(Queue.search([("partner_id", "=", contact.id)])), 2)

        _logger.info("\nArchivage contact\n----")
        contact.write({"active": False})
        queue_item = Queue.search([("partner_id", "=", contact.id)])
        for i in queue_item:
            _logger.info(i.partner_id.name + " | " + i.action + " | " + i.status)
        self.assertEqual(len(Queue.search([("partner_id", "=", contact.id)])), 3)

    def test_coop(self):
        Queue = self.env["drupal.job.queue"]

        _logger.info("\nCréation Coop \n----")
        coop = self.env["res.partner"].create(
            {
                "name": "Coop Test",
                "is_company": True,
                "is_cooperative": True,
            }
        )
        _logger.info("Coopérative " + coop.name + " créée, id = " + str(coop.id))
        queue_item = Queue.search([("partner_id", "=", coop.id)])
        self.assertEqual(len(queue_item), 0)

        _logger.info("\nPassage statut suivi \n----")
        coop.write({"project_status": "4_suivi"})
        queue_item = Queue.search([("partner_id", "=", coop.id)])
        self.assertEqual(len(queue_item), 0)

        _logger.info("\nActivation extranet \n----")
        coop.write({"extranet_update_company": True})
        queue_item = Queue.search([("partner_id", "=", coop.id)])
        for i in queue_item:
            _logger.info(i.partner_id.name + " | " + i.action + " | " + i.status)
        self.assertEqual(len(queue_item), 1)

        _logger.info("\nModification email \n----")
        coop.write({"email": "cooptest@coop.com"})
        queue_item = Queue.search([("partner_id", "=", coop.id)])
        for i in queue_item:
            _logger.info(i.partner_id.name + " | " + i.action + " | " + i.status)
        self.assertEqual(len(Queue.search([("partner_id", "=", coop.id)])), 1)

        _logger.info("\nModification statut Queue Item puis modification email\n----")
        queue_item = Queue.search([("partner_id", "=", coop.id)])
        queue_item.write({"status": "done"})
        coop.write({"email": "cooptest2@coop.com"})
        queue_item = Queue.search([("partner_id", "=", coop.id)])
        for i in queue_item:
            _logger.info(i.partner_id.name + " | " + i.action + " | " + i.status)
        self.assertEqual(len(Queue.search([("partner_id", "=", coop.id)])), 2)

        _logger.info("\nArchivage coop\n----")
        coop.write({"active": False})
        queue_item = Queue.search([("partner_id", "=", coop.id)])
        for i in queue_item:
            _logger.info(i.partner_id.name + " | " + i.action + " | " + i.status)
        self.assertEqual(len(Queue.search([("partner_id", "=", coop.id)])), 3)