more tests unit tests on dataprep

rowers/dataprep.py

@@ -949,21 +949,6 @@ def update_c2id_sql(id,c2id):
 
     return 1
 
-def fitnessmetric_to_sql(m,table='powertimefitnessmetric',debug=False):
-    engine = create_engine(database_url, echo=False)
-    columns = ', '.join(m.keys())
-    placeholders  = ", ".join(["?"] * len(m))
-
-    query = "INSERT into %s ( %s ) Values (%s)" % (table, columns, placeholders)
-
-    values = tuple(m[key] for key in m.keys())
-    with engine.connect() as conn, conn.begin():
-        result = conn.execute(query,values)
-
-    conn.close()
-    engine.dispose()
-
-    return 1
 
 
 def getcpdata_sql(rower_id,table='cpdata'):

rowers/dataprepnodjango.py

@@ -543,40 +543,6 @@ def update_c2id_sql(id,c2id):
 
     return 1
 
-def fitnessmetric_to_sql(m,table='powertimefitnessmetric',debug=False,
-                         doclean=False):
-    # test if nan among values
-    for key in m.keys():
-        if str(m[key]) == 'nan':
-            m[key] = -1
-        if 'inf' in str(m[key]):
-            m[key] = -1
-
-    if debug:
-        engine = create_engine(database_url_debug, echo=False)
-    else:
-        engine = create_engine(database_url, echo=False)
-
-    columns = ', '.join(m.keys())
-    if use_sqlite:
-        placeholders  = ", ".join(["?"] * len(m))
-    else:
-        placeholders  = ", ".join(["%s"] * len(m))
-
-    query = "INSERT into %s ( %s ) Values (%s)" % (table, columns, placeholders)
-    query2 = "DELETE FROM %s WHERE PowerFourMin < 0 and PowerOneHour < 0 and PowerTwoK < 0 and user_id = %s " % (table,m['user_id'])
-
-    values = tuple(m[key] for key in m.keys())
-    with engine.connect() as conn, conn.begin():
-        if doclean:
-            result2 = conn.execute(query2)
-        result = conn.execute(query,values)
-
-
-    conn.close()
-    engine.dispose()
-
-    return 1
 
 
 

rowers/tasks.py

@@ -72,7 +72,7 @@ from rowers.dataprepnodjango import (
     update_strokedata,
     getsmallrowdata_db, updatecpdata_sql,update_c2id_sql,
     update_workout_field_sql,
-    update_agegroup_db,fitnessmetric_to_sql,
+    update_agegroup_db,
     add_c2_stroke_data_db,totaltime_sec_to_string,
     create_c2_stroke_data_db,update_empower,
     database_url_debug,database_url,dataprep,

rowers/tests/test_unit_tests.py

@@ -11,6 +11,72 @@ nu = datetime.datetime.now()
 
 # interactive plots
 from rowers import interactiveplots
+from rowers import dataprep
+
+class DataPrepTests(TestCase):
+    def setUp(self):
+        self.u = UserFactory()
+
+        self.r = Rower.objects.create(user=self.u,
+                                      birthdate=faker.profile()['birthdate'],
+                                      gdproptin=True,surveydone=True,
+                                      gdproptindate=timezone.now(),
+                                      rowerplan='coach')
+
+        self.c = Client()
+        self.user_workouts = WorkoutFactory.create_batch(5, user=self.r)
+        self.factory = RequestFactory()
+        self.password = faker.word()
+        self.u.set_password(self.password)
+        self.u.save()
+
+        result = get_random_file(filename='rowers/tests/testdata/uherskehradiste_otw.csv')
+
+        self.wuh_otw = WorkoutFactory(user=self.r,
+                                     csvfilename=result['filename'],
+                                     starttime=result['starttime'],
+                                     startdatetime=result['startdatetime'],
+                                     duration=result['duration'],
+                                     distance=result['totaldist'],
+                                     workouttype = 'water',
+        )
+
+
+    def tearDown(self):
+        pass
+
+    @patch('rowers.dataprep.getsmallrowdata_db',side_effect=mocked_getsmallrowdata_uh)
+    def test_get_videodata(self,mocked_getsmallrowdata_uh):
+        data, metrics, maxtime = dataprep.get_video_data(self.wuh_otw)
+
+        self.assertEqual(len(data),9)
+        self.assertEqual(len(metrics),6)
+        self.assertEqual(int(maxtime),1737)
+
+    def test_polarization_index(self):
+        df = pd.read_csv('rowers/tests/testdata/uhfull.csv')
+        index = dataprep.polarization_index(df,self.r)
+        self.assertEqual(int(100*index),-67)
+
+    def test_get_latlon(self):
+        data = dataprep.get_latlon(self.wuh_otw.id)
+        self.assertEqual(len(data),2)
+
+    def test_workout_summary_to_df(self):
+        df = dataprep.workout_summary_to_df(self.r)
+        self.assertEqual(len(df),6)
+
+    @patch('rowers.dataprep.create_engine')
+    def test_update_c2id(self,mocked_sqlalchemy):
+        res = dataprep.update_c2id_sql(1,1)
+        self.assertEqual(res,1)
+
+    def test_checkmarker(self):
+        workouts = Workout.objects.all().order_by("-date")
+        wmax = dataprep.check_marker(workouts[0])
+        self.assertTrue(wmax.rankingpiece)
+
+
 
 class InteractivePlotTests(TestCase):
     def setUp(self):
@@ -29,6 +95,9 @@ class InteractivePlotTests(TestCase):
         self.u.set_password(self.password)
         self.u.save()
 
+    def tearDown(self):
+        pass
+
     def test_interactive_hr_piechart(self):
         df = pd.read_csv('rowers/tests/testdata/getrowdata_mock.csv')