Odoo18-Base/addons/test_event_full/tests/test_event_mail.py
2025-01-06 10:57:38 +07:00

330 lines
16 KiB
Python

# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from datetime import datetime, timedelta
from unittest.mock import patch
from odoo.addons.test_event_full.tests.common import TestEventFullCommon, TestEventMailCommon
from odoo.tests import tagged, users
from odoo.tools import formataddr
@tagged('event_mail', 'post_install', '-at_install')
class TestEventMailInternals(TestEventMailCommon):
def test_template_ref_delete_lines(self):
""" When deleting a template, related lines should be deleted too """
event_type = self.env['event.type'].create({
'name': 'Event Type',
'default_timezone': 'Europe/Brussels',
'event_type_mail_ids': [
(0, 0, {
'interval_unit': 'now',
'interval_type': 'after_sub',
'template_ref': 'mail.template,%i' % self.env['ir.model.data']._xmlid_to_res_id('event.event_subscription')}),
(0, 0, {
'interval_unit': 'now',
'interval_type': 'after_sub',
'notification_type': 'sms',
'template_ref': 'sms.template,%i' % self.env['ir.model.data']._xmlid_to_res_id('event_sms.sms_template_data_event_registration')}),
],
})
template_mail = event_type.event_type_mail_ids[0].template_ref
template_sms = event_type.event_type_mail_ids[1].template_ref
event = self.env['event.event'].create({
'name': 'event mail template removed',
'event_type_id': event_type.id,
'date_begin': datetime(2020, 2, 1, 8, 30, 0),
'date_end': datetime(2020, 2, 4, 18, 45, 0),
'date_tz': 'Europe/Brussels',
})
self.assertEqual(len(event_type.event_type_mail_ids), 2)
self.assertEqual(len(event.event_mail_ids), 2)
template_mail.unlink()
self.assertEqual(len(event_type.event_type_mail_ids.exists()), 1)
self.assertEqual(len(event.event_mail_ids.exists()), 1)
template_sms.unlink()
self.assertEqual(len(event_type.event_type_mail_ids.exists()), 0)
self.assertEqual(len(event.event_mail_ids.exists()), 0)
@tagged('event_mail', 'post_install', '-at_install')
class TestEventMailSchedule(TestEventMailCommon):
def test_event_mail_before_trigger_sent_count(self):
""" Emails are only sent to confirmed attendees. """
test_event = self.test_event
mail_schedulers = test_event.event_mail_ids
self.assertEqual(len(mail_schedulers), 6)
before = mail_schedulers.filtered(lambda m: m.interval_type == "before_event" and m.interval_unit == "days")
self.assertEqual(len(before), 2)
# Add registrations
_dummy, _dummy, open_reg, done_reg = self.env['event.registration'].create([{
'event_id': test_event.id,
'name': 'RegistrationUnconfirmed',
'email': 'Registration@Unconfirmed.com',
'phone': '1',
'state': 'draft',
}, {
'event_id': test_event.id,
'name': 'RegistrationCanceled',
'email': 'Registration@Canceled.com',
'phone': '2',
'state': 'cancel',
}, {
'event_id': test_event.id,
'name': 'RegistrationConfirmed',
'email': 'Registration@Confirmed.com',
'phone': '3',
'state': 'open',
}, {
'event_id': test_event.id,
'name': 'RegistrationDone',
'email': 'Registration@Done.com',
'phone': '4',
'state': 'done',
}])
with self.mock_datetime_and_now(self.event_date_begin - timedelta(days=2)), \
self.mock_mail_gateway(), \
self.mockSMSGateway():
before.execute()
for registration in open_reg, done_reg:
with self.subTest(registration_state=registration.state, medium='mail'):
self.assertMailMailWEmails(
[formataddr((registration.name, registration.email))],
'outgoing',
)
with self.subTest(registration_state=registration.state, medium='sms'):
self.assertSMS(
self.env['res.partner'],
registration.phone,
None,
)
self.assertEqual(len(self._new_mails), 2, 'Mails should not be sent to draft or cancel registrations')
self.assertEqual(len(self._new_sms), 2, 'SMS should not be sent to draft or cancel registrations')
self.assertEqual(test_event.seats_taken, 2, 'Wrong number of seats_taken')
for scheduler in before:
self.assertEqual(
scheduler.mail_count_done, 2,
'Wrong Emails Sent Count! Probably emails sent to unconfirmed attendees were not included into the Sent Count'
)
@users('user_eventmanager')
def test_schedule_event_scalability(self):
""" Test scalability / iterative work on event-based schedulers """
test_event = self.env['event.event'].browse(self.test_event.ids)
registrations = self._create_registrations(test_event, 30)
registrations = registrations.sorted("id")
# check event-based schedulers
after_mail = test_event.event_mail_ids.filtered(lambda s: s.interval_type == "after_event" and s.notification_type == "mail")
self.assertEqual(len(after_mail), 1)
self.assertEqual(after_mail.mail_count_done, 0)
self.assertFalse(after_mail.mail_done)
after_sms = test_event.event_mail_ids.filtered(lambda s: s.interval_type == "after_event" and s.notification_type == "sms")
self.assertEqual(len(after_sms), 1)
self.assertEqual(after_sms.mail_count_done, 0)
self.assertFalse(after_sms.mail_done)
before_mail = test_event.event_mail_ids.filtered(lambda s: s.interval_type == "before_event" and s.notification_type == "mail")
self.assertEqual(len(before_mail), 1)
self.assertEqual(before_mail.mail_count_done, 0)
self.assertFalse(before_mail.mail_done)
before_sms = test_event.event_mail_ids.filtered(lambda s: s.interval_type == "before_event" and s.notification_type == "sms")
self.assertEqual(len(before_sms), 1)
self.assertEqual(before_sms.mail_count_done, 0)
self.assertFalse(before_sms.mail_done)
# setup batch and cron limit sizes to check iterative behavior
batch_size, cron_limit = 5, 20
self.env["ir.config_parameter"].sudo().set_param("mail.batch_size", batch_size)
self.env["ir.config_parameter"].sudo().set_param("mail.render.cron.limit", cron_limit)
# launch before event schedulers -> all communications are sent
current_now = self.event_date_begin - timedelta(days=1)
EventMail = type(self.env['event.mail'])
exec_origin = EventMail._execute_event_based_for_registrations
with patch.object(
EventMail, '_execute_event_based_for_registrations', autospec=True, wraps=EventMail, side_effect=exec_origin,
) as mock_exec, \
self.mock_datetime_and_now(current_now), \
self.mockSMSGateway(), \
self.mock_mail_gateway(), \
self.capture_triggers('event.event_mail_scheduler') as capture:
self.event_cron_id.method_direct_trigger()
self.assertFalse(after_mail.last_registration_id)
self.assertEqual(after_mail.mail_count_done, 0)
self.assertFalse(after_mail.mail_done)
self.assertFalse(after_sms.last_registration_id)
self.assertEqual(after_sms.mail_count_done, 0)
self.assertFalse(after_sms.mail_done)
# iterative work on registrations: only 20 (cron limit) are taken into account
self.assertEqual(before_mail.last_registration_id, registrations[19])
self.assertEqual(before_mail.mail_count_done, 20)
self.assertFalse(before_mail.mail_done)
self.assertEqual(before_sms.last_registration_id, registrations[19])
self.assertEqual(before_sms.mail_count_done, 20)
self.assertFalse(before_sms.mail_done)
self.assertEqual(mock_exec.call_count, 8, "Batch of 5 to make 20 registrations: 4 calls / scheduler")
# cron should have been triggered for the remaining registrations
self.assertSchedulerCronTriggers(capture, [current_now] * 2)
# relaunch to close scheduler
with self.mock_datetime_and_now(current_now), \
self.mockSMSGateway(), \
self.mock_mail_gateway(), \
self.capture_triggers('event.event_mail_scheduler') as capture:
self.event_cron_id.method_direct_trigger()
self.assertEqual(before_mail.last_registration_id, registrations[-1])
self.assertEqual(before_mail.mail_count_done, 30)
self.assertTrue(before_mail.mail_done)
self.assertEqual(before_sms.last_registration_id, registrations[-1])
self.assertEqual(before_sms.mail_count_done, 30)
self.assertTrue(before_sms.mail_done)
self.assertFalse(capture.records)
# launch after event schedulers -> all communications are sent
current_now = self.event_date_end + timedelta(hours=1)
with self.mock_datetime_and_now(current_now), \
self.mockSMSGateway(), \
self.mock_mail_gateway(), \
self.capture_triggers('event.event_mail_scheduler') as capture:
self.event_cron_id.method_direct_trigger()
# iterative work on registrations: only 20 (cron limit) are taken into account
self.assertEqual(after_mail.last_registration_id, registrations[19])
self.assertEqual(after_mail.mail_count_done, 20)
self.assertFalse(after_mail.mail_done)
self.assertEqual(after_sms.last_registration_id, registrations[19])
self.assertEqual(after_sms.mail_count_done, 20)
self.assertFalse(after_sms.mail_done)
self.assertEqual(mock_exec.call_count, 8, "Batch of 5 to make 20 registrations: 4 calls / scheduler")
# cron should have been triggered for the remaining registrations
self.assertSchedulerCronTriggers(capture, [current_now] * 2)
# relaunch to close scheduler
with self.mock_datetime_and_now(current_now), \
self.mockSMSGateway(), \
self.mock_mail_gateway(), \
self.capture_triggers('event.event_mail_scheduler') as capture:
self.event_cron_id.method_direct_trigger()
self.assertEqual(after_mail.last_registration_id, registrations[-1])
self.assertEqual(after_mail.mail_count_done, 30)
self.assertTrue(after_mail.mail_done)
self.assertEqual(after_sms.last_registration_id, registrations[-1])
self.assertEqual(after_sms.mail_count_done, 30)
self.assertTrue(after_sms.mail_done)
self.assertFalse(capture.records)
@users('user_eventmanager')
def test_schedule_subscription_scalability(self):
""" Test scalability / iterative work on subscription-based schedulers """
test_event = self.env['event.event'].browse(self.test_event.ids)
sub_mail = test_event.event_mail_ids.filtered(lambda s: s.interval_type == "after_sub" and s.interval_unit == "now" and s.notification_type == "mail")
self.assertEqual(len(sub_mail), 1)
self.assertEqual(sub_mail.mail_count_done, 0)
sub_sms = test_event.event_mail_ids.filtered(lambda s: s.interval_type == "after_sub" and s.interval_unit == "now" and s.notification_type == "sms")
self.assertEqual(len(sub_sms), 1)
self.assertEqual(sub_sms.mail_count_done, 0)
# setup batch and cron limit sizes to check iterative behavior
batch_size, cron_limit = 5, 20
self.env["ir.config_parameter"].sudo().set_param("mail.batch_size", batch_size)
self.env["ir.config_parameter"].sudo().set_param("mail.render.cron.limit", cron_limit)
# create registrations -> each one receives its on subscribe communication
EventMailRegistration = type(self.env['event.mail.registration'])
exec_origin = EventMailRegistration._execute_on_registrations
with patch.object(
EventMailRegistration, '_execute_on_registrations', autospec=True, wraps=EventMailRegistration, side_effect=exec_origin,
) as mock_exec, \
self.mock_datetime_and_now(self.reference_now + timedelta(hours=1)), \
self.mockSMSGateway(), \
self.mock_mail_gateway(), \
self.capture_triggers('event.event_mail_scheduler') as capture:
self._create_registrations(test_event, 30)
# iterative work on registrations: only 20 (cron limit) are taken into account
self.assertEqual(sub_mail.mail_count_done, 20)
self.assertEqual(sub_sms.mail_count_done, 20)
self.assertEqual(mock_exec.call_count, 8, "Batch of 5 to make 20 registrations: 4 calls / scheduler")
# cron should have been triggered for the remaining registrations
self.assertSchedulerCronTriggers(capture, [self.reference_now + timedelta(hours=1)] * 2)
# iterative work on registrations, force cron to close those
with patch.object(
EventMailRegistration, '_execute_on_registrations', autospec=True, wraps=EventMailRegistration, side_effect=exec_origin,
) as mock_exec, \
self.mock_datetime_and_now(self.reference_now + timedelta(hours=1)), \
self.mockSMSGateway(), \
self.mock_mail_gateway(), \
self.capture_triggers('event.event_mail_scheduler') as capture:
self.event_cron_id.method_direct_trigger()
# finished sending communications
self.assertEqual(sub_mail.mail_count_done, 30)
self.assertEqual(sub_sms.mail_count_done, 30)
self.assertFalse(capture.records)
self.assertEqual(mock_exec.call_count, 4, "Batch of 5 to make 10 remaining registrations: 2 calls / scheduler")
@tagged('event_mail', 'post_install', '-at_install')
class TestEventSaleMail(TestEventFullCommon):
def test_event_mail_on_sale_confirmation(self):
"""Test that a mail is sent to the customer when a sale order is confirmed."""
ticket = self.test_event.event_ticket_ids[0]
self.test_event.env.company.partner_id.email = 'test.email@test.example.com'
order_line_vals = {
"event_id": self.test_event.id,
"event_ticket_id": ticket.id,
"product_id": ticket.product_id.id,
"product_uom_qty": 1,
}
self.customer_so.write({"order_line": [(0, 0, order_line_vals)]})
# check sale mail configuration
aftersub = self.test_event.event_mail_ids.filtered(
lambda m: m.interval_type == "after_sub"
)
self.assertTrue(aftersub)
aftersub.template_ref.email_from = "{{ (object.event_id.organizer_id.email_formatted or object.event_id.user_id.email_formatted or '') }}"
self.assertEqual(self.test_event.organizer_id, self.test_event.env.company.partner_id)
registration = self.env["event.registration"].create(
{
**self.website_customer_data[0],
"partner_id": self.event_customer.id,
"sale_order_line_id": self.customer_so.order_line[0].id,
}
)
self.assertEqual(self.test_event.registration_ids, registration)
self.assertEqual(self.customer_so.state, "draft")
self.assertEqual(registration.state, "draft")
with self.mock_mail_gateway():
self.customer_so.action_confirm()
self.assertEqual(self.customer_so.state, "sale")
self.assertEqual(registration.state, "open")
# Ensure mails are sent to customers right after subscription
self.assertMailMailWRecord(
registration,
[self.event_customer.id],
"outgoing",
author=self.test_event.organizer_id,
fields_values={
"email_from": self.test_event.organizer_id.email_formatted,
},
)