<s> import unittest <EOL> from datetime import datetime <EOL> from trac . notification . model import Subscription <EOL> from trac . test import EnvironmentStub , MockRequest <EOL> from trac . util . datefmt import to_utimestamp , utc <EOL> class SubscriptionTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . env = EnvironmentStub ( ) <EOL> def tearDown ( self ) : <EOL> self . env . reset_db ( ) <EOL> def _add_subscriber ( self , req , class_ , distributor = '<STR_LIT:email>' , <EOL> format = '<STR_LIT>' , adverb = '<STR_LIT>' ) : <EOL> session = req . session <EOL> args = { '<STR_LIT>' : session . sid , '<STR_LIT>' : session . authenticated , <EOL> '<STR_LIT>' : distributor , '<STR_LIT>' : format , '<STR_LIT>' : adverb , <EOL> '<STR_LIT:class>' : class_ } <EOL> return Subscription . add ( self . env , args ) <EOL> def _insert_rows ( self ) : <EOL> rows = [ <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT:email>' , '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT:email>' , '<STR_LIT>' , <NUM_LIT:2> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT:email>' , '<STR_LIT>' , <NUM_LIT:3> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:2> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:3> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:2> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:3> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT:email>' , '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT:email>' , '<STR_LIT>' , <NUM_LIT:2> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT:email>' , '<STR_LIT>' , <NUM_LIT:3> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:2> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:3> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:2> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:3> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> ts = to_utimestamp ( datetime ( <NUM_LIT> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:12> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , utc ) ) <EOL> with self . env . db_transaction as db : <EOL> cursor = db . cursor ( ) <EOL> cursor . executemany ( """<STR_LIT>""" , <EOL> [ ( ts + idx , ts + idx * <NUM_LIT:2> ) + row for idx , row <EOL> in enumerate ( rows ) ] ) <EOL> def _props ( self , items , name ) : <EOL> return [ item [ name ] for item in items ] <EOL> def test_add ( self ) : <EOL> req = MockRequest ( self . env , authname = '<STR_LIT>' ) <EOL> with self . env . db_transaction : <EOL> self . _add_subscriber ( req , '<STR_LIT>' , format = '<STR_LIT>' ) <EOL> self . _add_subscriber ( req , '<STR_LIT>' ) <EOL> self . _add_subscriber ( req , '<STR_LIT>' , format = '<STR_LIT>' ) <EOL> self . _add_subscriber ( req , '<STR_LIT>' , distributor = '<STR_LIT>' , <EOL> adverb = '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> [ ( u'<STR_LIT>' , <NUM_LIT:1> , u'<STR_LIT:email>' , u'<STR_LIT>' , <NUM_LIT:1> , u'<STR_LIT>' , <EOL> u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , <NUM_LIT:1> , u'<STR_LIT:email>' , u'<STR_LIT>' , <NUM_LIT:2> , u'<STR_LIT>' , <EOL> u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , <NUM_LIT:1> , u'<STR_LIT:email>' , u'<STR_LIT>' , <NUM_LIT:3> , u'<STR_LIT>' , <EOL> u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , <NUM_LIT:1> , u'<STR_LIT>' , u'<STR_LIT>' , <NUM_LIT:1> , u'<STR_LIT>' , <EOL> u'<STR_LIT>' ) ] , <EOL> self . env . db_query ( """<STR_LIT>""" , ( '<STR_LIT>' , <NUM_LIT:1> ) ) ) <EOL> def test_delete ( self ) : <EOL> req = MockRequest ( self . env , authname = '<STR_LIT>' ) <EOL> with self . env . db_transaction : <EOL> ids = [ self . _add_subscriber ( req , '<STR_LIT>' ) , <EOL> self . _add_subscriber ( req , '<STR_LIT>' ) , <EOL> self . _add_subscriber ( req , '<STR_LIT>' ) , <EOL> self . _add_subscriber ( req , '<STR_LIT>' , <EOL> distributor = '<STR_LIT>' , adverb = '<STR_LIT>' ) , <EOL> self . _add_subscriber ( req , '<STR_LIT>' , <EOL> distributor = '<STR_LIT>' ) ] <EOL> self . assertEqual ( <NUM_LIT:5> , self . env . db_query ( """<STR_LIT>""" , ( '<STR_LIT>' , <NUM_LIT:1> ) ) [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ) <EOL> Subscription . delete ( self . env , ids [ <NUM_LIT:1> ] ) <EOL> rows = self . env . db_query ( """<STR_LIT>""" , ( '<STR_LIT>' , <NUM_LIT:1> ) ) <EOL> self . assertEqual ( ( ids [ <NUM_LIT:0> ] , '<STR_LIT:email>' , <NUM_LIT:1> , '<STR_LIT>' ) , rows [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( ( ids [ <NUM_LIT:2> ] , '<STR_LIT:email>' , <NUM_LIT:2> , '<STR_LIT>' ) , rows [ <NUM_LIT:1> ] ) <EOL> self . assertEqual ( ( ids [ <NUM_LIT:3> ] , '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' ) , rows [ <NUM_LIT:2> ] ) <EOL> self . assertEqual ( ( ids [ <NUM_LIT:4> ] , '<STR_LIT>' , <NUM_LIT:2> , '<STR_LIT>' ) , rows [ <NUM_LIT:3> ] ) <EOL> self . assertEqual ( <NUM_LIT:4> , len ( rows ) ) <EOL> def test_find_by_sid_and_distributor ( self ) : <EOL> self . _insert_rows ( ) <EOL> items = Subscription . find_by_sid_and_distributor ( self . env , '<STR_LIT>' , True , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( [ '<STR_LIT>' ] * <NUM_LIT:3> , self . _props ( items , '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ <NUM_LIT:1> ] * <NUM_LIT:3> , self . _props ( items , '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' ] * <NUM_LIT:3> , self . _props ( items , '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> self . _props ( items , '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] , self . _props ( items , '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , self . _props ( items , '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] , self . _props ( items , '<STR_LIT:class>' ) ) <EOL> def test_find_by_sids_and_class ( self ) : <EOL> self . _insert_rows ( ) <EOL> sids = [ ( '<STR_LIT>' , True ) , ( '<STR_LIT>' , True ) , ( '<STR_LIT>' , True ) , ( '<STR_LIT>' , True ) ] <EOL> items = Subscription . find_by_sids_and_class ( self . env , sids , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT>' ] , self . _props ( items , '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ <NUM_LIT:1> ] * <NUM_LIT:2> , self . _props ( items , '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' ] * <NUM_LIT:2> , self . _props ( items , '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> self . _props ( items , '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ <NUM_LIT:3> , <NUM_LIT:3> ] , self . _props ( items , '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT>' ] , self . _props ( items , '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> self . _props ( items , '<STR_LIT:class>' ) ) <EOL> def test_move ( self ) : <EOL> def query_subs ( ) : <EOL> return self . env . db_query ( """<STR_LIT>""" , ( '<STR_LIT>' , <NUM_LIT:1> ) ) <EOL> req = MockRequest ( self . env , authname = '<STR_LIT>' ) <EOL> with self . env . db_transaction : <EOL> rule_ids = { } <EOL> for class_ , distributor in [ ( '<STR_LIT>' , '<STR_LIT:email>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT:email>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT:email>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT:email>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] : <EOL> rule_ids [ ( class_ , distributor ) ] = self . _add_subscriber ( req , class_ , distributor ) <EOL> self . assertEqual ( [ ( '<STR_LIT:email>' , <NUM_LIT:1> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:email>' , <NUM_LIT:2> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:email>' , <NUM_LIT:3> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:email>' , <NUM_LIT:4> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:2> , '<STR_LIT>' ) , <EOL> ] , query_subs ( ) ) <EOL> Subscription . move ( self . env , rule_ids [ ( '<STR_LIT>' , '<STR_LIT:email>' ) ] , <NUM_LIT:1> ) <EOL> self . assertEqual ( [ ( '<STR_LIT:email>' , <NUM_LIT:1> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:email>' , <NUM_LIT:2> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:email>' , <NUM_LIT:3> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:email>' , <NUM_LIT:4> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:2> , '<STR_LIT>' ) , <EOL> ] , query_subs ( ) ) <EOL> Subscription . move ( self . env , rule_ids [ ( '<STR_LIT>' , '<STR_LIT:email>' ) ] , <NUM_LIT:4> ) <EOL> self . assertEqual ( [ ( '<STR_LIT:email>' , <NUM_LIT:1> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:email>' , <NUM_LIT:2> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:email>' , <NUM_LIT:3> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:email>' , <NUM_LIT:4> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:2> , '<STR_LIT>' ) , <EOL> ] , query_subs ( ) ) <EOL> Subscription . move ( self . env , rule_ids [ ( '<STR_LIT>' , '<STR_LIT:email>' ) ] , <NUM_LIT:3> ) <EOL> self . assertEqual ( [ ( '<STR_LIT:email>' , <NUM_LIT:1> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:email>' , <NUM_LIT:2> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:email>' , <NUM_LIT:3> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:email>' , <NUM_LIT:4> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:2> , '<STR_LIT>' ) , <EOL> ] , query_subs ( ) ) <EOL> def test_replace_all ( self ) : <EOL> def query ( sid , authenticated ) : <EOL> return self . env . db_query ( """<STR_LIT>""" , ( sid , authenticated ) ) <EOL> req = MockRequest ( self . env , authname = '<STR_LIT>' ) <EOL> sess = req . session <EOL> items = [ <EOL> ( '<STR_LIT:email>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:email>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:email>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> items = [ dict ( zip ( ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:class>' ) , item ) ) <EOL> for item in items ] <EOL> Subscription . replace_all ( self . env , sess . sid , sess . authenticated , items ) <EOL> rows = query ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> expected = [ <EOL> ( '<STR_LIT:email>' , '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:email>' , '<STR_LIT>' , <NUM_LIT:2> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:email>' , '<STR_LIT>' , <NUM_LIT:3> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:2> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> self . assertEqual ( expected , rows ) <EOL> items = [ <EOL> ( '<STR_LIT:email>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> items = [ dict ( zip ( ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:class>' ) , item ) ) <EOL> for item in items ] <EOL> Subscription . replace_all ( self . env , sess . sid , sess . authenticated , items ) <EOL> rows = query ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> expected = [ <EOL> ( '<STR_LIT:email>' , '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> self . assertEqual ( expected , rows ) <EOL> Subscription . replace_all ( self . env , sess . sid , sess . authenticated , [ ] ) <EOL> self . assertEqual ( [ ] , query ( '<STR_LIT>' , <NUM_LIT:1> ) ) <EOL> def test_update_format_by_distributor_and_sid ( self ) : <EOL> self . _insert_rows ( ) <EOL> Subscription . update_format_by_distributor_and_sid ( <EOL> self . env , '<STR_LIT:email>' , '<STR_LIT>' , True , '<STR_LIT>' ) <EOL> rows = self . env . db_query ( """<STR_LIT>""" , ( '<STR_LIT>' , <NUM_LIT:1> ) ) <EOL> expected = [ <EOL> ( '<STR_LIT:email>' , '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:email>' , '<STR_LIT>' , <NUM_LIT:2> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:email>' , '<STR_LIT>' , <NUM_LIT:3> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:2> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:3> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:2> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:3> , '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> self . assertEqual ( expected , rows ) <EOL> def suite ( ) : <EOL> suite = unittest . TestSuite ( ) <EOL> suite . addTest ( unittest . makeSuite ( SubscriptionTestCase ) ) <EOL> return suite <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( defaultTest = '<STR_LIT>' ) </s>
