# -*- coding: utf-8 -*- # # Copyright (C) 2017-2023 Edgewall Software # All rights reserved. # # This software is licensed as described in the file COPYING, which # you should have received as part of this distribution. The terms # are also available at https://trac.edgewall.org/wiki/TracLicense. # # This software consists of voluntary contributions made by many # individuals. For the exact contribution history, see the revision # history and logs, available at https://trac.edgewall.org/log/. import unittest from trac.notification.model import Subscription from trac.notification.prefs import NotificationPreferences from trac.prefs.web_ui import PreferencesModule from trac.test import EnvironmentStub, MockRequest, makeSuite from trac.web.api import RequestDone class NotificationPreferencesTestCase(unittest.TestCase): def setUp(self): self.env = EnvironmentStub() self.prefs_mod = PreferencesModule(self.env) ids = self._add_subscriptions([ {'sid': 'foo', 'class_': 'TicketOwnerSubscriber'}, {'sid': 'foo', 'class_': 'CarbonCopySubscriber'}, {'sid': 'foo', 'class_': 'TicketReporterSubscriber'}, {'sid': 'bar', 'class_': 'CarbonCopySubscriber'}, ]) self.foo_rule0 = ids[0] self.foo_rule1 = ids[1] self.foo_rule2 = ids[2] self.bar_rule0 = ids[3] def tearDown(self): self.env.reset_db() def _add_subscription(self, **kwargs): props = {'sid': None, 'authenticated': 1, 'distributor': 'email', 'format': 'text/plain', 'adverb': 'always', 'class': 'TicketOwnerSubscriber'} props.update((k.rstrip('_'), (v or None) if isinstance(v, str) else v) for k, v in kwargs.items()) assert props['sid'] is not None return Subscription.add(self.env, props) def _add_subscriptions(self, rows): with self.env.db_transaction: return [self._add_subscription(**row) for row in rows] def _get_rules(self, sid, authenticated=1): return self.env.db_query(""" SELECT distributor, format, priority, adverb, class FROM notify_subscription WHERE sid=%s AND authenticated=1 ORDER BY distributor, priority """, (sid,)) def _request(self, **kwargs): kwargs.setdefault('method', 'POST') kwargs.setdefault('path_info', '/prefs/notification') return MockRequest(self.env, **kwargs) def _process(self, req): self.assertTrue(self.prefs_mod.match_request(req)) return self.prefs_mod.process_request(req) def test_add_rule(self): args = {'action': 'add-rule_email', 'new-adverb-email': 'never', 'format-email': '', 'new-rule-email': 'TicketOwnerSubscriber'} req = self._request(authname='baz', args=args) self.assertRaises(RequestDone, self._process, req) self.assertEqual( [('email', None, 1, 'never', 'TicketOwnerSubscriber')], self._get_rules('baz')) args = {'action': 'add-rule_email', 'new-adverb-email': 'always', 'format-email': 'text/plain', 'new-rule-email': 'CarbonCopySubscriber'} req = self._request(authname='baz', args=args) self.assertRaises(RequestDone, self._process, req) self.assertEqual( [('email', None, 1, 'never', 'TicketOwnerSubscriber'), ('email', 'text/plain', 2, 'always', 'CarbonCopySubscriber')], self._get_rules('baz')) def test_delete_rule(self): req = self._request(authname='foo', args={'action': 'delete-rule_%d' % self.foo_rule1}) self.assertRaises(RequestDone, self._process, req) self.assertEqual( [('email', 'text/plain', 1, 'always', 'TicketOwnerSubscriber'), ('email', 'text/plain', 2, 'always', 'TicketReporterSubscriber')], self._get_rules('foo')) # try to delete non-existent rule rules = self._get_rules('foo') req = self._request(authname='foo', args={'action': 'delete-rule_%d' % self.foo_rule1}) self.assertRaises(RequestDone, self._process, req) self.assertEqual(rules, self._get_rules('foo')) # try to delete non-owned rules rules = self._get_rules('bar') req = self._request(authname='foo', args={'action': 'delete-rule_%d' % self.bar_rule0}) self.assertRaises(RequestDone, self._process, req) self.assertEqual(rules, self._get_rules('bar')) def test_move_rule(self): # move to last req = self._request(authname='foo', args={'action': 'move-rule_%d-3' % self.foo_rule1}) self.assertRaises(RequestDone, self._process, req) self.assertEqual( [('email', 'text/plain', 1, 'always', 'TicketOwnerSubscriber'), ('email', 'text/plain', 2, 'always', 'TicketReporterSubscriber'), ('email', 'text/plain', 3, 'always', 'CarbonCopySubscriber')], self._get_rules('foo')) self.assertEqual( [('email', 'text/plain', 1, 'always', 'CarbonCopySubscriber')], self._get_rules('bar')) # move to first req = self._request(authname='foo', args={'action': 'move-rule_%d-1' % self.foo_rule1}) self.assertRaises(RequestDone, self._process, req) self.assertEqual( [('email', 'text/plain', 1, 'always', 'CarbonCopySubscriber'), ('email', 'text/plain', 2, 'always', 'TicketOwnerSubscriber'), ('email', 'text/plain', 3, 'always', 'TicketReporterSubscriber')], self._get_rules('foo')) # move to same position rules = self._get_rules('foo') req = self._request(authname='foo', args={'action': 'move-rule_%d-1' % self.foo_rule1}) self.assertRaises(RequestDone, self._process, req) self.assertEqual(rules, self._get_rules('foo')) # try to move to out of range rules = self._get_rules('foo') req = self._request(authname='foo', args={'action': 'move-rule_%d-42' % self.foo_rule1}) self.assertRaises(RequestDone, self._process, req) self.assertEqual(rules, self._get_rules('foo')) # try to move non-owned rules rules = self._get_rules('foo') req = self._request(authname='bar', args={'action': 'move-rule_%d-3' % self.foo_rule1}) self.assertRaises(RequestDone, self._process, req) self.assertEqual(rules, self._get_rules('foo')) rules = self._get_rules('foo') # try to move non-existent rule req = self._request(authname='foo', args={'action': 'move-rule_42-1'}) self.assertRaises(RequestDone, self._process, req) self.assertEqual(rules, self._get_rules('foo')) # invalid id req = self._request(authname='foo', args={'action': 'move-rule_0-1'}) self.assertRaises(RequestDone, self._process, req) self.assertEqual(rules, self._get_rules('foo')) req = self._request(authname='foo', args={'action': 'move-rule_a-1'}) self.assertRaises(RequestDone, self._process, req) self.assertEqual(rules, self._get_rules('foo')) # invalid priority req = self._request(authname='foo', args={'action': 'move-rule_%d-0' % self.foo_rule1}) self.assertRaises(RequestDone, self._process, req) self.assertEqual(rules, self._get_rules('foo')) req = self._request(authname='foo', args={'action': 'move-rule_%d-a' % self.foo_rule1}) self.assertRaises(RequestDone, self._process, req) self.assertEqual(rules, self._get_rules('foo')) def test_change_format(self): # set text/html rules = self._get_rules('foo') arg_list = [ ('action', 'replace_all'), ('format-email', 'text/html'), ] + \ [('adverb-email', r[3]) for r in rules] + \ [('class-email', r[4]) for r in rules] req = self._request(authname='foo', arg_list=arg_list) self.assertNotIn('notification.format.email', req.session) self.assertRaises(RequestDone, self._process, req) self.assertEqual( [('email', 'text/html', 1, 'always', 'TicketOwnerSubscriber'), ('email', 'text/html', 2, 'always', 'CarbonCopySubscriber'), ('email', 'text/html', 3, 'always', 'TicketReporterSubscriber')], self._get_rules('foo')) self.assertEqual( [('email', 'text/plain', 1, 'always', 'CarbonCopySubscriber')], self._get_rules('bar')) self.assertEqual('text/html', req.session['notification.format.email']) # set default format arg_list[1] = ('format-email', None) req = self._request(authname='foo', arg_list=arg_list) self.assertRaises(RequestDone, self._process, req) self.assertEqual( [('email', None, 1, 'always', 'TicketOwnerSubscriber'), ('email', None, 2, 'always', 'CarbonCopySubscriber'), ('email', None, 3, 'always', 'TicketReporterSubscriber')], self._get_rules('foo')) self.assertNotIn('notification.format.email', req.session) def test_replace(self): arg_list = [ ('action', 'replace_all'), ('format-email', 'text/plain'), ('adverb-email', 'always'), ('adverb-email', 'never'), ('adverb-email', 'always'), ('adverb-email', 'always'), ('class-email', 'TicketOwnerSubscriber'), ('class-email', 'CarbonCopySubscriber'), ('class-email', 'TicketReporterSubscriber'), ('class-email', 'TicketUpdaterSubscriber'), ] req = self._request(authname='foo', arg_list=arg_list) self.assertRaises(RequestDone, self._process, req) self.assertEqual( [('email', 'text/plain', 1, 'always', 'TicketOwnerSubscriber'), ('email', 'text/plain', 2, 'never', 'CarbonCopySubscriber'), ('email', 'text/plain', 3, 'always', 'TicketReporterSubscriber'), ('email', 'text/plain', 4, 'always', 'TicketUpdaterSubscriber')], self._get_rules('foo')) self.assertEqual( [('email', 'text/plain', 1, 'always', 'CarbonCopySubscriber')], self._get_rules('bar')) arg_list = [ ('action', 'replace_all'), ('format-email', ''), ('adverb-email', 'always'), ('class-email', 'TicketOwnerSubscriber'), ] req = self._request(authname='foo', arg_list=arg_list) self.assertRaises(RequestDone, self._process, req) self.assertEqual( [('email', None, 1, 'always', 'TicketOwnerSubscriber')], self._get_rules('foo')) arg_list = [('action', 'replace_all'), ('format-email', '')] req = self._request(authname='foo', arg_list=arg_list) self.assertRaises(RequestDone, self._process, req) self.assertEqual([], self._get_rules('foo')) def test_suite(): suite = unittest.TestSuite() suite.addTest(makeSuite(NotificationPreferencesTestCase)) return suite if __name__ == '__main__': unittest.main(defaultTest='test_suite')