本文整理汇总了Python中tapiriik.testing.testtools.TestTools类的典型用法代码示例。如果您正苦于以下问题:Python TestTools类的具体用法?Python TestTools怎么用?Python TestTools使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了TestTools类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: 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
示例2: 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
示例3: 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
示例4: 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
示例5: 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
示例6: 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
示例7: test_eligibility_flowexception_reverse
def test_eligibility_flowexception_reverse(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)
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)
开发者ID:7e7,项目名称:tapiriik,代码行数:13,代码来源:sync.py
示例8: 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]
actA.CalculateUID()
actA.UIDs = set([actA.UID])
actA.Record = ActivityRecord.FromActivity(actA)
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]
actB.CalculateUID()
actB.UIDs = set([actB.UID])
actB.Record = ActivityRecord.FromActivity(actB)
s = SynchronizationTask(None)
s._serviceConnections = [recA, recB]
s._activities = []
s._accumulateActivities(recA, [actA])
s._accumulateActivities(recB, [actB])
syncToA = s._determineRecipientServices(actA)
syncToB = s._determineRecipientServices(actB)
self.assertEqual(len(syncToA), 0)
self.assertEqual(len(syncToB), 0)
svcB.SupportedActivities = svcA.SupportedActivities
syncToA = s._determineRecipientServices(actA)
syncToB = s._determineRecipientServices(actB)
self.assertEqual(len(syncToA), 1)
self.assertEqual(len(syncToB), 0)
svcB.SupportedActivities = svcA.SupportedActivities = [ActivityType.CrossCountrySkiing, ActivityType.Cycling]
syncToA = s._determineRecipientServices(actA)
syncToB = s._determineRecipientServices(actB)
self.assertEqual(len(syncToA), 1)
self.assertEqual(len(syncToB), 1)
开发者ID:OkanEsen,项目名称:tapiriik,代码行数:51,代码来源:sync.py
示例9: test_constant_representation
def test_constant_representation(self):
''' ensures that gpx import/export is symetric '''
svcA, other = TestTools.create_mock_services()
svcA.SupportsHR = svcA.SupportsCadence = svcA.SupportsTemp = True
svcA.SupportsPower = svcA.SupportsCalories = False
act = TestTools.create_random_activity(svcA, tz=True)
mid = GPXIO.Dump(act)
act2 = GPXIO.Parse(bytes(mid,"UTF-8"))
act2.TZ = act.TZ # we need to fake this since local TZ isn't defined in GPX files, and TZ discovery will flail with random activities
act2.AdjustTZ()
act.Stats.Distance = act2.Stats.Distance = None # same here
self.assertActivitiesEqual(act2, act)
开发者ID:7e7,项目名称:tapiriik,代码行数:16,代码来源:gpx.py
示例10: 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]
s = SynchronizationTask(None)
s._excludedServices = {recA._id: UserException(UserExceptionType.Private)}
s.user = user
s._serviceConnections = recipientServices
act.UIDs = set([act.UID])
act.Record = ActivityRecord.FromActivity(act)
eligible = s._determineEligibleRecipientServices(act, recipientServices)
self.assertTrue(recB in eligible)
self.assertTrue(recA not in eligible)
开发者ID:OkanEsen,项目名称:tapiriik,代码行数:16,代码来源:sync.py
示例11: 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()
svcA.SupportedActivities = [ActivityType.CrossCountrySkiing]
svcB.SupportedActivities = [ActivityType.Cycling]
actA = Activity()
actA.StartTime = datetime(1, 2, 3, 4, 5, 6, 7)
actA.UploadedTo = [TestTools.create_mock_upload_record(svcA)]
actA.Type = svcA.SupportedActivities[0]
actB = Activity()
actB.StartTime = datetime(5, 6, 7, 8, 9, 10, 11)
actB.UploadedTo = [TestTools.create_mock_upload_record(svcB)]
actB.Type = [x for x in svcB.SupportedActivities if x != actA.Type][0]
actA.CalculateUID()
actB.CalculateUID()
allConns = [actA.UploadedTo[0]["Connection"], actB.UploadedTo[0]["Connection"]]
activities = []
Sync._accumulateActivities(svcA, [actA], activities)
Sync._accumulateActivities(svcB, [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:euphoria,项目名称:tapiriik,代码行数:45,代码来源:sync.py
示例12: 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()
s = SynchronizationTask(None)
s._activities = []
s._accumulateActivities(recB, [copy.deepcopy(actB)])
s._accumulateActivities(recA, [copy.deepcopy(actA)])
self.assertEqual(len(s._activities), 1)
# Ensure that it is deduplicated on non-exact match
actB.StartTime = actA.StartTime.replace(tzinfo=pytz.timezone("America/Denver")) + timedelta(hours=5, seconds=1)
s._activities = []
s._accumulateActivities(recB, [copy.deepcopy(actB)])
s._accumulateActivities(recA, [copy.deepcopy(actA)])
self.assertEqual(len(s._activities), 1)
# Ensure that it is *not* deduplicated when it really doesn't match
actB.StartTime = actA.StartTime.replace(tzinfo=pytz.timezone("America/Denver")) + timedelta(hours=5, minutes=7)
s._activities = []
s._accumulateActivities(recB, [copy.deepcopy(actB)])
s._accumulateActivities(recA, [copy.deepcopy(actA)])
self.assertEqual(len(s._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)
s._activities = []
s._accumulateActivities(recB, [copy.deepcopy(actB)])
s._accumulateActivities(recA, [copy.deepcopy(actA)])
self.assertEqual(len(s._activities), 2)
开发者ID:OkanEsen,项目名称:tapiriik,代码行数:45,代码来源:sync.py
示例13: 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
示例14: test_eligibility_flowexception_reverse
def test_eligibility_flowexception_reverse(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
act.UIDs = set([act.UID])
act.Record = ActivityRecord.FromActivity(act)
User.SetFlowException(user, recA, recB, flowToSource=False)
recipientServices = [recA, recB]
s = SynchronizationTask(None)
s._excludedServices = {}
s.user = user
s._serviceConnections = recipientServices
eligible = s._determineEligibleRecipientServices(act, recipientServices)
self.assertFalse(recA in eligible)
self.assertTrue(recB in eligible)
开发者ID:OkanEsen,项目名称:tapiriik,代码行数:18,代码来源:sync.py
示例15: 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
示例16: 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
示例17: 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
示例18: 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]
s = SynchronizationTask(None)
s._excludedServices = {}
s.user = user
s._serviceConnections = recipientServices
act.UIDs = set([act.UID])
act.Record = ActivityRecord.FromActivity(act)
eligible = s._determineEligibleRecipientServices(act, recipientServices)
self.assertTrue(recB in eligible)
self.assertTrue(recA not in eligible)
开发者ID:OkanEsen,项目名称:tapiriik,代码行数:18,代码来源:sync.py
示例19: 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
示例20: test_accumulate_exclusions
def test_accumulate_exclusions(self):
svcA, svcB = TestTools.create_mock_services()
recA = TestTools.create_mock_svc_record(svcA)
# regular
s = SynchronizationTask(None)
s._syncExclusions = {recA._id: {}}
exc = APIExcludeActivity("Messag!e", activity_id=3.14)
s._accumulateExclusions(recA, exc)
exclusionstore = s._syncExclusions
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", activity_id=42, permanent=False, activity=act)
s = SynchronizationTask(None)
s._syncExclusions = {recA._id: {}}
s._accumulateExclusions(recA, exc)
exclusionstore = s._syncExclusions
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", activity_id=13)
exc3 = APIExcludeActivity("FNIM", activity_id=37)
s._accumulateExclusions(recA, [exc2, exc3])
exclusionstore = s._syncExclusions
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")
s = SynchronizationTask(None)
s._syncExclusions = {}
self.assertRaises(ValueError, s._accumulateExclusions, recA, [exc4])
开发者ID:OkanEsen,项目名称:tapiriik,代码行数:44,代码来源:sync.py
注:本文中的tapiriik.testing.testtools.TestTools类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论