Merge pull request #303 from wwade/fixJobAgo

utils: clean up pyright errors in utils.py

jobrunner/db/__init__.py

@@ -27,6 +27,7 @@
     dateTimeFromJson,
     dateTimeToJson,
     doMsg,
+    humanTimeDeltaSecs,
     pidDebug,
     safeSleep,
     sprint,
@@ -771,12 +772,7 @@ def _byAge(refA, refB):
             for res in ["pass", "fail"]:
                 if wkspace in perWs and res in perWs[wkspace]:
                     j = perWs[wkspace][res]
-                    sec = int((unow - j.stopTime).total_seconds())
-                    tmHour = sec / (60 * 60)
-                    sec -= tmHour * 60 * 60
-                    tmMin = sec / 60
-                    sec -= tmMin * 60
-                    diffTime = "%d:%02d:%02d" % (tmHour, tmMin, sec)
+                    diffTime = humanTimeDeltaSecs(unow, j.stopTime)
                     sprint(
                         "  last %s, \033[97m%s\033[0m ago" %
                         (res, diffTime))

jobrunner/test/utils_test.py

@@ -4,9 +4,12 @@
 
 from __future__ import absolute_import, division, print_function
 
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+
 import pytest
 
-from jobrunner.utils import autoDecode
+from jobrunner.utils import autoDecode, humanTimeDeltaSecs
 
 
 @pytest.mark.parametrize(("value", "encoding"), [
@@ -16,3 +19,28 @@
 ])
 def testAutoDecode(value, encoding):
     assert value.decode(encoding) == autoDecode(value)
+
+
+@dataclass(frozen=True)
+class HTDCase:
+    delta: timedelta
+    expected: str
+
+
+@pytest.mark.parametrize("tc", [
+    HTDCase(timedelta(), "0:00:00"),
+    HTDCase(timedelta(seconds=1), "0:00:01"),
+    HTDCase(timedelta(seconds=59), "0:00:59"),
+    HTDCase(timedelta(minutes=1), "0:01:00"),
+    HTDCase(timedelta(minutes=59), "0:59:00"),
+    HTDCase(timedelta(hours=1), "1:00:00"),
+    HTDCase(timedelta(hours=23), "23:00:00"),
+    HTDCase(timedelta(days=6), "6 days, 0:00:00"),
+    HTDCase(timedelta(days=4, hours=3, minutes=2, seconds=1), "4 days, 3:02:01"),
+    HTDCase(timedelta(milliseconds=900), "0:00:01"),
+])
+def testHumanTimeDeltaSecs(tc: HTDCase) -> None:
+    b = datetime.now()
+    a = b + tc.delta
+    actual = humanTimeDeltaSecs(a, b)
+    assert tc.expected == actual

jobrunner/utils.py

@@ -38,7 +38,7 @@ def strForEach(value):
     try:
         return text_type(value)
     except (UnicodeDecodeError, UnicodeEncodeError):
-        LOG.debug("%r", value, exc_info=1)
+        LOG.debug("%r", value, exc_info=True)
         return f"{value!r}"
 
 
@@ -47,12 +47,12 @@ def sprint(*args, **kwargs):
     try:
         print(*list(map(strForEach, args)), **kwargs)
     except IOError:
-        LOG.debug("sprint ignore IOError", exc_info=1)
+        LOG.debug("sprint ignore IOError", exc_info=True)
     except (UnicodeEncodeError, UnicodeDecodeError):
         print("codec error", repr(args))
-        LOG.debug("%r", args, exc_info=1)
+        LOG.debug("%r", args, exc_info=True)
     except BaseException:
-        LOG.debug("sprint caught error", exc_info=1)
+        LOG.debug("sprint caught error", exc_info=True)
         raise
 
 
@@ -90,7 +90,7 @@ def lockedSection(jobs):
     try:
         yield
     except BaseException:
-        LOG.debug("lockedSection exception", exc_info=1)
+        LOG.debug("lockedSection exception", exc_info=True)
         raise
     finally:
         jobs.unlock()
@@ -353,11 +353,23 @@ def sudoKillProcGroup(pgrp):
     return None
 
 
-def autoDecode(byteArray):
+def autoDecode(byteArray: bytes) -> str:
     detected = chardet.detect(byteArray)
+    if not detected:
+        return byteArray.decode()
+
     encoding = detected["encoding"]
     if detected["confidence"] < 0.8:  # very arbitrary
         LOG.debug("char encoding below confidence level 0.8 (%r). "
                   "Fall back to UTF-8.", detected)
         encoding = "utf-8"
+
     return byteArray.decode(encoding)
+
+
+def humanTimeDeltaSecs(a: datetime.datetime, b: datetime.datetime) -> str:
+    """
+    Returns a human readable string for the time difference a - b.
+    """
+    seconds = round((a - b).total_seconds(), 0)
+    return f"{datetime.timedelta(seconds=seconds)}"