# -*- coding: utf-8 -*- # # Copyright (C) 2016-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 datetime import datetime from trac.notification.model import Subscription from trac.test import EnvironmentStub, MockRequest, makeSuite from trac.util.datefmt import to_utimestamp, utc class SubscriptionTestCase(unittest.TestCase): def setUp(self): self.env = EnvironmentStub() def tearDown(self): self.env.reset_db() def _add_subscriber(self, req, class_, distributor='email', format='text/plain', adverb='always'): session = req.session args = {'sid': session.sid, 'authenticated': session.authenticated, 'distributor': distributor, 'format': format, 'adverb': adverb, 'class': class_} return Subscription.add(self.env, args) def _insert_rows(self): rows = [ ('joe', 1, 'email', 'text/plain', 1, 'always', 'EmailSubscriber1'), ('joe', 1, 'email', 'text/html', 2, 'always', 'EmailSubscriber2'), ('joe', 1, 'email', 'text/plain', 3, 'always', 'EmailSubscriber3'), ('joe', 1, 'xmpp', 'text/html', 1, 'always', 'XmppSubscriber1'), ('joe', 1, 'xmpp', 'text/plain', 2, 'never', 'XmppSubscriber2'), ('joe', 1, 'xmpp', 'text/html', 3, 'never', 'XmppSubscriber3'), ('joe', 1, 'irc', 'text/plain', 1, 'never', 'IrcSubscriber1'), ('joe', 1, 'irc', 'text/plain', 2, 'never', 'IrcSubscriber2'), ('joe', 1, 'irc', 'text/plain', 3, 'never', 'IrcSubscriber3'), ('jes', 1, 'email', 'text/html', 1, 'always', 'EmailSubscriber1'), ('jes', 1, 'email', 'text/plain', 2, 'never', 'EmailSubscriber2'), ('jes', 1, 'email', 'text/html', 3, 'always', 'EmailSubscriber3'), ('jan', 1, 'xmpp', 'text/plain', 1, 'always', 'XmppSubscriber1'), ('jan', 1, 'xmpp', 'text/html', 2, 'never', 'XmppSubscriber2'), ('jan', 1, 'xmpp', 'text/plain', 3, 'never', 'XmppSubscriber3'), ('jim', 1, 'irc', 'text/html', 1, 'always', 'IrcSubscriber1'), ('jim', 1, 'irc', 'text/plain', 2, 'never', 'IrcSubscriber2'), ('jim', 1, 'irc', 'text/html', 3, 'always', 'IrcSubscriber3'), ] ts = to_utimestamp(datetime(2016, 2, 3, 12, 34, 56, 987654, utc)) with self.env.db_transaction as db: cursor = db.cursor() cursor.executemany(""" INSERT INTO notify_subscription ( time, changetime, sid, authenticated, distributor, format, priority, adverb, class) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)""", [(ts + idx, ts + idx * 2) + row for idx, row in enumerate(rows)]) def _props(self, items, name): return [item[name] for item in items] def test_add(self): req = MockRequest(self.env, authname='joe') with self.env.db_transaction: self._add_subscriber(req, 'TicketSubscriber1', format='text/html') self._add_subscriber(req, 'TicketSubscriber2') self._add_subscriber(req, 'TicketSubscriber3', format='text/html') self._add_subscriber(req, 'XmppSubscriber1', distributor='xmpp', adverb='never') self.assertEqual( [('joe', 1, 'email', 'text/html', 1, 'always', 'TicketSubscriber1'), ('joe', 1, 'email', 'text/plain', 2, 'always', 'TicketSubscriber2'), ('joe', 1, 'email', 'text/html', 3, 'always', 'TicketSubscriber3'), ('joe', 1, 'xmpp', 'text/plain', 1, 'never', 'XmppSubscriber1')], self.env.db_query("""\ SELECT sid, authenticated, distributor, format, priority, adverb, class FROM notify_subscription WHERE sid=%s AND authenticated=%s ORDER BY distributor, priority""", ('joe', 1))) def test_delete(self): req = MockRequest(self.env, authname='joe') with self.env.db_transaction: ids = [self._add_subscriber(req, 'TicketSubscriber1'), self._add_subscriber(req, 'TicketSubscriber2'), self._add_subscriber(req, 'TicketSubscriber3'), self._add_subscriber(req, 'XmppSubscriber1', distributor='xmpp', adverb='never'), self._add_subscriber(req, 'XmppSubscriber2', distributor='xmpp')] self.assertEqual(5, self.env.db_query("""\ SELECT COUNT(*) FROM notify_subscription WHERE sid=%s AND authenticated=%s""", ('joe', 1))[0][0]) Subscription.delete(self.env, ids[1]) rows = self.env.db_query("""\ SELECT id, distributor, priority, class FROM notify_subscription WHERE sid=%s AND authenticated=%s ORDER BY distributor, priority""", ('joe', 1)) self.assertEqual((ids[0], 'email', 1, 'TicketSubscriber1'), rows[0]) self.assertEqual((ids[2], 'email', 2, 'TicketSubscriber3'), rows[1]) self.assertEqual((ids[3], 'xmpp', 1, 'XmppSubscriber1'), rows[2]) self.assertEqual((ids[4], 'xmpp', 2, 'XmppSubscriber2'), rows[3]) self.assertEqual(4, len(rows)) def test_find_by_sid_and_distributor(self): self._insert_rows() items = Subscription.find_by_sid_and_distributor(self.env, 'joe', True, 'xmpp') self.assertEqual(['joe'] * 3, self._props(items, 'sid')) self.assertEqual([1] * 3, self._props(items, 'authenticated')) self.assertEqual(['xmpp'] * 3, self._props(items, 'distributor')) self.assertEqual(['text/html', 'text/plain', 'text/html'], self._props(items, 'format')) self.assertEqual([1, 2, 3], self._props(items, 'priority')) self.assertEqual(['always', 'never', 'never'], self._props(items, 'adverb')) self.assertEqual(['XmppSubscriber1', 'XmppSubscriber2', 'XmppSubscriber3'], self._props(items, 'class')) def test_find_by_sids_and_class(self): self._insert_rows() sids = [('joe', True), ('jes', True), ('jan', True), ('jim', True)] items = Subscription.find_by_sids_and_class(self.env, sids, 'IrcSubscriber3') self.assertEqual(['joe', 'jim'], self._props(items, 'sid')) self.assertEqual([1] * 2, self._props(items, 'authenticated')) self.assertEqual(['irc'] * 2, self._props(items, 'distributor')) self.assertEqual(['text/plain', 'text/html'], self._props(items, 'format')) self.assertEqual([3, 3], self._props(items, 'priority')) self.assertEqual(['never', 'always'], self._props(items, 'adverb')) self.assertEqual(['IrcSubscriber3', 'IrcSubscriber3'], self._props(items, 'class')) def test_move(self): def query_subs(): return self.env.db_query("""\ SELECT distributor, priority, class FROM notify_subscription WHERE sid=%s AND authenticated=%s ORDER BY distributor, priority""", ('joe', 1)) req = MockRequest(self.env, authname='joe') with self.env.db_transaction: rule_ids = {} for class_, distributor in [('EmailSubscriber1', 'email'), ('EmailSubscriber2', 'email'), ('EmailSubscriber3', 'email'), ('EmailSubscriber4', 'email'), ('XmppSubscriber1', 'xmpp'), ('XmppSubscriber2', 'xmpp')]: rule_ids[(class_, distributor)] = \ self._add_subscriber(req, class_, distributor) self.assertEqual([('email', 1, 'EmailSubscriber1'), ('email', 2, 'EmailSubscriber2'), ('email', 3, 'EmailSubscriber3'), ('email', 4, 'EmailSubscriber4'), ('xmpp', 1, 'XmppSubscriber1'), ('xmpp', 2, 'XmppSubscriber2'), ], query_subs()) Subscription.move(self.env, rule_ids[('EmailSubscriber3', 'email')], 1) self.assertEqual([('email', 1, 'EmailSubscriber3'), ('email', 2, 'EmailSubscriber1'), ('email', 3, 'EmailSubscriber2'), ('email', 4, 'EmailSubscriber4'), ('xmpp', 1, 'XmppSubscriber1'), ('xmpp', 2, 'XmppSubscriber2'), ], query_subs()) Subscription.move(self.env, rule_ids[('EmailSubscriber1', 'email')], 4) self.assertEqual([('email', 1, 'EmailSubscriber3'), ('email', 2, 'EmailSubscriber2'), ('email', 3, 'EmailSubscriber4'), ('email', 4, 'EmailSubscriber1'), ('xmpp', 1, 'XmppSubscriber1'), ('xmpp', 2, 'XmppSubscriber2'), ], query_subs()) Subscription.move(self.env, rule_ids[('EmailSubscriber3', 'email')], 3) self.assertEqual([('email', 1, 'EmailSubscriber2'), ('email', 2, 'EmailSubscriber4'), ('email', 3, 'EmailSubscriber3'), ('email', 4, 'EmailSubscriber1'), ('xmpp', 1, 'XmppSubscriber1'), ('xmpp', 2, 'XmppSubscriber2'), ], query_subs()) def test_replace_all(self): def query(sid, authenticated): return self.env.db_query("""\ SELECT distributor, format, priority, adverb, class FROM notify_subscription WHERE sid=%s AND authenticated=%s ORDER BY distributor, priority""", (sid, authenticated)) req = MockRequest(self.env, authname='joe') sess = req.session items = [ ('email', 'text/plain', 'always', 'TicketSubscriber1'), ('email', 'text/html', 'always', 'TicketSubscriber2'), ('email', 'text/html', 'always', 'TicketSubscriber3'), ('xmpp', 'text/html', 'never', 'XmppSubscriber1'), ('xmpp', 'text/plain', 'always', 'XmppSubscriber2'), ] items = [dict(zip(('distributor', 'format', 'adverb', 'class'), item)) for item in items] Subscription.replace_all(self.env, sess.sid, sess.authenticated, items) rows = query('joe', 1) expected = [ ('email', 'text/plain', 1, 'always', 'TicketSubscriber1'), ('email', 'text/html', 2, 'always', 'TicketSubscriber2'), ('email', 'text/html', 3, 'always', 'TicketSubscriber3'), ('xmpp', 'text/html', 1, 'never', 'XmppSubscriber1'), ('xmpp', 'text/plain', 2, 'always', 'XmppSubscriber2'), ] self.assertEqual(expected, rows) items = [ ('email', 'text/plain', 'never', 'TicketSubscriber3'), ('xmpp', 'text/html', 'always', 'XmppSubscriber1'), ] items = [dict(zip(('distributor', 'format', 'adverb', 'class'), item)) for item in items] Subscription.replace_all(self.env, sess.sid, sess.authenticated, items) rows = query('joe', 1) expected = [ ('email', 'text/plain', 1, 'never', 'TicketSubscriber3'), ('xmpp', 'text/html', 1, 'always', 'XmppSubscriber1'), ] self.assertEqual(expected, rows) Subscription.replace_all(self.env, sess.sid, sess.authenticated, []) self.assertEqual([], query('joe', 1)) def test_update_format_by_distributor_and_sid(self): self._insert_rows() Subscription.update_format_by_distributor_and_sid( self.env, 'email', 'joe', True, 'application/pdf') rows = self.env.db_query("""\ SELECT distributor, format, priority, adverb, class FROM notify_subscription WHERE sid=%s AND authenticated=%s ORDER BY distributor, priority""", ('joe', 1)) expected = [ ('email', 'application/pdf', 1, 'always', 'EmailSubscriber1'), ('email', 'application/pdf', 2, 'always', 'EmailSubscriber2'), ('email', 'application/pdf', 3, 'always', 'EmailSubscriber3'), ('irc', 'text/plain', 1, 'never', 'IrcSubscriber1'), ('irc', 'text/plain', 2, 'never', 'IrcSubscriber2'), ('irc', 'text/plain', 3, 'never', 'IrcSubscriber3'), ('xmpp', 'text/html', 1, 'always', 'XmppSubscriber1'), ('xmpp', 'text/plain', 2, 'never', 'XmppSubscriber2'), ('xmpp', 'text/html', 3, 'never', 'XmppSubscriber3'), ] self.assertEqual(expected, rows) def test_suite(): suite = unittest.TestSuite() suite.addTest(makeSuite(SubscriptionTestCase)) return suite if __name__ == '__main__': unittest.main(defaultTest='test_suite')