# -*- coding: utf-8 -*- # Part of Odoo. See LICENSE file for full copyright and licensing details. import itertools import random from dateutil.relativedelta import relativedelta from lxml import html from werkzeug.urls import url_encode from odoo import fields, SUPERUSER_ID from odoo.addons.base.tests.common import HttpCaseWithUserDemo from odoo.addons.mail.tests import common as mail_test from odoo.tests import tagged from odoo.tests.common import users class TestDigest(mail_test.MailCommon): @classmethod def setUpClass(cls): super(TestDigest, cls).setUpClass() cls._activate_multi_company() # clean messages cls.env['mail.message'].search([ ('subtype_id', '=', cls.env.ref('mail.mt_comment').id), ('message_type', 'in', ['comment', 'email']), ]).unlink() cls._setup_messages() # clean demo users so that we keep only the test users cls.env['res.users'].search([('login', 'in', ['demo', 'portal'])]).action_archive() # clean logs so that town down is activated cls.env['res.users.log'].search([('create_uid', 'in', (cls.user_admin + cls.user_employee).ids)]).unlink() cls.test_digest = cls.env['digest.digest'].create({ 'kpi_mail_message_total': True, 'kpi_res_users_connected': True, 'name': "My Digest", 'periodicity': 'daily', }) @classmethod def _setup_messages(cls): """ Remove all existing messages, then create a bunch of them on random partners with the correct types in correct time-bucket: - 3 in the previous 24h - 5 more in the 6 days before that for a total of 8 in the previous week - 7 more in the 20 days before *that* (because digest doc lies and is based around weeks and months not days), for a total of 15 in the previous month """ # regular employee can't necessarily access "private" addresses partners = cls.env['res.partner'].search([('type', '!=', 'private')]) messages = cls.env['mail.message'] counter = itertools.count() now = fields.Datetime.now() for count, (low, high) in [(3, (0 * 24, 1 * 24)), (5, (1 * 24, 7 * 24)), (7, (7 * 24, 27 * 24)), ]: for _ in range(count): create_date = now - relativedelta(hours=random.randint(low + 1, high - 1)) messages += random.choice(partners).message_post( author_id=cls.partner_admin.id, body=f"Awesome Partner! ({next(counter)})", email_from=cls.partner_admin.email_formatted, message_type='comment', subtype_xmlid='mail.mt_comment', # adjust top and bottom by 1h to avoid overlapping with the # range limit and dropping out of the digest's selection thing create_date=create_date, ) messages.flush() @users('admin') def test_digest_numbers(self): digest = self.env['digest.digest'].browse(self.test_digest.ids) digest._action_subscribe_users(self.user_employee) # digest creates its mails in auto_delete mode so we need to capture # the formatted body during the sending process digest.flush() with self.mock_mail_gateway(): digest.action_send() self.assertEqual(len(self._new_mails), 1, "A new mail.mail should have been created") mail = self._new_mails[0] # check mail.mail content self.assertEqual(mail.author_id, self.partner_admin) self.assertEqual(mail.email_from, self.company_admin.email_formatted) self.assertEqual(mail.state, 'outgoing', 'Mail should use the queue') kpi_message_values = html.fromstring(mail.body_html).xpath('//div[@data-field="kpi_mail_message_total"]//*[hasclass("kpi_value")]/text()') self.assertEqual( [t.strip() for t in kpi_message_values], ['3', '8', '15'] ) @users('admin') def test_digest_subscribe(self): digest_user = self.test_digest.with_user(self.user_employee) self.assertFalse(digest_user.is_subscribed) # subscribe a user so at least one mail gets sent digest_user.action_subscribe() self.assertTrue( digest_user.is_subscribed, "check the user was subscribed as action_subscribe will silently " "ignore subs of non-employees" ) @users('admin') def test_digest_tone_down(self): digest = self.env['digest.digest'].browse(self.test_digest.ids) digest._action_subscribe_users(self.user_employee) # initial data self.assertEqual(digest.periodicity, 'daily') # no logs for employee -> should tone down periodicity digest.flush() with self.mock_mail_gateway(): digest.action_send() self.assertEqual(digest.periodicity, 'weekly') # no logs for employee -> should tone down periodicity digest.flush() with self.mock_mail_gateway(): digest.action_send() self.assertEqual(digest.periodicity, 'monthly') # no logs for employee -> should tone down periodicity digest.flush() with self.mock_mail_gateway(): digest.action_send() self.assertEqual(digest.periodicity, 'quarterly') @users('admin') def test_digest_tone_down_wlogs(self): digest = self.env['digest.digest'].browse(self.test_digest.ids) digest._action_subscribe_users(self.user_employee) # initial data self.assertEqual(digest.periodicity, 'daily') # logs for employee -> should not tone down logs = self.env['res.users.log'].with_user(SUPERUSER_ID).create({'create_uid': self.user_employee.id}) digest.flush() with self.mock_mail_gateway(): digest.action_send() logs.unlink() logs = self.env['res.users.log'].with_user(SUPERUSER_ID).create({ 'create_uid': self.user_employee.id, 'create_date': fields.Datetime.now() - relativedelta(days=20), }) # logs for employee are more than 3 days old -> should tone down digest.flush() with self.mock_mail_gateway(): digest.action_send() self.assertEqual(digest.periodicity, 'weekly') # logs for employee are more than 2 weeks old -> should tone down digest.flush() with self.mock_mail_gateway(): digest.action_send() self.assertEqual(digest.periodicity, 'monthly') # logs for employee are less than 1 month old -> should not tone down digest.flush() with self.mock_mail_gateway(): digest.action_send() self.assertEqual(digest.periodicity, 'monthly') @tagged('-at_install', 'post_install') class TestUnsubscribe(HttpCaseWithUserDemo): def setUp(self): super(TestUnsubscribe, self).setUp() self.test_digest = self.env['digest.digest'].create({ 'kpi_mail_message_total': True, 'kpi_res_users_connected': True, 'name': "My Digest", 'periodicity': 'daily', 'user_ids': self.user_demo.ids, }) self.test_digest._action_subscribe_users(self.user_demo) self.base_url = self.test_digest.get_base_url() self.user_demo_unsubscribe_token = self.test_digest._get_unsubscribe_token(self.user_demo.id) @users('demo') def test_unsubscribe_classic(self): self.assertIn(self.user_demo, self.test_digest.user_ids) self.authenticate(self.env.user.login, self.env.user.login) response = self._url_unsubscribe() self.assertEqual(response.status_code, 200) self.assertNotIn(self.user_demo, self.test_digest.user_ids) @users('demo') def test_unsubscribe_issues(self): """ Test when not being member """ self.test_digest.write({'user_ids': [(3, self.user_demo.id)]}) self.assertNotIn(self.user_demo, self.test_digest.user_ids) # unsubscribe self.authenticate(self.env.user.login, self.env.user.login) response = self._url_unsubscribe() self.assertEqual(response.status_code, 200) self.assertNotIn(self.user_demo, self.test_digest.user_ids) def test_unsubscribe_token(self): self.assertIn(self.user_demo, self.test_digest.user_ids) self.authenticate(None, None) response = self._url_unsubscribe(token=self.user_demo_unsubscribe_token, user_id=self.user_demo.id) self.assertEqual(response.status_code, 200) self.test_digest.invalidate_cache() self.assertNotIn(self.user_demo, self.test_digest.user_ids) def test_unsubscribe_public(self): """ Check public users are redirected when trying to catch unsubscribe route. """ self.authenticate(None, None) response = self._url_unsubscribe() self.assertEqual(response.status_code, 404) def _url_unsubscribe(self, token=None, user_id=None): url_params = {} if token is not None: url_params['token'] = token if user_id is not None: url_params['user_id'] = user_id url = "%s/digest/%s/unsubscribe?%s" % ( self.base_url, self.test_digest.id, url_encode(url_params) ) return self.url_open(url)