fix: guard slice() against non-datetime timestamps and broken symlinks

ovos_utils/log.py

@@ -396,7 +396,7 @@ def get_log_path(service: str, directories: Optional[List[str]] = None) \
     if directories:
         for directory in directories:
             file = os.path.join(directory, f"{service}.log")
-            if os.path.exists(file):
+            if os.path.lexists(file):
                 return directory
         return None
 

ovos_utils/log_parser.py

@@ -150,7 +150,7 @@ def parse(cls, log_line, last_timestamp=None) -> LogLine:
             data['timestamp'] = datetime.strptime(data['timestamp'], TIME_FORMAT)
             return LogLine(**data)
 
-        data["timestamp"] = last_timestamp or ""
+        data["timestamp"] = last_timestamp
         data["message"] = log_line
         return LogLine(**data)
 
@@ -374,7 +374,7 @@ def slice(start, until, logs, paths, file):
             continue
         _templog[service] = []
         for log in OVOSLogParser.parse_file(logfile):
-            if start <= log.timestamp < end:
+            if log.timestamp is not None and start <= log.timestamp < end:
                 if isinstance(log, Traceback):
                     _templog[service].extend(log.to_loglines())
                 else:

test/unittests/test_log.py

@@ -394,6 +394,17 @@ def test_get_log_path(self, get_config):
         self.assertEqual(get_log_path("test"), self.test_dir)
         get_config.assert_called_once_with(service_name="test")
 
+    def test_get_log_path_broken_symlink(self):
+        from ovos_utils.log import get_log_path
+
+        symlink_path = join(self.test_dir, "broken.log")
+        os.symlink(join(self.test_dir, "does_not_exist.log"), symlink_path)
+        try:
+            self.assertEqual(get_log_path("broken", [self.test_dir]),
+                              self.test_dir)
+        finally:
+            os.unlink(symlink_path)
+
     @patch('ovos_config.Configuration')
     def test_get_log_paths(self, config):
         from ovos_utils.log import get_log_paths

test/unittests/test_log_parser.py

@@ -214,6 +214,13 @@ def test_parse_invalid_line_with_last_timestamp(self) -> None:
         result = OVOSLogParser.parse("some system message\n", last_timestamp=ts)
         self.assertEqual(result.timestamp, ts)
 
+    def test_parse_invalid_line_without_last_timestamp(self) -> None:
+        """parse should leave timestamp as None, not a string, when there is
+        no last_timestamp to fall back to."""
+        from ovos_utils.log_parser import OVOSLogParser
+        result = OVOSLogParser.parse("some system message\n")
+        self.assertIsNone(result.timestamp)
+
     def test_parse_file_valid(self) -> None:
         """parse_file should yield LogLine objects from a valid log file."""
         from ovos_utils.log_parser import OVOSLogParser, LogLine
@@ -283,6 +290,36 @@ def test_parse_file_skips_blank_lines(self) -> None:
         finally:
             os.unlink(fname)
 
+    def test_parse_file_leading_line_without_timestamp(self) -> None:
+        """A log line that does not match LOG_PATTERN (e.g. it is missing a
+        field) before any timestamped line is seen must not leave a string
+        in LogLine.timestamp, since callers compare it against datetimes."""
+        from ovos_utils.log_parser import OVOSLogParser
+
+        content = (
+            "2024-07-17 21:59:57.530 - common_query.openvoiceos - INFO - "
+            "First run of common_query.openvoiceos\n"
+            "2024-07-17 22:00:01.123 - skills - core.MSM - DEBUG - loaded skill\n"
+        )
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".log", delete=False) as f:
+            f.write(content)
+            fname = f.name
+
+        try:
+            results = list(OVOSLogParser.parse_file(fname))
+            first, second = results
+            self.assertIsNone(first.timestamp)
+            self.assertIsInstance(second.timestamp, datetime)
+            start = datetime(2024, 1, 1)
+            end = datetime.now()
+            # must not raise: comparing None against datetimes previously
+            # raised TypeError because the fallback timestamp was ""
+            filtered = [log for log in results
+                        if log.timestamp is not None and start <= log.timestamp < end]
+            self.assertEqual(len(filtered), 1)
+        finally:
+            os.unlink(fname)
+
 
 class TestParseTime(unittest.TestCase):
     """Tests for the parse_time helper."""