Python sync.Sync类代码示例

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Python›Python编程经验

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Python中tapiriik.sync.Sync类的典型用法代码示例。如果您正苦于以下问题：Python Sync类的具体用法？Python Sync怎么用？Python Sync使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

在下文中一共展示了Sync类的19个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: test_eligibility_flowexception_shortcircuit

    def test_eligibility_flowexception_shortcircuit(self):
        user = TestTools.create_mock_user()
        svcA, svcB = TestTools.create_mock_services()
        svcC = TestTools.create_mock_service("mockC")
        recA = TestTools.create_mock_svc_record(svcA)
        recB = TestTools.create_mock_svc_record(svcB)
        recC = TestTools.create_mock_svc_record(svcC)
        act = TestTools.create_blank_activity(svcA, record=recA)
        User.SetFlowException(user, recA, recC, flowToTarget=False)

        # Behaviour with known origin and no override set
        act.Origin = recA
        recipientServices = [recC, recB]
        excludedServices = []
        eligible = Sync._determineEligibleRecipientServices(activity=act, connectedServices=[recA, recB, recC], recipientServices=recipientServices, excludedServices=excludedServices, user=user)
        self.assertTrue(recA not in eligible)
        self.assertTrue(recB in eligible)
        self.assertTrue(recC not in eligible)

        # Enable alternate routing
        recB.SetConfiguration({"allow_activity_flow_exception_bypass_via_self":True}, no_save=True)
        self.assertTrue(recB.GetConfiguration()["allow_activity_flow_exception_bypass_via_self"])
        # We should now be able to arrive at recC via recB
        act.Origin = recA
        recipientServices = [recC, recB]
        excludedServices = []
        eligible = Sync._determineEligibleRecipientServices(activity=act, connectedServices=[recA, recB, recC], recipientServices=recipientServices, excludedServices=excludedServices, user=user)
        self.assertTrue(recA not in eligible)
        self.assertTrue(recB in eligible)
        self.assertTrue(recC in eligible)

开发者ID:7e7，项目名称:tapiriik，代码行数:30，代码来源:sync.py

示例2: test_svc_level_dupe

    def test_svc_level_dupe(self):
        ''' check that service-level duplicate activities are caught (no DB involvement) '''

        svcA, svcB = TestTools.create_mock_services()
        recA = TestTools.create_mock_svc_record(svcA)
        recB = TestTools.create_mock_svc_record(svcB)

        actA = Activity()
        actA.StartTime = datetime(1, 2, 3, 4, 5, 6, 7)
        actA.ServiceDataCollection = TestTools.create_mock_servicedatacollection(svcA, record=recA)
        actB = Activity()
        actB.StartTime = actA.StartTime
        actB.ServiceDataCollection = TestTools.create_mock_servicedatacollection(svcB, record=recB)

        actA.CalculateUID()
        actB.CalculateUID()

        activities = []



        Sync._accumulateActivities(recA, [actA], activities)
        Sync._accumulateActivities(recB, [actB], activities)

        self.assertEqual(len(activities), 1)

开发者ID:7e7，项目名称:tapiriik，代码行数:25，代码来源:sync.py

示例3: test_svc_supported_activity_types

    def test_svc_supported_activity_types(self):
        ''' check that only activities are only sent to services which support them '''
        svcA, svcB = TestTools.create_mock_services()
        recA = TestTools.create_mock_svc_record(svcA)
        recB = TestTools.create_mock_svc_record(svcB)
        svcA.SupportedActivities = [ActivityType.CrossCountrySkiing]
        svcB.SupportedActivities = [ActivityType.Cycling]

        actA = Activity()
        actA.StartTime = datetime(1, 2, 3, 4, 5, 6, 7)
        actA.ServiceDataCollection = TestTools.create_mock_servicedatacollection(svcA, record=recA)
        actA.Type = svcA.SupportedActivities[0]
        actB = Activity()
        actB.StartTime = datetime(5, 6, 7, 8, 9, 10, 11)
        actB.ServiceDataCollection = TestTools.create_mock_servicedatacollection(svcB, record=recB)
        actB.Type = [x for x in svcB.SupportedActivities if x != actA.Type][0]

        actA.CalculateUID()
        actB.CalculateUID()

        allConns = [recA, recB]

        activities = []
        Sync._accumulateActivities(recA, [actA], activities)
        Sync._accumulateActivities(recB, [actB], activities)

        syncToA = Sync._determineRecipientServices(actA, allConns)
        syncToB = Sync._determineRecipientServices(actB, allConns)

        self.assertEqual(len(syncToA), 0)
        self.assertEqual(len(syncToB), 0)

        svcB.SupportedActivities = svcA.SupportedActivities

        syncToA = Sync._determineRecipientServices(actA, allConns)
        syncToB = Sync._determineRecipientServices(actB, allConns)

        self.assertEqual(len(syncToA), 1)
        self.assertEqual(len(syncToB), 0)

        svcB.SupportedActivities = svcA.SupportedActivities = [ActivityType.CrossCountrySkiing, ActivityType.Cycling]

        syncToA = Sync._determineRecipientServices(actA, allConns)
        syncToB = Sync._determineRecipientServices(actB, allConns)

        self.assertEqual(len(syncToA), 1)
        self.assertEqual(len(syncToB), 1)

开发者ID:7e7，项目名称:tapiriik，代码行数:47，代码来源:sync.py

示例4: test_svc_level_dupe_tz_irregular

    def test_svc_level_dupe_tz_irregular(self):
        ''' check that service-level duplicate activities with irregular TZs are caught '''
        svcA, svcB = TestTools.create_mock_services()
        actA = Activity()
        actA.StartTime = pytz.timezone("America/Edmonton").localize(datetime(1, 2, 3, 4, 5, 6, 7))
        actA.UploadedTo = [TestTools.create_mock_upload_record(svcA)]
        actB = Activity()
        actB.StartTime = actA.StartTime.astimezone(pytz.timezone("America/Iqaluit"))
        actB.UploadedTo = [TestTools.create_mock_upload_record(svcB)]

        actA.CalculateUID()
        actB.CalculateUID()

        activities = []
        Sync._accumulateActivities(Service.FromID("mockA"), [actA], activities)
        Sync._accumulateActivities(Service.FromID("mockB"), [actB], activities)

        self.assertEqual(len(activities), 1)

开发者ID:euphoria，项目名称:tapiriik，代码行数:18，代码来源:sync.py

示例5: test_svc_level_dupe_tz_nonuniform

    def test_svc_level_dupe_tz_nonuniform(self):
        ''' check that service-level duplicate activities with non-uniform TZs are caught '''
        svcA, svcB = TestTools.create_mock_services()
        actA = Activity()
        actA.StartTime = datetime(1, 2, 3, 4, 5, 6, 7)
        actA.UploadedTo = [TestTools.create_mock_upload_record(svcA)]
        actB = Activity()
        actB.StartTime = pytz.timezone("America/Denver").localize(actA.StartTime)
        actB.UploadedTo = [TestTools.create_mock_upload_record(svcB)]

        actA.CalculateUID()
        actB.CalculateUID()

        activities = []
        Sync._accumulateActivities(Service.FromID("mockA"), [actA], activities)
        Sync._accumulateActivities(Service.FromID("mockB"), [actB], activities)

        self.assertEqual(len(activities), 1)

开发者ID:euphoria，项目名称:tapiriik，代码行数:18，代码来源:sync.py

示例6: test_svc_level_dupe

    def test_svc_level_dupe(self):
        ''' check that service-level duplicate activities are caught (no DB involvement) '''
        svcA, svcB = TestTools.create_mock_services()
        actA = Activity()
        actA.StartTime = datetime(1, 2, 3, 4, 5, 6, 7)
        actA.UploadedTo = [TestTools.create_mock_upload_record(svcA)]
        actB = Activity()
        actB.StartTime = actA.StartTime
        actB.UploadedTo = [TestTools.create_mock_upload_record(svcB)]

        actA.CalculateUID()
        actB.CalculateUID()

        activities = []
        Sync._accumulateActivities(Service.FromID("mockA"), [actA], activities)
        Sync._accumulateActivities(Service.FromID("mockB"), [actB], activities)

        self.assertEqual(len(activities), 1)

开发者ID:euphoria，项目名称:tapiriik，代码行数:18，代码来源:sync.py

示例7: test_activity_deduplicate_normaltz

    def test_activity_deduplicate_normaltz(self):
        ''' ensure that we can't deduplicate activities with non-pytz timezones '''
        svcA, svcB = TestTools.create_mock_services()
        actA = TestTools.create_random_activity(svcA, tz=UTC())

        actB = Activity()
        actB.StartTime = actA.StartTime.replace(tzinfo=None) + timedelta(seconds=10)
        actB.EndTime = actA.EndTime.replace(tzinfo=None)
        actB.UploadedTo = [TestTools.create_mock_upload_record(svcB)]
        actA.Name = "Not this"
        actB.Name = "Heya"
        actB.Type = ActivityType.Walking
        actA.CalculateUID()
        actB.CalculateUID()

        activities = []
        Sync._accumulateActivities(Service.FromID("mockB"), [copy.deepcopy(actB)], activities)
        self.assertRaises(ValueError, Sync._accumulateActivities, Service.FromID("mockA"), [copy.deepcopy(actA)], activities)

开发者ID:euphoria，项目名称:tapiriik，代码行数:18，代码来源:sync.py

示例8: test_eligibility_flowexception_both

    def test_eligibility_flowexception_both(self):
        user = TestTools.create_mock_user()
        svcA, svcB = TestTools.create_mock_services()
        recA = TestTools.create_mock_svc_record(svcA)
        recB = TestTools.create_mock_svc_record(svcB)
        act = TestTools.create_blank_activity(svcA, record=recB)
        act.Origin = recB
        User.SetFlowException(user, recA, recB, flowToSource=False, flowToTarget=False)
        recipientServices = [recA, recB]
        excludedServices = []
        eligible = Sync._determineEligibleRecipientServices(activity=act, connectedServices=recipientServices, recipientServices=recipientServices, excludedServices=excludedServices, user=user)
        self.assertFalse(recA in eligible)
        self.assertTrue(recB in eligible)

        act.Origin = recA
        act.ServiceDataCollection = TestTools.create_mock_servicedatacollection(svcA, record=recA)
        eligible = Sync._determineEligibleRecipientServices(activity=act, connectedServices=recipientServices, recipientServices=recipientServices, excludedServices=excludedServices, user=user)
        self.assertTrue(recA in eligible)
        self.assertFalse(recB in eligible)

开发者ID:7e7，项目名称:tapiriik，代码行数:19，代码来源:sync.py

示例9: test_eligibility_flowexception_none

    def test_eligibility_flowexception_none(self):
        user = TestTools.create_mock_user()
        svcA, svcB = TestTools.create_mock_services()
        recA = TestTools.create_mock_svc_record(svcA)
        recB = TestTools.create_mock_svc_record(svcB)
        act = TestTools.create_blank_activity(svcA, record=recB)
        act.Origin = recB
        User.SetFlowException(user, recA, recB, flowToSource=False, flowToTarget=False)
        recipientServices = [recA]
        excludedServices = []
        eligible = Sync._determineEligibleRecipientServices(activity=act, recipientServices=recipientServices, excludedServices=excludedServices, user=user)
        self.assertTrue(recA not in eligible)
        self.assertTrue(recB not in eligible)

        recipientServices = [recB]
        act.Origin = recA
        act.UploadedTo = [TestTools.create_mock_upload_record(svcA, record=recA)]
        eligible = Sync._determineEligibleRecipientServices(activity=act, recipientServices=recipientServices, excludedServices=excludedServices, user=user)
        self.assertTrue(recA not in eligible)
        self.assertTrue(recB not in eligible)

开发者ID:euphoria，项目名称:tapiriik，代码行数:20，代码来源:sync.py

示例10: test_svc_level_dupe_tz_irregular

    def test_svc_level_dupe_tz_irregular(self):
        ''' check that service-level duplicate activities with irregular TZs are caught '''
        svcA, svcB = TestTools.create_mock_services()
        recA = TestTools.create_mock_svc_record(svcA)
        recB = TestTools.create_mock_svc_record(svcB)
        actA = Activity()
        actA.StartTime = pytz.timezone("America/Edmonton").localize(datetime(1, 2, 3, 4, 5, 6, 7))
        actA.ServiceDataCollection = TestTools.create_mock_servicedatacollection(svcA, record=recA)
        actB = Activity()
        actB.StartTime = actA.StartTime.astimezone(pytz.timezone("America/Iqaluit"))
        actB.ServiceDataCollection = TestTools.create_mock_servicedatacollection(svcB, record=recB)

        actA.CalculateUID()
        actB.CalculateUID()

        activities = []
        Sync._accumulateActivities(recA, [actA], activities)
        Sync._accumulateActivities(recB, [actB], activities)

        self.assertEqual(len(activities), 1)

开发者ID:7e7，项目名称:tapiriik，代码行数:20，代码来源:sync.py

示例11: test_svc_level_dupe_tz_nonuniform

    def test_svc_level_dupe_tz_nonuniform(self):
        ''' check that service-level duplicate activities with non-uniform TZs are caught '''
        svcA, svcB = TestTools.create_mock_services()
        recA = TestTools.create_mock_svc_record(svcA)
        recB = TestTools.create_mock_svc_record(svcB)
        actA = Activity()
        actA.StartTime = datetime(1, 2, 3, 4, 5, 6, 7)
        actA.ServiceDataCollection = TestTools.create_mock_servicedatacollection(svcA, record=recA)
        actB = Activity()
        actB.StartTime = pytz.timezone("America/Denver").localize(actA.StartTime)
        actB.ServiceDataCollection = TestTools.create_mock_servicedatacollection(svcB, record=recB)

        actA.CalculateUID()
        actB.CalculateUID()

        activities = []
        Sync._accumulateActivities(recA, [actA], activities)
        Sync._accumulateActivities(recB, [actB], activities)

        self.assertEqual(len(activities), 1)

开发者ID:7e7，项目名称:tapiriik，代码行数:20，代码来源:sync.py

示例12: test_eligibility_excluded

 def test_eligibility_excluded(self):
     user = TestTools.create_mock_user()
     svcA, svcB = TestTools.create_mock_services()
     recA = TestTools.create_mock_svc_record(svcA)
     recB = TestTools.create_mock_svc_record(svcB)
     act = TestTools.create_blank_activity(svcA, record=recB)
     recipientServices = [recA, recB]
     excludedServices = [recA]
     eligible = Sync._determineEligibleRecipientServices(activity=act, connectedServices=recipientServices, recipientServices=recipientServices, excludedServices=excludedServices, user=user)
     self.assertTrue(recB in eligible)
     self.assertTrue(recA not in eligible)

开发者ID:7e7，项目名称:tapiriik，代码行数:11，代码来源:sync.py

示例13: test_eligibility_config

 def test_eligibility_config(self):
     user = TestTools.create_mock_user()
     svcA, svcB = TestTools.create_mock_services()
     svcA.Configurable = True
     svcA.RequiresConfiguration = lambda x: True
     recA = TestTools.create_mock_svc_record(svcA)
     recB = TestTools.create_mock_svc_record(svcB)
     act = TestTools.create_blank_activity(svcA, record=recB)
     recipientServices = [recA, recB]
     excludedServices = []
     eligible = Sync._determineEligibleRecipientServices(activity=act, connectedServices=recipientServices, recipientServices=recipientServices, excludedServices=excludedServices, user=user)
     self.assertTrue(recB in eligible)
     self.assertTrue(recA not in eligible)

开发者ID:7e7，项目名称:tapiriik，代码行数:13，代码来源:sync.py

示例14: test_svc_level_dupe_time_leeway

    def test_svc_level_dupe_time_leeway(self):
        ''' check that service-level duplicate activities within the defined time leeway are caught '''
        svcA, svcB = TestTools.create_mock_services()
        recA = TestTools.create_mock_svc_record(svcA)
        recB = TestTools.create_mock_svc_record(svcB)
        actA = Activity()
        actA.StartTime = datetime(1, 2, 3, 4, 5, 6, 7)
        actA.ServiceDataCollection = TestTools.create_mock_servicedatacollection(svcA, record=recA)
        actA.Type = set(svcA.SupportedActivities).intersection(set(svcB.SupportedActivities)).pop()
        actB = Activity()
        actB.StartTime = datetime(1, 2, 3, 4, 6, 6, 7)
        actB.ServiceDataCollection = TestTools.create_mock_servicedatacollection(svcB, record=recB)
        actB.Type = actA.Type

        actA.CalculateUID()
        actB.CalculateUID()

        activities = []
        Sync._accumulateActivities(recA, [actA], activities)
        Sync._accumulateActivities(recB, [actB], activities)

        self.assertIn(actA.UID, actA.UIDs)
        self.assertIn(actB.UID, actA.UIDs)
        self.assertIn(actA.UID, actB.UIDs)
        self.assertIn(actB.UID, actB.UIDs)

        # we need to fake up the service records to avoid having to call the actual sync method where these values are normally preset
        recA = TestTools.create_mock_svc_record(svcA)
        recB = TestTools.create_mock_svc_record(svcB)
        recA.SynchronizedActivities = [actA.UID]
        recB.SynchronizedActivities = [actB.UID]

        recipientServicesA = Sync._determineRecipientServices(actA, [recA, recB])
        recipientServicesB = Sync._determineRecipientServices(actB, [recA, recB])

        self.assertEqual(len(recipientServicesA), 0)
        self.assertEqual(len(recipientServicesB), 0)
        self.assertEqual(len(activities), 1)

开发者ID:7e7，项目名称:tapiriik，代码行数:38，代码来源:sync.py

示例15: test_accumulate_exclusions

    def test_accumulate_exclusions(self):
        svcA, svcB = TestTools.create_mock_services()
        recA = TestTools.create_mock_svc_record(svcA)

        exclusionstore = {recA._id: {}}
        # regular
        exc = APIExcludeActivity("Messag!e", activityId=3.14)
        Sync._accumulateExclusions(recA, exc, exclusionstore)
        self.assertTrue("3_14" in exclusionstore[recA._id])
        self.assertEqual(exclusionstore[recA._id]["3_14"]["Message"], "Messag!e")
        self.assertEqual(exclusionstore[recA._id]["3_14"]["Activity"], None)
        self.assertEqual(exclusionstore[recA._id]["3_14"]["ExternalActivityID"], 3.14)
        self.assertEqual(exclusionstore[recA._id]["3_14"]["Permanent"], True)

        # updating
        act = TestTools.create_blank_activity(svcA)
        act.UID = "3_14"  # meh
        exc = APIExcludeActivity("Messag!e2", activityId=42, permanent=False, activity=act)
        Sync._accumulateExclusions(recA, exc, exclusionstore)
        self.assertTrue("3_14" in exclusionstore[recA._id])
        self.assertEqual(exclusionstore[recA._id]["3_14"]["Message"], "Messag!e2")
        self.assertNotEqual(exclusionstore[recA._id]["3_14"]["Activity"], None)  # Who knows what the string format will be down the road?
        self.assertEqual(exclusionstore[recA._id]["3_14"]["ExternalActivityID"], 42)
        self.assertEqual(exclusionstore[recA._id]["3_14"]["Permanent"], False)

        # multiple, retaining existing
        exc2 = APIExcludeActivity("INM", activityId=13)
        exc3 = APIExcludeActivity("FNIM", activityId=37)
        Sync._accumulateExclusions(recA, [exc2, exc3], exclusionstore)
        self.assertTrue("3_14" in exclusionstore[recA._id])
        self.assertTrue("37" in exclusionstore[recA._id])
        self.assertTrue("13" in exclusionstore[recA._id])

        # don't allow with no identifiers
        exc4 = APIExcludeActivity("nooooo")
        self.assertRaises(ValueError, Sync._accumulateExclusions, recA, [exc4], exclusionstore)

开发者ID:7e7，项目名称:tapiriik，代码行数:36，代码来源:sync.py

示例16: test_activity_coalesce

    def test_activity_coalesce(self):
        ''' ensure that activity data is getting coalesced by _accumulateActivities '''
        svcA, svcB = TestTools.create_mock_services()
        actA = TestTools.create_random_activity(svcA, tz=pytz.timezone("America/Iqaluit"))
        actB = Activity()
        actB.StartTime = actA.StartTime.replace(tzinfo=None)
        actB.UploadedTo = [TestTools.create_mock_upload_record(svcB)]
        actA.Name = "Not this"
        actB.Name = "Heya"
        actB.Type = ActivityType.Walking
        actA.CalculateUID()
        actB.CalculateUID()

        activities = []
        Sync._accumulateActivities(Service.FromID("mockB"), [copy.deepcopy(actB)], activities)
        Sync._accumulateActivities(Service.FromID("mockA"), [copy.deepcopy(actA)], activities)

        self.assertEqual(len(activities), 1)
        act = activities[0]

        self.assertEqual(act.StartTime, actA.StartTime)
        self.assertEqual(act.EndTime, actA.EndTime)
        self.assertEqual(act.EndTime.tzinfo, actA.StartTime.tzinfo)
        self.assertEqual(act.StartTime.tzinfo, actA.StartTime.tzinfo)
        self.assertEqual(act.Waypoints, actA.Waypoints)
        self.assertEqual(act.Name, actB.Name)  # The first activity takes priority.
        self.assertEqual(act.Type, actB.Type)  # Same here.
        self.assertTrue(actB.UploadedTo[0] in act.UploadedTo)
        self.assertTrue(actA.UploadedTo[0] in act.UploadedTo)

        activities = []
        Sync._accumulateActivities(Service.FromID("mockA"), [copy.deepcopy(actA)], activities)
        Sync._accumulateActivities(Service.FromID("mockB"), [copy.deepcopy(actB)], activities)

        self.assertEqual(len(activities), 1)
        act = activities[0]

        self.assertEqual(act.StartTime, actA.StartTime)
        self.assertEqual(act.EndTime, actA.EndTime)
        self.assertEqual(act.EndTime.tzinfo, actA.StartTime.tzinfo)
        self.assertEqual(act.StartTime.tzinfo, actA.StartTime.tzinfo)
        self.assertEqual(act.Waypoints, actA.Waypoints)
        self.assertEqual(act.Name, actA.Name)  # The first activity takes priority.
        self.assertEqual(act.Type, actB.Type)  # Exception: ActivityType.Other does not take priority
        self.assertTrue(actB.UploadedTo[0] in act.UploadedTo)
        self.assertTrue(actA.UploadedTo[0] in act.UploadedTo)

        actA.Type = ActivityType.CrossCountrySkiing
        activities = []
        Sync._accumulateActivities(Service.FromID("mockA"), [copy.deepcopy(actA)], activities)
        Sync._accumulateActivities(Service.FromID("mockB"), [copy.deepcopy(actB)], activities)

        self.assertEqual(len(activities), 1)
        act = activities[0]
        self.assertEqual(act.Type, actA.Type)  # Here, it will take priority.

开发者ID:euphoria，项目名称:tapiriik，代码行数:55，代码来源:sync.py

示例17: test_activity_coalesce

    def test_activity_coalesce(self):
        ''' ensure that activity data is getting coalesced by _accumulateActivities '''
        svcA, svcB = TestTools.create_mock_services()
        recA = TestTools.create_mock_svc_record(svcA)
        recB = TestTools.create_mock_svc_record(svcB)
        actA = TestTools.create_random_activity(svcA, tz=pytz.timezone("America/Iqaluit"))
        actB = Activity()
        actB.StartTime = actA.StartTime.replace(tzinfo=None)
        actB.ServiceDataCollection = TestTools.create_mock_servicedatacollection(svcB)
        actA.Name = "Not this"
        actA.Private = True
        actB.Name = "Heya"
        actB.Type = ActivityType.Walking
        actA.CalculateUID()
        actB.CalculateUID()

        activities = []
        Sync._accumulateActivities(recB, [copy.deepcopy(actB)], activities)
        Sync._accumulateActivities(recA, [copy.deepcopy(actA)], activities)

        self.assertEqual(len(activities), 1)
        act = activities[0]

        self.assertEqual(act.StartTime, actA.StartTime)
        self.assertEqual(act.EndTime, actA.EndTime)
        self.assertEqual(act.EndTime.tzinfo, actA.StartTime.tzinfo)
        self.assertEqual(act.StartTime.tzinfo, actA.StartTime.tzinfo)
        self.assertEqual(act.Waypoints, actA.Waypoints)
        self.assertTrue(act.Private)  # Most restrictive setting
        self.assertEqual(act.Name, actB.Name)  # The first activity takes priority.
        self.assertEqual(act.Type, actB.Type)  # Same here.
        self.assertTrue(list(actB.ServiceDataCollection.keys())[0] in act.ServiceDataCollection)
        self.assertTrue(list(actA.ServiceDataCollection.keys())[0] in act.ServiceDataCollection)

        activities = []
        Sync._accumulateActivities(recA, [copy.deepcopy(actA)], activities)
        Sync._accumulateActivities(recB, [copy.deepcopy(actB)], activities)

        self.assertEqual(len(activities), 1)
        act = activities[0]

        self.assertEqual(act.StartTime, actA.StartTime)
        self.assertEqual(act.EndTime, actA.EndTime)
        self.assertEqual(act.EndTime.tzinfo, actA.StartTime.tzinfo)
        self.assertEqual(act.StartTime.tzinfo, actA.StartTime.tzinfo)
        self.assertEqual(act.Waypoints, actA.Waypoints)
        self.assertEqual(act.Name, actA.Name)  # The first activity takes priority.
        self.assertEqual(act.Type, actB.Type)  # Exception: ActivityType.Other does not take priority
        self.assertTrue(list(actB.ServiceDataCollection.keys())[0] in act.ServiceDataCollection)
        self.assertTrue(list(actA.ServiceDataCollection.keys())[0] in act.ServiceDataCollection)

        actA.Type = ActivityType.CrossCountrySkiing
        activities = []
        Sync._accumulateActivities(recA, [copy.deepcopy(actA)], activities)
        Sync._accumulateActivities(recB, [copy.deepcopy(actB)], activities)

        self.assertEqual(len(activities), 1)
        act = activities[0]
        self.assertEqual(act.Type, actA.Type)  # Here, it will take priority.

开发者ID:7e7，项目名称:tapiriik，代码行数:59，代码来源:sync.py

示例18: test_activity_deduplicate_tzerror

    def test_activity_deduplicate_tzerror(self):
        ''' Test that probably-duplicate activities with starttimes like 09:12:22 and 15:12:22 (on the same day) are recognized as one '''
        svcA, svcB = TestTools.create_mock_services()
        actA = TestTools.create_random_activity(svcA, tz=pytz.timezone("America/Iqaluit"))
        actB = Activity()
        actB.StartTime = actA.StartTime.replace(tzinfo=pytz.timezone("America/Denver")) + timedelta(hours=5)
        actB.UploadedTo = [TestTools.create_mock_upload_record(svcB)]
        actA.Name = "Not this"
        actB.Name = "Heya"
        actB.Type = ActivityType.Walking
        actA.CalculateUID()
        actB.CalculateUID()

        activities = []
        Sync._accumulateActivities(Service.FromID("mockB"), [copy.deepcopy(actB)], activities)
        Sync._accumulateActivities(Service.FromID("mockA"), [copy.deepcopy(actA)], activities)

        self.assertEqual(len(activities), 1)

        # Ensure that it is an exact match
        actB.StartTime = actA.StartTime.replace(tzinfo=pytz.timezone("America/Denver")) + timedelta(hours=5, seconds=1)
        activities = []
        Sync._accumulateActivities(Service.FromID("mockB"), [copy.deepcopy(actB)], activities)
        Sync._accumulateActivities(Service.FromID("mockA"), [copy.deepcopy(actA)], activities)

        self.assertEqual(len(activities), 2)

        # Ensure that overly large differences >14hr - not possible via TZ differences - are not deduplicated
        actB.StartTime = actA.StartTime.replace(tzinfo=pytz.timezone("America/Denver")) + timedelta(hours=15)
        activities = []
        Sync._accumulateActivities(Service.FromID("mockB"), [copy.deepcopy(actB)], activities)
        Sync._accumulateActivities(Service.FromID("mockA"), [copy.deepcopy(actA)], activities)

        self.assertEqual(len(activities), 2)

开发者ID:euphoria，项目名称:tapiriik，代码行数:34，代码来源:sync.py

示例19: test_activity_deduplicate_tzerror

    def test_activity_deduplicate_tzerror(self):
        ''' Test that probably-duplicate activities with starttimes like 09:12:22 and 15:12:22 (on the same day) are recognized as one '''
        svcA, svcB = TestTools.create_mock_services()
        recA = TestTools.create_mock_svc_record(svcA)
        recB = TestTools.create_mock_svc_record(svcB)
        actA = TestTools.create_random_activity(svcA, tz=pytz.timezone("America/Iqaluit"))
        actB = Activity()
        actB.StartTime = actA.StartTime.replace(tzinfo=pytz.timezone("America/Denver")) + timedelta(hours=5)
        actB.ServiceDataCollection = TestTools.create_mock_servicedatacollection(svcB)
        actA.Name = "Not this"
        actB.Name = "Heya"
        actB.Type = ActivityType.Walking
        actA.CalculateUID()
        actB.CalculateUID()

        activities = []
        Sync._accumulateActivities(recB, [copy.deepcopy(actB)], activities)
        Sync._accumulateActivities(recA, [copy.deepcopy(actA)], activities)

        self.assertEqual(len(activities), 1)

        # Ensure that it is an exact match
        actB.StartTime = actA.StartTime.replace(tzinfo=pytz.timezone("America/Denver")) + timedelta(hours=5, seconds=1)
        activities = []
        Sync._accumulateActivities(recB, [copy.deepcopy(actB)], activities)
        Sync._accumulateActivities(recA, [copy.deepcopy(actA)], activities)

        self.assertEqual(len(activities), 2)

        # Ensure that overly large differences >38hr - not possible via TZ differences & shamefully bad import/export code on the part of some services - are not deduplicated
        actB.StartTime = actA.StartTime.replace(tzinfo=pytz.timezone("America/Denver")) + timedelta(hours=50)
        activities = []
        Sync._accumulateActivities(recB, [copy.deepcopy(actB)], activities)
        Sync._accumulateActivities(recA, [copy.deepcopy(actA)], activities)

        self.assertEqual(len(activities), 2)

开发者ID:7e7，项目名称:tapiriik，代码行数:36，代码来源:sync.py

注：本文中的tapiriik.sync.Sync类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。

鲜花

握手

雷人

路过

鸡蛋

该文章已有0人参与评论

请发表评论

全部评论

专题导读

More+

10-27 六六分期app的软件客服如何联系？(六六分期

11-06 可心卡盟:win10系统火狐flash插件崩溃怎么

11-06 亲亲特价:怎么删除回收站图标

11-06 济南大学虚拟社区:鲁大师节能降温的具体办

11-06 xlueops.exe:无线网络安装向导

11-06 女斗合众国:win7系统cf与主机连接不稳定怎

11-06 0xc000022-[cf烟雾头]cf怎么调烟雾头

11-06 qizideyouhuo:应用程序无法正常启动0xc0000

11-06 ipz-185:win7系统vcf文件怎么打开

11-06 傻哥蹦迪:win10系统s4怎么打开usb调试

11-06 八神浩树gtaste:回收站清空了怎么恢复

11-06 妖尾之黑色守护:win10系统电脑没有1440x900

11-06 校园至尊魔王小说:win7系统浏览网页时字体

11-06 女斗合众国:win10系统访问共享文件夹提示请

11-06 tokyo hot n0654:恢复win7系统默认字体一招

11-06 雨酷仙境:设置win7系统转移临时文件夹腾出

11-06 阿穆纳伊之杖:win7系统开始菜单在右边还原

11-06 tunespotting:win10系统火狐flash插件总是

11-06 甘尔葛分析师：计谋网站seo关键词暴涨有什

11-06 蔡贵霖: 计谋网站seo关键词暴涨有什么秘密

11-06 博益网首页:ao3网页版进入不了解决方法

11-06 漏斗子专栏: 网站数据分析小白易懂精华篇

11-06 见证双虹怎么做:win7系统开启telnet命令的

11-06 颾狐蝶蜋:系统资源不足无法完成请求的服务

11-06 国光中学校歌:提交网站到alexa查询详细步骤

11-06 西安有情天:静态网页和动态网页的区别

11-06 红木雅尚斋:外部链接构造对网站的好处

11-06 前官礼遇：防止域名劫持–增强域安全性的10

11-06 密传二转答案: 中文分词算法有哪些

11-06 金泉家园邮编:百度快照劫持的表现及应对方

Python testtools.TestTools类代码示例发布时间：2022-05-27

Python interchange.Waypoint类代码示例发布时间：2022-05-27

Python util.grid_equal函数代码示例

1 Python 入门教程

Python入门教程 Python 是一种解释型、面向对象、动态数据类型的高级程序设计语言。 P

阅读：13941|2022-01-22

2 Python wikiutil.getFrontPage函数代码示例

Python wikiutil.getFrontPage函数代码示例

阅读：10292|2022-05-24

3 Python 简介

Python 简介 Python 是一个高层次的结合了解释性、编译性、互动性和面向对象的脚本

阅读：4177|2022-01-22

4 Python tests.group函数代码示例

Python tests.group函数代码示例

阅读：4064|2022-05-27

5 Python util.check_if_user_has_permission

Python util.check_if_user_has_permission函数代码示例

阅读：3889|2022-05-27

6 Python 操练实例98

Python 练习实例98 Python 100例题目：从键盘输入一个字符串，将小写字母全部转换成大

阅读：3539|2022-01-22

7 Python 环境搭建

Python 环境搭建本章节我们将向大家介绍如何在本地搭建 Python 开发环境。 Py

阅读：3070|2022-01-22

8 Python 基础语法

Python 基础语法 Python 语言与 Perl，C 和 Java 等语言有许多相似之处。但是，也

阅读：2731|2022-01-22

9 Python output.darkgreen函数代码示例

Python output.darkgreen函数代码示例

阅读：2682|2022-05-25

10 Python 中文编码

Python 中文编码前面章节中我们已经学会了如何用 Python 输出 Hello, World!，英文没

阅读：2349|2022-01-22

客服电话

电子邮件

Python sync.Sync类代码示例

示例1: test_eligibility_flowexception_shortcircuit

示例2: test_svc_level_dupe

示例3: test_svc_supported_activity_types

示例4: test_svc_level_dupe_tz_irregular

示例5: test_svc_level_dupe_tz_nonuniform

示例6: test_svc_level_dupe

示例7: test_activity_deduplicate_normaltz

示例8: test_eligibility_flowexception_both

示例9: test_eligibility_flowexception_none

示例10: test_svc_level_dupe_tz_irregular

示例11: test_svc_level_dupe_tz_nonuniform

示例12: test_eligibility_excluded

示例13: test_eligibility_config

示例14: test_svc_level_dupe_time_leeway

示例15: test_accumulate_exclusions

示例16: test_activity_coalesce

示例17: test_activity_coalesce

示例18: test_activity_deduplicate_tzerror

示例19: test_activity_deduplicate_tzerror

请发表评论

全部评论

上一篇：

下一篇：

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.get_stdout函数代码示例

关于我们

产品与服务

解决方案

139-2527-9053