Log progress when reading files

python/publish/junit.py

@@ -1,6 +1,6 @@
 import os
 from collections import defaultdict
-from typing import Optional, Iterable, Union, Any, List, Dict
+from typing import Optional, Iterable, Union, Any, List, Dict, Callable, TypeVar
 
 import junitparser
 from junitparser import Element, JUnitXml, TestCase, TestSuite, Skipped
@@ -98,7 +98,13 @@ def end(self, tag: Union[str, bytes]) -> Element:
                 self._stack.pop()
 
 
-def parse_junit_xml_files(files: Iterable[str], time_factor: float = 1.0, drop_testcases: bool = False) -> ParsedUnitTestResults:
+T = TypeVar("T")
+
+
+def parse_junit_xml_files(files: Iterable[str],
+                          time_factor: float = 1.0,
+                          drop_testcases: bool = False,
+                          progress: Callable[[T], T] = lambda x: x) -> ParsedUnitTestResults:
     """Parses junit xml files and returns aggregated statistics as a ParsedUnitTestResults."""
     def parse(path: str) -> Union[str, Any]:
         if not os.path.exists(path):
@@ -114,7 +120,7 @@ def parse(path: str) -> Union[str, Any]:
         except BaseException as e:
             return e
 
-    parsed_files = [(result_file, parse(result_file))
+    parsed_files = [progress((result_file, parse(result_file)))
                     for result_file in files]
     junits = [(result_file, junit)
               for result_file, junit in parsed_files

python/publish/progress.py

@@ -0,0 +1,81 @@
+from datetime import datetime
+from logging import Logger
+from threading import Timer
+from typing import Generic, TypeVar, Optional, Callable
+import contextlib
+
+import humanize
+
+T = TypeVar('T')
+
+
+@contextlib.contextmanager
+def progress_logger(items: int,
+                    interval_seconds: int,
+                    progress_template: str,
+                    finish_template: Optional[str],
+                    logger: Logger) -> Callable[[T], T]:
+    progress = Progress(items)
+    plogger = ProgressLogger(progress, interval_seconds, progress_template, logger).start()
+    try:
+        yield progress.observe
+    finally:
+        plogger.finish(finish_template)
+
+
+class Progress(Generic[T]):
+    def __init__(self, items: int):
+        self.items = items
+        self.observations = 0
+
+    def observe(self, observation: T) -> T:
+        self.observations = self.observations + 1
+        return observation
+
+    def get_progress(self) -> str:
+        return f'{self.observations} of {self.items}'
+
+
+class ProgressLogger:
+    def __init__(self, progress: Progress, interval_seconds: int, template: str, logger: Logger):
+        self._progress = progress
+        self._interval_seconds = interval_seconds
+        self._template = template
+        self._logger = logger
+
+        self._start = None
+        self._duration = None
+        self._timer = self._get_progress_timer()
+
+    def start(self) -> 'ProgressLogger':
+        self._start = datetime.utcnow()
+        self._timer.start()
+        return self
+
+    def finish(self, template: Optional[str] = None):
+        self._duration = datetime.utcnow() - self._start
+        self._start = None
+        self._timer.cancel()
+
+        if template:
+            self._logger.info(template.format(items=self._progress.items,
+                                              observations=self._progress.observations,
+                                              duration=self.duration))
+
+    @property
+    def duration(self) -> str:
+        return humanize.precisedelta(self._duration)
+
+    def _get_progress_timer(self):
+        timer = Timer(self._interval_seconds, self._log_progress)
+        timer.setDaemon(daemonic=True)
+        return timer
+
+    def _log_progress(self):
+        if self._start is None:
+            return
+
+        delta = datetime.utcnow() - self._start
+        self._logger.info(self._template.format(progress=self._progress.get_progress(), time=humanize.precisedelta(delta)))
+        self._timer = self._get_progress_timer()
+        self._timer.start()

python/publish_unit_test_results.py

@@ -19,6 +19,7 @@
     comment_mode_off, comment_mode_update, comment_modes, punctuation_space
 from publish.github_action import GithubAction
 from publish.junit import parse_junit_xml_files
+from publish.progress import progress_logger
 from publish.publisher import Publisher, Settings
 from publish.retry import GitHubRetry
 from publish.unittestresults import get_test_results, get_stats, ParsedUnitTestResults
@@ -101,10 +102,16 @@ def main(settings: Settings, gha: GithubAction) -> None:
     avail_mem = humanize.naturalsize(psutil.virtual_memory().available, binary=True)
     logger.info(f'Available memory to read files: {avail_mem}')
 
-    # get the unit test results
-    parsed = parse_junit_xml_files(files, settings.time_factor, settings.ignore_runs).with_commit(settings.commit)
-    [gha.error(message=f'Error processing result file: {error.message}', file=error.file, line=error.line, column=error.column)
-     for error in parsed.errors]
+    # log the progress
+    with progress_logger(items=len(files),
+                         interval_seconds=10,
+                         progress_template='Read {progress} files in {time}',
+                         finish_template='Finished reading {observations} files in {duration}',
+                         logger=logger) as progress:
+        # get the unit test results
+        parsed = parse_junit_xml_files(files, settings.time_factor, settings.ignore_runs, progress).with_commit(settings.commit)
+        [gha.error(message=f'Error processing result file: {error.message}', file=error.file, line=error.line, column=error.column)
+         for error in parsed.errors]
 
     # process the parsed results
     results = get_test_results(parsed, settings.dedup_classes_by_file_name)

python/test/test_progress.py

@@ -0,0 +1,65 @@
+import unittest
+from datetime import datetime
+
+import mock
+
+from publish.progress import Progress, ProgressLogger
+
+
+class TestProgress(unittest.TestCase):
+    def test_get_progress(self):
+        progress = Progress(10)
+        self.assertEqual('0 of 10', progress.get_progress())
+        self.assertEqual('0 of 10', progress.get_progress())
+        self.assertEqual('item', progress.observe('item'))
+        self.assertEqual('1 of 10', progress.get_progress())
+        self.assertEqual('1 of 10', progress.get_progress())
+        self.assertEqual(1, progress.observe(1))
+        self.assertEqual('2 of 10', progress.get_progress())
+        self.assertEqual('2 of 10', progress.get_progress())
+        self.assertEqual(1.2, progress.observe(1.2))
+        self.assertEqual('3 of 10', progress.get_progress())
+        self.assertEqual('3 of 10', progress.get_progress())
+        obj = object()
+        self.assertEqual(obj, progress.observe(obj))
+        self.assertEqual('4 of 10', progress.get_progress())
+        self.assertEqual('4 of 10', progress.get_progress())
+
+
+class TestProgressLogger(unittest.TestCase):
+    def test(self):
+        progress = Progress(10)
+        logger = mock.MagicMock(info=mock.Mock())
+        plogger = ProgressLogger(progress, 60, 'progress: {progress} in {time}', logger)
+        try:
+            ts = datetime.fromisoformat('2022-06-01 12:34:56')
+            with mock.patch('publish.progress.datetime', utcnow=mock.Mock(return_value=ts)):
+                plogger.start()
+            logger.info.assert_not_called()
+
+            progress.observe('item')
+            logger.info.assert_not_called()
+
+            ts = datetime.fromisoformat('2022-06-01 12:35:00')
+            with mock.patch('publish.progress.datetime', utcnow=mock.Mock(return_value=ts)):
+                plogger._log_progress()
+            self.assertEqual([mock.call('progress: 1 of 10 in 4 seconds')], logger.info.call_args_list)
+            logger.info.reset_mock()
+
+            progress.observe('item')
+            progress.observe('item')
+            logger.info.assert_not_called()
+
+            ts = datetime.fromisoformat('2022-06-01 12:40:00')
+            with mock.patch('publish.progress.datetime', utcnow=mock.Mock(return_value=ts)):
+                plogger._log_progress()
+            self.assertEqual([mock.call('progress: 3 of 10 in 5 minutes and 4 seconds')], logger.info.call_args_list)
+            logger.info.reset_mock()
+        finally:
+            ts = datetime.fromisoformat('2022-06-01 12:41:23')
+            with mock.patch('publish.progress.datetime', utcnow=mock.Mock(return_value=ts)):
+                plogger.finish('finished: {observations} of {items} in {duration}')
+            self.assertEqual([mock.call('finished: 3 of 10 in 6 minutes and 27 seconds')], logger.info.call_args_list)
+            logger.info.reset_mock()
+
+        self.assertEqual('6 minutes and 27 seconds', plogger.duration)