<s> from tapiriik . testing . testtools import TestTools , TapiriikTestCase <EOL> from tapiriik . sync import SynchronizationTask <EOL> from tapiriik . sync . activity_record import ActivityRecord <EOL> from tapiriik . services import UserException , UserExceptionType <EOL> from tapiriik . services . api import APIExcludeActivity <EOL> from tapiriik . services . interchange import Activity , ActivityType <EOL> from tapiriik . auth import User <EOL> from datetime import datetime , timedelta , tzinfo <EOL> import pytz <EOL> import copy <EOL> class UTC ( tzinfo ) : <EOL> """<STR_LIT>""" <EOL> def utcoffset ( self , dt ) : <EOL> return timedelta ( <NUM_LIT:0> ) <EOL> def tzname ( self , dt ) : <EOL> return "<STR_LIT>" <EOL> def dst ( self , dt ) : <EOL> return timedelta ( <NUM_LIT:0> ) <EOL> class SyncTests ( TapiriikTestCase ) : <EOL> def test_svc_level_dupe ( self ) : <EOL> '''<STR_LIT>''' <EOL> svcA , svcB = TestTools . create_mock_services ( ) <EOL> recA = TestTools . create_mock_svc_record ( svcA ) <EOL> recB = TestTools . create_mock_svc_record ( svcB ) <EOL> actA = Activity ( ) <EOL> actA . StartTime = datetime ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> ) <EOL> actA . ServiceDataCollection = TestTools . create_mock_servicedatacollection ( svcA , record = recA ) <EOL> actB = Activity ( ) <EOL> actB . StartTime = actA . StartTime <EOL> actB . ServiceDataCollection = TestTools . create_mock_servicedatacollection ( svcB , record = recB ) <EOL> actA . CalculateUID ( ) <EOL> actB . CalculateUID ( ) <EOL> s = SynchronizationTask ( None ) <EOL> s . _activities = [ ] <EOL> s . _accumulateActivities ( recA , [ actA ] ) <EOL> s . _accumulateActivities ( recB , [ actB ] ) <EOL> self . assertEqual ( len ( s . _activities ) , <NUM_LIT:1> ) <EOL> def test_svc_level_dupe_tz_uniform ( self ) : <EOL> '''<STR_LIT>''' <EOL> svcA , svcB = TestTools . create_mock_services ( ) <EOL> recA = TestTools . create_mock_svc_record ( svcA ) <EOL> recB = TestTools . create_mock_svc_record ( svcB ) <EOL> actA = Activity ( ) <EOL> actA . StartTime = pytz . timezone ( "<STR_LIT>" ) . localize ( datetime ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> ) ) <EOL> actA . ServiceDataCollection = TestTools . create_mock_servicedatacollection ( svcA , record = recA ) <EOL> actB = Activity ( ) <EOL> actB . StartTime = actA . StartTime <EOL> actB . ServiceDataCollection = TestTools . create_mock_servicedatacollection ( svcB , record = recB ) <EOL> actA . CalculateUID ( ) <EOL> actB . CalculateUID ( ) <EOL> s = SynchronizationTask ( None ) <EOL> s . _activities = [ ] <EOL> s . _accumulateActivities ( recA , [ actA ] ) <EOL> s . _accumulateActivities ( recB , [ actB ] ) <EOL> self . assertEqual ( len ( s . _activities ) , <NUM_LIT:1> ) <EOL> def test_svc_level_dupe_tz_nonuniform ( self ) : <EOL> '''<STR_LIT>''' <EOL> svcA , svcB = TestTools . create_mock_services ( ) <EOL> recA = TestTools . create_mock_svc_record ( svcA ) <EOL> recB = TestTools . create_mock_svc_record ( svcB ) <EOL> actA = Activity ( ) <EOL> actA . StartTime = datetime ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> ) <EOL> actA . ServiceDataCollection = TestTools . create_mock_servicedatacollection ( svcA , record = recA ) <EOL> actB = Activity ( ) <EOL> actB . StartTime = pytz . timezone ( "<STR_LIT>" ) . localize ( actA . StartTime ) <EOL> actB . ServiceDataCollection = TestTools . create_mock_servicedatacollection ( svcB , record = recB ) <EOL> actA . CalculateUID ( ) <EOL> actB . CalculateUID ( ) <EOL> s = SynchronizationTask ( None ) <EOL> s . _activities = [ ] <EOL> s . _accumulateActivities ( recA , [ actA ] ) <EOL> s . _accumulateActivities ( recB , [ actB ] ) <EOL> self . assertEqual ( len ( s . _activities ) , <NUM_LIT:1> ) <EOL> def test_svc_level_dupe_tz_irregular ( self ) : <EOL> '''<STR_LIT>''' <EOL> svcA , svcB = TestTools . create_mock_services ( ) <EOL> recA = TestTools . create_mock_svc_record ( svcA ) <EOL> recB = TestTools . create_mock_svc_record ( svcB ) <EOL> actA = Activity ( ) <EOL> actA . StartTime = pytz . timezone ( "<STR_LIT>" ) . localize ( datetime ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> ) ) <EOL> actA . ServiceDataCollection = TestTools . create_mock_servicedatacollection ( svcA , record = recA ) <EOL> actB = Activity ( ) <EOL> actB . StartTime = actA . StartTime . astimezone ( pytz . timezone ( "<STR_LIT>" ) ) <EOL> actB . ServiceDataCollection = TestTools . create_mock_servicedatacollection ( svcB , record = recB ) <EOL> actA . CalculateUID ( ) <EOL> actB . CalculateUID ( ) <EOL> s = SynchronizationTask ( None ) <EOL> s . _activities = [ ] <EOL> s . _accumulateActivities ( recA , [ actA ] ) <EOL> s . _accumulateActivities ( recB , [ actB ] ) <EOL> self . assertEqual ( len ( s . _activities ) , <NUM_LIT:1> ) <EOL> def test_svc_level_dupe_time_leeway ( self ) : <EOL> '''<STR_LIT>''' <EOL> svcA , svcB = TestTools . create_mock_services ( ) <EOL> recA = TestTools . create_mock_svc_record ( svcA ) <EOL> recB = TestTools . create_mock_svc_record ( svcB ) <EOL> actA = Activity ( ) <EOL> actA . StartTime = datetime ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> ) <EOL> actA . ServiceDataCollection = TestTools . create_mock_servicedatacollection ( svcA , record = recA ) <EOL> actA . Type = set ( svcA . SupportedActivities ) . intersection ( set ( svcB . SupportedActivities ) ) . pop ( ) <EOL> actB = Activity ( ) <EOL> actB . StartTime = datetime ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:6> , <NUM_LIT:6> , <NUM_LIT:7> ) <EOL> actB . ServiceDataCollection = TestTools . create_mock_servicedatacollection ( svcB , record = recB ) <EOL> actB . Type = actA . Type <EOL> actA . CalculateUID ( ) <EOL> actB . CalculateUID ( ) <EOL> s = SynchronizationTask ( None ) <EOL> s . _activities = [ ] <EOL> s . _accumulateActivities ( recA , [ actA ] ) <EOL> s . _accumulateActivities ( recB , [ actB ] ) <EOL> self . assertIn ( actA . UID , actA . UIDs ) <EOL> self . assertIn ( actB . UID , actA . UIDs ) <EOL> self . assertIn ( actA . UID , actB . UIDs ) <EOL> self . assertIn ( actB . UID , actB . UIDs ) <EOL> recA = TestTools . create_mock_svc_record ( svcA ) <EOL> recB = TestTools . create_mock_svc_record ( svcB ) <EOL> recA . SynchronizedActivities = [ actA . UID ] <EOL> recB . SynchronizedActivities = [ actB . UID ] <EOL> s . _serviceConnections = [ recA , recB ] <EOL> recipientServicesA = s . _determineRecipientServices ( actA ) <EOL> recipientServicesB = s . _determineRecipientServices ( actB ) <EOL> self . assertEqual ( len ( recipientServicesA ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( len ( recipientServicesB ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( len ( s . _activities ) , <NUM_LIT:1> ) <EOL> def test_svc_supported_activity_types ( self ) : <EOL> '''<STR_LIT>''' <EOL> svcA , svcB = TestTools . create_mock_services ( ) <EOL> recA = TestTools . create_mock_svc_record ( svcA ) <EOL> recB = TestTools . create_mock_svc_record ( svcB ) <EOL> svcA . SupportedActivities = [ ActivityType . CrossCountrySkiing ] <EOL> svcB . SupportedActivities = [ ActivityType . Cycling ] <EOL> actA = Activity ( ) <EOL> actA . StartTime = datetime ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> ) <EOL> actA . ServiceDataCollection = TestTools . create_mock_servicedatacollection ( svcA , record = recA ) <EOL> actA . Type = svcA . SupportedActivities [ <NUM_LIT:0> ] <EOL> actA . CalculateUID ( ) <EOL> actA . UIDs = set ( [ actA . UID ] ) <EOL> actA . Record = ActivityRecord . FromActivity ( actA ) <EOL> actB = Activity ( ) <EOL> actB . StartTime = datetime ( <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:7> , <NUM_LIT:8> , <NUM_LIT:9> , <NUM_LIT:10> , <NUM_LIT:11> ) <EOL> actB . ServiceDataCollection = TestTools . create_mock_servicedatacollection ( svcB , record = recB ) <EOL> actB . Type = [ x for x in svcB . SupportedActivities if x != actA . Type ] [ <NUM_LIT:0> ] <EOL> actB . CalculateUID ( ) <EOL> actB . UIDs = set ( [ actB . UID ] ) <EOL> actB . Record = ActivityRecord . FromActivity ( actB ) <EOL> s = SynchronizationTask ( None ) <EOL> s . _serviceConnections = [ recA , recB ] <EOL> s . _activities = [ ] <EOL> s . _accumulateActivities ( recA , [ actA ] ) <EOL> s . _accumulateActivities ( recB , [ actB ] ) <EOL> syncToA = s . _determineRecipientServices ( actA ) <EOL> syncToB = s . _determineRecipientServices ( actB ) <EOL> self . assertEqual ( len ( syncToA ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( len ( syncToB ) , <NUM_LIT:0> ) <EOL> svcB . SupportedActivities = svcA . SupportedActivities <EOL> syncToA = s . _determineRecipientServices ( actA ) <EOL> syncToB = s . _determineRecipientServices ( actB ) <EOL> self . assertEqual ( len ( syncToA ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( syncToB ) , <NUM_LIT:0> ) <EOL> svcB . SupportedActivities = svcA . SupportedActivities = [ ActivityType . CrossCountrySkiing , ActivityType . Cycling ] <EOL> syncToA = s . _determineRecipientServices ( actA ) <EOL> syncToB = s . _determineRecipientServices ( actB ) <EOL> self . assertEqual ( len ( syncToA ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( syncToB ) , <NUM_LIT:1> ) <EOL> def test_accumulate_exclusions ( self ) : <EOL> svcA , svcB = TestTools . create_mock_services ( ) <EOL> recA = TestTools . create_mock_svc_record ( svcA ) <EOL> s = SynchronizationTask ( None ) <EOL> s . _syncExclusions = { recA . _id : { } } <EOL> exc = APIExcludeActivity ( "<STR_LIT>" , activity_id = <NUM_LIT> ) <EOL> s . _accumulateExclusions ( recA , exc ) <EOL> exclusionstore = s . _syncExclusions <EOL> self . assertTrue ( "<STR_LIT>" in exclusionstore [ recA . _id ] ) <EOL> self . assertEqual ( exclusionstore [ recA . _id ] [ "<STR_LIT>" ] [ "<STR_LIT>" ] , "<STR_LIT>" ) <EOL> self . assertEqual ( exclusionstore [ recA . _id ] [ "<STR_LIT>" ] [ "<STR_LIT>" ] , None ) <EOL> self . assertEqual ( exclusionstore [ recA . _id ] [ "<STR_LIT>" ] [ "<STR_LIT>" ] , <NUM_LIT> ) <EOL> self . assertEqual ( exclusionstore [ recA . _id ] [ "<STR_LIT>" ] [ "<STR_LIT>" ] , True ) <EOL> act = TestTools . create_blank_activity ( svcA ) <EOL> act . UID = "<STR_LIT>" <EOL> exc = APIExcludeActivity ( "<STR_LIT>" , activity_id = <NUM_LIT> , permanent = False , activity = act ) <EOL> s = SynchronizationTask ( None ) <EOL> s . _syncExclusions = { recA . _id : { } } <EOL> s . _accumulateExclusions ( recA , exc ) <EOL> exclusionstore = s . _syncExclusions <EOL> self . assertTrue ( "<STR_LIT>" in exclusionstore [ recA . _id ] ) <EOL> self . assertEqual ( exclusionstore [ recA . _id ] [ "<STR_LIT>" ] [ "<STR_LIT>" ] , "<STR_LIT>" ) <EOL> self . assertNotEqual ( exclusionstore [ recA . _id ] [ "<STR_LIT>" ] [ "<STR_LIT>" ] , None ) <EOL> self . assertEqual ( exclusionstore [ recA . _id ] [ "<STR_LIT>" ] [ "<STR_LIT>" ] , <NUM_LIT> ) <EOL> self . assertEqual ( exclusionstore [ recA . _id ] [ "<STR_LIT>" ] [ "<STR_LIT>" ] , False ) <EOL> exc2 = APIExcludeActivity ( "<STR_LIT>" , activity_id = <NUM_LIT> ) <EOL> exc3 = APIExcludeActivity ( "<STR_LIT>" , activity_id = <NUM_LIT> ) <EOL> s . _accumulateExclusions ( recA , [ exc2 , exc3 ] ) <EOL> exclusionstore = s . _syncExclusions <EOL> self . assertTrue ( "<STR_LIT>" in exclusionstore [ recA . _id ] ) <EOL> self . assertTrue ( "<STR_LIT>" in exclusionstore [ recA . _id ] ) <EOL> self . assertTrue ( "<STR_LIT>" in exclusionstore [ recA . _id ] ) <EOL> exc4 = APIExcludeActivity ( "<STR_LIT>" ) <EOL> s = SynchronizationTask ( None ) <EOL> s . _syncExclusions = { } <EOL> self . assertRaises ( ValueError , s . _accumulateExclusions , recA , [ exc4 ] ) <EOL> def test_activity_deduplicate_normaltz ( self ) : <EOL> '''<STR_LIT>''' <EOL> svcA , svcB = TestTools . create_mock_services ( ) <EOL> recA = TestTools . create_mock_svc_record ( svcA ) <EOL> recB = TestTools . create_mock_svc_record ( svcB ) <EOL> actA = TestTools . create_random_activity ( svcA , tz = UTC ( ) ) <EOL> actB = Activity ( ) <EOL> actB . StartTime = actA . StartTime . replace ( tzinfo = None ) + timedelta ( seconds = <NUM_LIT:10> ) <EOL> actB . EndTime = actA . EndTime . replace ( tzinfo = None ) <EOL> actB . ServiceDataCollection = TestTools . create_mock_servicedatacollection ( svcB , record = recB ) <EOL> actA . Name = "<STR_LIT>" <EOL> actB . Name = "<STR_LIT>" <EOL> actB . Type = ActivityType . Walking <EOL> actA . CalculateUID ( ) <EOL> actB . CalculateUID ( ) <EOL> s = SynchronizationTask ( None ) <EOL> s . _activities = [ ] <EOL> s . _accumulateActivities ( recB , [ copy . deepcopy ( actB ) ] ) <EOL> self . assertRaises ( ValueError , s . _accumulateActivities , recA , [ copy . deepcopy ( actA ) ] ) <EOL> def test_activity_deduplicate_tzerror ( self ) : <EOL> '''<STR_LIT>''' <EOL> svcA , svcB = TestTools . create_mock_services ( ) <EOL> recA = TestTools . create_mock_svc_record ( svcA ) <EOL> recB = TestTools . create_mock_svc_record ( svcB ) <EOL> actA = TestTools . create_random_activity ( svcA , tz = pytz . timezone ( "<STR_LIT>" ) ) <EOL> actB = Activity ( ) <EOL> actB . StartTime = actA . StartTime . replace ( tzinfo = pytz . timezone ( "<STR_LIT>" ) ) + timedelta ( hours = <NUM_LIT:5> ) <EOL> actB . ServiceDataCollection = TestTools . create_mock_servicedatacollection ( svcB ) <EOL> actA . Name = "<STR_LIT>" <EOL> actB . Name = "<STR_LIT>" <EOL> actB . Type = ActivityType . Walking <EOL> actA . CalculateUID ( ) <EOL> actB . CalculateUID ( ) <EOL> s = SynchronizationTask ( None ) <EOL> s . _activities = [ ] <EOL> s . _accumulateActivities ( recB , [ copy . deepcopy ( actB ) ] ) <EOL> s . _accumulateActivities ( recA , [ copy . deepcopy ( actA ) ] ) <EOL> self . assertEqual ( len ( s . _activities ) , <NUM_LIT:1> ) <EOL> actB . StartTime = actA . StartTime . replace ( tzinfo = pytz . timezone ( "<STR_LIT>" ) ) + timedelta ( hours = <NUM_LIT:5> , seconds = <NUM_LIT:1> ) <EOL> s . _activities = [ ] <EOL> s . _accumulateActivities ( recB , [ copy . deepcopy ( actB ) ] ) <EOL> s . _accumulateActivities ( recA , [ copy . deepcopy ( actA ) ] ) <EOL> self . assertEqual ( len ( s . _activities ) , <NUM_LIT:1> ) <EOL> actB . StartTime = actA . StartTime . replace ( tzinfo = pytz . timezone ( "<STR_LIT>" ) ) + timedelta ( hours = <NUM_LIT:5> , minutes = <NUM_LIT:7> ) <EOL> s . _activities = [ ] <EOL> s . _accumulateActivities ( recB , [ copy . deepcopy ( actB ) ] ) <EOL> s . _accumulateActivities ( recA , [ copy . deepcopy ( actA ) ] ) <EOL> self . assertEqual ( len ( s . _activities ) , <NUM_LIT:2> ) <EOL> actB . StartTime = actA . StartTime . replace ( tzinfo = pytz . timezone ( "<STR_LIT>" ) ) + timedelta ( hours = <NUM_LIT:50> ) <EOL> s . _activities = [ ] <EOL> s . _accumulateActivities ( recB , [ copy . deepcopy ( actB ) ] ) <EOL> s . _accumulateActivities ( recA , [ copy . deepcopy ( actA ) ] ) <EOL> self . assertEqual ( len ( s . _activities ) , <NUM_LIT:2> ) <EOL> def test_activity_coalesce ( self ) : <EOL> '''<STR_LIT>''' <EOL> svcA , svcB = TestTools . create_mock_services ( ) <EOL> recA = TestTools . create_mock_svc_record ( svcA ) <EOL> recB = TestTools . create_mock_svc_record ( svcB ) <EOL> actA = TestTools . create_random_activity ( svcA , tz = pytz . timezone ( "<STR_LIT>" ) ) <EOL> actB = Activity ( ) <EOL> actB . StartTime = actA . StartTime . replace ( tzinfo = None ) <EOL> actB . ServiceDataCollection = TestTools . create_mock_servicedatacollection ( svcB ) <EOL> actA . Name = "<STR_LIT>" <EOL> actA . Private = True <EOL> actB . Name = "<STR_LIT>" <EOL> actB . Type = ActivityType . Walking <EOL> actA . CalculateUID ( ) <EOL> actB . CalculateUID ( ) <EOL> s = SynchronizationTask ( None ) <EOL> s . _activities = [ ] <EOL> s . _accumulateActivities ( recB , [ copy . deepcopy ( actB ) ] ) <EOL> s . _accumulateActivities ( recA , [ copy . deepcopy ( actA ) ] ) <EOL> self . assertEqual ( len ( s . _activities ) , <NUM_LIT:1> ) <EOL> act = s . _activities [ <NUM_LIT:0> ] <EOL> self . assertEqual ( act . StartTime , actA . StartTime ) <EOL> self . assertEqual ( act . EndTime , actA . EndTime ) <EOL> self . assertEqual ( act . EndTime . tzinfo , actA . StartTime . tzinfo ) <EOL> self . assertEqual ( act . StartTime . tzinfo , actA . StartTime . tzinfo ) <EOL> self . assertLapsListsEqual ( act . Laps , actA . Laps ) <EOL> self . assertTrue ( act . Private ) <EOL> self . assertEqual ( act . Name , actB . Name ) <EOL> self . assertEqual ( act . Type , actB . Type ) <EOL> self . assertTrue ( list ( actB . ServiceDataCollection . keys ( ) ) [ <NUM_LIT:0> ] in act . ServiceDataCollection ) <EOL> self . assertTrue ( list ( actA . ServiceDataCollection . keys ( ) ) [ <NUM_LIT:0> ] in act . ServiceDataCollection ) <EOL> s . _activities = [ ] <EOL> s . _accumulateActivities ( recA , [ copy . deepcopy ( actA ) ] ) <EOL> s . _accumulateActivities ( recB , [ copy . deepcopy ( actB ) ] ) <EOL> self . assertEqual ( len ( s . _activities ) , <NUM_LIT:1> ) <EOL> act = s . _activities [ <NUM_LIT:0> ] <EOL> self . assertEqual ( act . StartTime , actA . StartTime ) <EOL> self . assertEqual ( act . EndTime , actA . EndTime ) <EOL> self . assertEqual ( act . EndTime . tzinfo , actA . StartTime . tzinfo ) <EOL> self . assertEqual ( act . StartTime . tzinfo , actA . StartTime . tzinfo ) <EOL> self . assertLapsListsEqual ( act . Laps , actA . Laps ) <EOL> self . assertEqual ( act . Name , actA . Name ) <EOL> self . assertEqual ( act . Type , actB . Type ) <EOL> self . assertTrue ( list ( actB . ServiceDataCollection . keys ( ) ) [ <NUM_LIT:0> ] in act . ServiceDataCollection ) <EOL> self . assertTrue ( list ( actA . ServiceDataCollection . keys ( ) ) [ <NUM_LIT:0> ] in act . ServiceDataCollection ) <EOL> actA . Type = ActivityType . Hiking <EOL> s . _activities = [ ] <EOL> s . _accumulateActivities ( recA , [ copy . deepcopy ( actA ) ] ) <EOL> s . _accumulateActivities ( recB , [ copy . deepcopy ( actB ) ] ) <EOL> self . assertEqual ( len ( s . _activities ) , <NUM_LIT:1> ) <EOL> act = s . _activities [ <NUM_LIT:0> ] <EOL> self . assertEqual ( act . Type , actA . Type ) <EOL> actA . Type = ActivityType . CrossCountrySkiing <EOL> s . _activities = [ ] <EOL> s . _accumulateActivities ( recA , [ copy . deepcopy ( actA ) ] ) <EOL> s . _accumulateActivities ( recB , [ copy . deepcopy ( actB ) ] ) <EOL> self . assertEqual ( len ( s . _activities ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( s . _activities [ <NUM_LIT:0> ] . Type , actB . Type ) <EOL> self . assertEqual ( s . _activities [ <NUM_LIT:1> ] . Type , actA . Type ) <EOL> def test_eligibility_excluded ( self ) : <EOL> user = TestTools . create_mock_user ( ) <EOL> svcA , svcB = TestTools . create_mock_services ( ) <EOL> recA = TestTools . create_mock_svc_record ( svcA ) <EOL> recB = TestTools . create_mock_svc_record ( svcB ) <EOL> act = TestTools . create_blank_activity ( svcA , record = recB ) <EOL> recipientServices = [ recA , recB ] <EOL> s = SynchronizationTask ( None ) <EOL> s . _excludedServices = { recA . _id : UserException ( UserExceptionType . Private ) } <EOL> s . user = user <EOL> s . _serviceConnections = recipientServices <EOL> act . UIDs = set ( [ act . UID ] ) <EOL> act . Record = ActivityRecord . FromActivity ( act ) <EOL> eligible = s . _determineEligibleRecipientServices ( act , recipientServices ) <EOL> self . assertTrue ( recB in eligible ) <EOL> self . assertTrue ( recA not in eligible ) <EOL> def test_eligibility_config ( self ) : <EOL> user = TestTools . create_mock_user ( ) <EOL> svcA , svcB = TestTools . create_mock_services ( ) <EOL> svcA . Configurable = True <EOL> svcA . RequiresConfiguration = lambda x : True <EOL> recA = TestTools . create_mock_svc_record ( svcA ) <EOL> recB = TestTools . create_mock_svc_record ( svcB ) <EOL> act = TestTools . create_blank_activity ( svcA , record = recB ) <EOL> recipientServices = [ recA , recB ] <EOL> s = SynchronizationTask ( None ) <EOL> s . _excludedServices = { } <EOL> s . user = user <EOL> s . _serviceConnections = recipientServices <EOL> act . UIDs = set ( [ act . UID ] ) <EOL> act . Record = ActivityRecord . FromActivity ( act ) <EOL> eligible = s . _determineEligibleRecipientServices ( act , recipientServices ) <EOL> self . assertTrue ( recB in eligible ) <EOL> self . assertTrue ( recA not in eligible ) <EOL> def test_eligibility_flowexception ( self ) : <EOL> user = TestTools . create_mock_user ( ) <EOL> svcA , svcB = TestTools . create_mock_services ( ) <EOL> recA = TestTools . create_mock_svc_record ( svcA ) <EOL> recB = TestTools . create_mock_svc_record ( svcB ) <EOL> act = TestTools . create_blank_activity ( svcA , record = recA ) <EOL> act . Origin = recA <EOL> act . UIDs = set ( [ act . UID ] ) <EOL> act . Record = ActivityRecord . FromActivity ( act ) <EOL> User . SetFlowException ( user , recA , recB , flowToTarget = False ) <EOL> recipientServices = [ recA , recB ] <EOL> s = SynchronizationTask ( None ) <EOL> s . _excludedServices = { } <EOL> s . user = user <EOL> s . _serviceConnections = recipientServices <EOL> eligible = s . _determineEligibleRecipientServices ( act , recipientServices ) <EOL> self . assertTrue ( recA in eligible ) <EOL> self . assertFalse ( recB in eligible ) <EOL> def test_eligibility_flowexception_shortcircuit ( self ) : <EOL> user = TestTools . create_mock_user ( ) <EOL> svcA , svcB = TestTools . create_mock_services ( ) <EOL> svcC = TestTools . create_mock_service ( "<STR_LIT>" ) <EOL> recA = TestTools . create_mock_svc_record ( svcA ) <EOL> recB = TestTools . create_mock_svc_record ( svcB ) <EOL> recC = TestTools . create_mock_svc_record ( svcC ) <EOL> act = TestTools . create_blank_activity ( svcA , record = recA ) <EOL> User . SetFlowException ( user , recA , recC , flowToTarget = False ) <EOL> act . Origin = recA <EOL> act . UIDs = set ( [ act . UID ] ) <EOL> act . Record = ActivityRecord . FromActivity ( act ) <EOL> recipientServices = [ recC , recB ] <EOL> s = SynchronizationTask ( None ) <EOL> s . _excludedServices = { } <EOL> s . user = user <EOL> s . _serviceConnections = [ recA , recB , recC ] <EOL> eligible = s . _determineEligibleRecipientServices ( act , recipientServices ) <EOL> self . assertTrue ( recA not in eligible ) <EOL> self . assertTrue ( recB in eligible ) <EOL> self . assertTrue ( recC not in eligible ) <EOL> recB . SetConfiguration ( { "<STR_LIT>" : True } , no_save = True ) <EOL> self . assertTrue ( recB . GetConfiguration ( ) [ "<STR_LIT>" ] ) <EOL> act . Origin = recA <EOL> act . UIDs = set ( [ act . UID ] ) <EOL> act . Record = ActivityRecord . FromActivity ( act ) <EOL> recipientServices = [ recC , recB ] <EOL> s . _excludedServices = { } <EOL> s . _serviceConnections = [ recA , recB , recC ] <EOL> eligible = s . _determineEligibleRecipientServices ( act , recipientServices ) <EOL> self . assertTrue ( recA not in eligible ) <EOL> self . assertTrue ( recB in eligible ) <EOL> def test_eligibility_flowexception_reverse ( self ) : <EOL> user = TestTools . create_mock_user ( ) <EOL> svcA , svcB = TestTools . create_mock_services ( ) <EOL> recA = TestTools . create_mock_svc_record ( svcA ) <EOL> recB = TestTools . create_mock_svc_record ( svcB ) <EOL> act = TestTools . create_blank_activity ( svcA , record = recB ) <EOL> act . Origin = recB <EOL> act . UIDs = set ( [ act . UID ] ) <EOL> act . Record = ActivityRecord . FromActivity ( act ) <EOL> User . SetFlowException ( user , recA , recB , flowToSource = False ) <EOL> recipientServices = [ recA , recB ] <EOL> s = SynchronizationTask ( None ) <EOL> s . _excludedServices = { } <EOL> s . user = user <EOL> s . _serviceConnections = recipientServices <EOL> eligible = s . _determineEligibleRecipientServices ( act , recipientServices ) <EOL> self . assertFalse ( recA in eligible ) <EOL> self . assertTrue ( recB in eligible ) <EOL> def test_eligibility_flowexception_both ( self ) : <EOL> user = TestTools . create_mock_user ( ) <EOL> svcA , svcB = TestTools . create_mock_services ( ) <EOL> recA = TestTools . create_mock_svc_record ( svcA ) <EOL> recB = TestTools . create_mock_svc_record ( svcB ) <EOL> act = TestTools . create_blank_activity ( svcA , record = recB ) <EOL> act . Origin = recB <EOL> act . UIDs = set ( [ act . UID ] ) <EOL> act . Record = ActivityRecord . FromActivity ( act ) <EOL> User . SetFlowException ( user , recA , recB , flowToSource = False , flowToTarget = False ) <EOL> recipientServices = [ recA , recB ] <EOL> s = SynchronizationTask ( None ) <EOL> s . _excludedServices = { } <EOL> s . user = user <EOL> s . _serviceConnections = recipientServices <EOL> eligible = s . _determineEligibleRecipientServices ( act , recipientServices ) <EOL> self . assertFalse ( recA in eligible ) <EOL> self . assertTrue ( recB in eligible ) <EOL> act . Origin = recA <EOL> act . ServiceDataCollection = TestTools . create_mock_servicedatacollection ( svcA , record = recA ) <EOL> eligible = s . _determineEligibleRecipientServices ( act , recipientServices ) <EOL> self . assertTrue ( recA in eligible ) <EOL> self . assertFalse ( recB in eligible ) <EOL> def test_eligibility_flowexception_none ( self ) : <EOL> user = TestTools . create_mock_user ( ) <EOL> svcA , svcB = TestTools . create_mock_services ( ) <EOL> recA = TestTools . create_mock_svc_record ( svcA ) <EOL> recB = TestTools . create_mock_svc_record ( svcB ) <EOL> act = TestTools . create_blank_activity ( svcA , record = recB ) <EOL> act . Origin = recB <EOL> act . UIDs = set ( [ act . UID ] ) <EOL> act . Record = ActivityRecord . FromActivity ( act ) <EOL> User . SetFlowException ( user , recA , recB , flowToSource = False , flowToTarget = False ) <EOL> recipientServices = [ recA ] <EOL> s = SynchronizationTask ( None ) <EOL> s . _excludedServices = { } <EOL> s . user = user <EOL> s . _serviceConnections = [ recA , recB ] <EOL> eligible = s . _determineEligibleRecipientServices ( act , recipientServices ) <EOL> self . assertTrue ( recA not in eligible ) <EOL> self . assertTrue ( recB not in eligible ) <EOL> recipientServices = [ recB ] <EOL> act . Origin = recA <EOL> act . ServiceDataCollection = TestTools . create_mock_servicedatacollection ( svcA , record = recA ) <EOL> eligible = s . _determineEligibleRecipientServices ( act , recipientServices ) <EOL> self . assertTrue ( recA not in eligible ) <EOL> self . assertTrue ( recB not in eligible ) <EOL> def test_eligibility_flowexception_change ( self ) : <EOL> user = TestTools . create_mock_user ( ) <EOL> svcA , svcB = TestTools . create_mock_services ( ) <EOL> recA = TestTools . create_mock_svc_record ( svcA ) <EOL> recB = TestTools . create_mock_svc_record ( svcB ) <EOL> act = TestTools . create_blank_activity ( svcA , record = recB ) <EOL> act . Origin = recB <EOL> act . UIDs = set ( [ act . UID ] ) <EOL> act . Record = ActivityRecord . FromActivity ( act ) <EOL> recipientServices = [ recA ] <EOL> s = SynchronizationTask ( None ) <EOL> s . _excludedServices = { } <EOL> s . user = user <EOL> s . _serviceConnections = [ recA , recB ] <EOL> User . SetFlowException ( user , recA , recB , flowToSource = False , flowToTarget = True ) <EOL> eligible = s . _determineEligibleRecipientServices ( act , recipientServices ) <EOL> self . assertTrue ( recA not in eligible ) <EOL> self . assertTrue ( recB not in eligible ) <EOL> recipientServices = [ recB ] <EOL> act . Origin = recA <EOL> act . ServiceDataCollection = TestTools . create_mock_servicedatacollection ( svcA , record = recA ) <EOL> User . SetFlowException ( user , recA , recB , flowToSource = True , flowToTarget = False ) <EOL> eligible = s . _determineEligibleRecipientServices ( act , recipientServices ) <EOL> self . assertTrue ( recA not in eligible ) <EOL> self . assertTrue ( recB not in eligible ) <EOL> User . SetFlowException ( user , recA , recB , flowToSource = False , flowToTarget = False ) <EOL> eligible = s . _determineEligibleRecipientServices ( act , recipientServices ) <EOL> self . assertTrue ( recA not in eligible ) <EOL> self . assertTrue ( recB not in eligible ) <EOL> recipientServices = [ recA , recB ] <EOL> User . SetFlowException ( user , recA , recB , flowToSource = True , flowToTarget = True ) <EOL> eligible = s . _determineEligibleRecipientServices ( act , recipientServices ) <EOL> self . assertTrue ( recA in eligible ) <EOL> self . assertTrue ( recB in eligible ) <EOL> eligible = s . _determineEligibleRecipientServices ( act , recipientServices ) <EOL> self . assertTrue ( recA in eligible ) <EOL> self . assertTrue ( recB in eligible ) </s>
