for pytest async test setting

Author: kmodexc

pytest.ini

@@ -1,2 +1,3 @@
 [pytest]
 norecursedirs = env venv
+asyncio_mode=auto

pytrackunit/sqlcache.py

@@ -68,6 +68,7 @@
     unit text, 
     start int, 
     end int,
+    CONSTRAINT time CHECK (start < end),
     primary key(unit, start)
 )
 '''
@@ -85,6 +86,39 @@
 )
 '''
 
+def error_item_to_sql_item(_x,meta):
+    """
+    returns the error as a tuple and converts the time to unix timestamp (milliseconds)
+    """
+    _id = meta['id']
+    #print(x['time'])
+    _time = int(get_datetime(_x['time']).timestamp()*1000)
+    #print(_time,x['time'],get_datetime(x['time']))
+    _spn = _x['spn']
+    _fmi = _x['fmi']
+    _oc = _x['occurrenceCount']
+    if 'name' in _x:
+        _name = _x['name']
+    else:
+        _name = 'none'
+    if 'description' in _x:
+        _desc = _x['description']
+    else:
+        _desc = 'none'
+    return (_id,_time,_spn,_fmi,_oc,_name,_desc)
+def sql_item_to_error_item(obj):
+    """
+    the operation error_item_to_sql_item reversed
+    """
+    _x = {}
+    _x['id'] = obj[0]
+    _x['time'] = datetime.fromtimestamp(obj[1]/1000.0).isoformat()
+    _x['spn'] = obj[2]
+    _x['fmi'] = obj[3]
+    _x['occurrenceCount'] = obj[4]
+    _x['name'] = obj[5]
+    _x['description'] = obj[6]
+    return _x
 class SqlInsertIter:
     """iterator for tucache data"""
     def __init__(self, sqliter, meta=None, _db=None):
@@ -103,29 +137,12 @@ def __aiter__(self):
         return self
 
     async def __anext__(self):
-        def map_f(_x,meta):
-            _id = meta['id']
-            #print(x['time'])
-            _time = int(get_datetime(_x['time']).timestamp()*1000)
-            #print(_time,x['time'],get_datetime(x['time']))
-            _spn = _x['spn']
-            _fmi = _x['fmi']
-            _oc = _x['occurrenceCount']
-            if 'name' in _x:
-                _name = _x['name']
-            else:
-                _name = 'none'
-            if 'description' in _x:
-                _desc = _x['description']
-            else:
-                _desc = 'none'
-            return (_id,_time,_spn,_fmi,_oc,_name,_desc)
-
         try:
             data, meta = await self.sqliter.__anext__()
             if self._db is not None:
                 #print(data)
-                sqldata = map(lambda x: map_f(x,meta),data)
+                data = list(data)
+                sqldata = map(lambda x: error_item_to_sql_item(x,meta),data)
                 self.cur.executemany("INSERT INTO error VALUES (?,?,?,?,?,?,?)",sqldata)
             return data, meta
         except StopAsyncIteration as exc:
@@ -136,7 +153,17 @@ def map_f(_x,meta):
             print("Committed")
             self._db.commit()
             raise StopAsyncIteration from exc
-
+        except sqlite3.IntegrityError as exc1:
+            print("Integrety error with",meta)
+            print("Try to find double entry (insert side)")
+            sqldata = map(lambda x: error_item_to_sql_item(x,meta),data)
+            for in_item in sqldata:
+                try:
+                    self.cur.execute("INSERT INTO error VALUES (?,?,?,?,?,?,?)",in_item)
+                except sqlite3.IntegrityError as exc2:
+                    print("Item throwing Integrity error",in_item)
+            self._db.rollback()
+            raise sqlite3.IntegrityError from exc1
 class SqlCache:
     """Sql cache can cache trackunit data in Sqlite DB"""
     def __init__(self,auth=None,_dir=None,db_file="webdb.db",upstream_cache=None):
@@ -170,16 +197,16 @@ def get_history(self, unit, start):
             print(meta)
         except StopIteration:
             print("no element found")
-
     def get_candata(self,unit,start):
         """returns history data of a vehicle with start and end date"""
         #tbd
-    def get_errors_upstream(self, veh_id, start_ts, end_ts, previter=None):
+    def get_faults_upstream(self, veh_id, start_ts, end_ts, previter=None):
         """gets errors from upstream cache"""
         cur = self._db.cursor()
 
         cur.execute("INSERT INTO errormeta VALUES (?,?,?)",(veh_id,start_ts,end_ts))
-        self._db.commit()
+        # wait until data is in database and commit then
+        #self._db.commit()
 
         start = datetime.fromtimestamp(start_ts/1000.0)
         end = datetime.fromtimestamp(end_ts/1000.0)
@@ -200,48 +227,48 @@ def get_errors_upstream(self, veh_id, start_ts, end_ts, previter=None):
 
         return previter, _len
 
-    def get_errors(self, veh_id, start, end, previter=None):
-        """returns error in between the given datetime objects"""
-        start_ts = int(start.timestamp()*1000)
-        end_ts = int(end.timestamp()*1000)
-        return self.get_errors_unixts(veh_id, start_ts, end_ts, previter)
-
-    def get_errors_unixts(self, veh_id, start_ts, end_ts, previter=None):
+    def get_faults_unixts(self, veh_id, start_ts, end_ts, previter=None):
         """returns error in between the given datetime objects"""
 
         cur = self._db.cursor()
 
         # Query meta database to check whether there the data is in database
         try:
-            _, me_start, me_end = next(iter(cur.execute("""
-                select * from errormeta where unit = ? and 
-                (start <= ? and end >= ?) or 
-                (start <= ? and end >= ?) order by start
-                """,(veh_id,start_ts,start_ts,end_ts,end_ts))))
+            me_vehid, me_start, me_end = next(iter(cur.execute("""
+                select * from errormeta where unit = ? and (
+                (start <= ? and end > ?) or 
+                (start < ? and end >= ?) or
+                (start >= ? and end <= ?)
+                ) order by start
+                """,(veh_id,start_ts,start_ts,end_ts,end_ts,start_ts,end_ts))))
             me_start = int(float(me_start))
             me_end = int(float(me_end))
         except StopIteration:
-            print("Stop iteration. Get errors from upstream now")
-            return self.get_errors_upstream(veh_id,start_ts,end_ts,previter)
+            print("Stop iteration. Didnt find",\
+                veh_id,start_ts,end_ts,\
+                "Get errors from upstream now")
+            return self.get_faults_upstream(veh_id,start_ts,end_ts,previter)
+
+        print("Found block",me_vehid,me_start,me_end)
 
         # Depending on the start and end of the next block, apply divide and conquer
         # by recursive calls of this function.
         if me_start <= start_ts:
             if me_end < end_ts:
-                previter, cnt1 = self.get_errors_unixts(veh_id,start_ts,me_end,previter)
-                previter, cnt2 = self.get_errors_unixts(veh_id,me_end+1,start_ts,previter)
+                previter, cnt1 = self.get_faults_unixts(veh_id,start_ts,me_end,previter)
+                previter, cnt2 = self.get_faults_unixts(veh_id,me_end+1,end_ts,previter)
                 return previter,(cnt1+cnt2)
-            return self.get_errors_sql(veh_id,start_ts,end_ts,previter)
+            return self.get_faults_sql(veh_id,start_ts,end_ts,previter)
         if me_end >= end_ts:
-            previter, cnt1 = self.get_errors_unixts(veh_id,start_ts,me_start-1,previter)
-            previter, cnt2 = self.get_errors_unixts(veh_id,me_start,end_ts,previter)
+            previter, cnt1 = self.get_faults_unixts(veh_id,start_ts,me_start-1,previter)
+            previter, cnt2 = self.get_faults_unixts(veh_id,me_start,end_ts,previter)
             return previter,(cnt1+cnt2)
-        previter, cnt1 = self.get_errors_unixts(veh_id,start_ts,me_start-1,previter)
-        previter, cnt2 = self.get_errors_unixts(veh_id,me_start,me_end,previter)
-        previter, cnt3 = self.get_errors_unixts(veh_id,me_end+1,end_ts,previter)
+        previter, cnt1 = self.get_faults_unixts(veh_id,start_ts,me_start-1,previter)
+        previter, cnt2 = self.get_faults_unixts(veh_id,me_start,me_end,previter)
+        previter, cnt3 = self.get_faults_unixts(veh_id,me_end+1,end_ts,previter)
         return previter,(cnt1+cnt2+cnt3)
 
-    def get_errors_sql(self, veh_id, start_ts, end_ts, previter=None):
+    def get_faults_sql(self, veh_id, start_ts, end_ts, previter=None):
         """gets data of this period from db whether or not it was actually stored there"""
         cur = self._db.cursor()
 
@@ -252,6 +279,8 @@ def get_errors_sql(self, veh_id, start_ts, end_ts, previter=None):
             order by time
             """,(veh_id,start_ts,end_ts)))[0]
 
+        print("found",cnt,"in sql")
+
         meta = {}
         meta["id"] = veh_id
         meta["start"] = start_ts
@@ -261,9 +290,11 @@ def get_errors_sql(self, veh_id, start_ts, end_ts, previter=None):
             previter = TuIter()
 
         if cnt > 0:
-            previter.add(SqlIter(iter(cur.execute(\
+            previter.add(SqlIter(iter(map(lambda x: [x],map(sql_item_to_error_item,cur.execute(\
                 "select * from error where unit = ? and time >= ? and time <= ? order by time",\
-                    (veh_id,start_ts,end_ts))),meta))
+                    (veh_id,start_ts,end_ts))))),meta))
+        else:
+            print("could not find any item in block ", start_ts, end_ts,"for unit",veh_id)
 
         return previter, cnt
     def get_faults(self,veh_id,tdelta=None,previter=None):
@@ -273,14 +304,16 @@ def get_faults(self,veh_id,tdelta=None,previter=None):
         else:
             end = self.tdelta_end
         end = end.replace(hour=0,minute=0,second=0,microsecond=0)
-        if isinstance(tdelta,datetime):
-            start = end+tdelta
+        if isinstance(tdelta,timedelta):
+            start = end-tdelta
         else:
             irange = int(tdelta)
             if irange <= 0:
                 return []
             start = end-timedelta(days=irange)
-        return self.get_errors(veh_id,start,end,previter)
+        return self.get_faults_timedelta(veh_id,start,end,previter)
     def get_faults_timedelta(self,veh_id,start,end,previter=None):
-        """get_faults_timedelta"""
-        return self.get_errors(veh_id,start,end,previter)
+        """returns error in between the given datetime objects"""
+        start_ts = int(start.timestamp()*1000)
+        end_ts = int(end.timestamp()*1000)
+        return self.get_faults_unixts(veh_id, start_ts, end_ts, previter)

pytrackunit/trackunit.py

@@ -60,15 +60,19 @@ def verbose(self, value):
         """sets the verbose mode. in verbose mode, diagnostic output is printed to console."""
         self.cache.cache.verbose = value
 
-    def get_unitlist(self,_type=None,sort_by_hours=True):
+    async def _a_get_unitlist(self,_type=None,sort_by_hours=True):
         """unitList method"""
-        data = asyncio.run(self.cache.get_url('Unit'))
+        data = await self.cache.get_url('Unit')
         if _type is not None:
             data = list(filter(lambda x: " " in x['name'] and _type in x['name'],data))
         if sort_by_hours:
             data.sort(key=lambda x: (x['run1'] if 'run1' in x else 0),reverse=True)
         return data
 
+    def get_unitlist(self,_type=None,sort_by_hours=True):
+        """unitList method"""
+        return asyncio.run(self._a_get_unitlist(_type,sort_by_hours))
+
     async def _a_get_history(self,veh_id,tdelta):
         """async getHistory method"""
         data = []

pytrackunit/tucache.py

@@ -34,8 +34,8 @@ def general_daydiff_get(self,furl,meta,previter=None):
         else:
             end = self.tdelta_end
         end = end.replace(hour=0,minute=0,second=0,microsecond=0)
-        if isinstance(tdelta,datetime):
-            start = end+tdelta
+        if isinstance(tdelta,timedelta):
+            start = end-tdelta
         else:
             irange = int(tdelta)
             if irange <= 0:

tests/test_helper.py

@@ -1,11 +1,13 @@
 from pytrackunit.helper import *
-def test_ma():
-	testdata = []
-	testobject = {}
-	testobject["val"] = 2
-	for i in range(10):
-		testdata.append(testobject)
-	moving_avg(testdata,"val")
+def test_get_datetime():
+	dt = get_datetime("2020-03-21T12:32:13.100000Z")
+	assert dt.date().year == 2020
+	assert dt.date().month == 3
+	assert dt.date().day == 21
+	assert dt.time().hour == 12
+	assert dt.time().minute == 32
+	assert dt.time().second == 13
+	assert dt.time().microsecond == 100000 
 def test_getsection_always_in_section():
 	dobj = {}
 	dobj['time'] = '2021-06-18T13:18:18.0000000Z'
@@ -22,3 +24,24 @@ def test_getsection_min_section_len():
 		indata[i]['value'] = i
 	data = get_next_section(indata,lambda x: x['value'] % 2 == 0,fendsec=None,min_insec_len=2)
 	assert data is None
+def test_ma():
+	testdata = []
+	testobject = {}
+	testobject["val"] = 2
+	for i in range(10):
+		testdata.append(testobject)
+	moving_avg(testdata,"val")
+def test_get_timediff_1():
+	arg1 = "2020-03-21T12:32:13.100000Z"
+	arg2 = "2020-03-21T12:32:13.200000Z"
+	dt = get_time_diff(arg1,arg2)
+	assert dt.days == 0
+	assert dt.seconds == 0
+	#assert dt.microseconds == 100000
+def test_get_timediff_2():
+	arg1 = "2020-03-21T12:32:14.100000Z"
+	arg2 = "2020-03-21T12:32:13.100000Z"
+	dt = get_time_diff(arg1,arg2)
+	assert dt.days == 0
+	assert dt.seconds == 1
+	#assert dt.microseconds == 100000