Add optional colorized docker test output

Add a --color flag to the local docker runner and pass it through to
sbin/run_tests.py.

Implement ANSI colorization for streamed unittest output so per-test
statuses and final summaries are easier to scan:

- ok in green
- skipped in yellow
- FAIL/ERROR in red
- "Ran ..." summary in cyan

Color mode supports auto/always/never and respects NO_COLOR,
FORCE_COLOR, and CLICOLOR_FORCE in auto mode.

Keep non-color mode as a fast raw passthrough path.

Author: kaste

docker/README.md

@@ -76,6 +76,18 @@ Text and Package Control versions) plus the generated schedule.
 ut-run-tests . --dry-run
 ```
 
+## Colored output
+
+Use `--color` to control ANSI colors in test output:
+
+- `--color auto` (default): color only when stdout is a TTY
+- `--color always`: force color
+- `--color never`: disable color
+
+```sh
+ut-run-tests . --color always
+```
+
 ## Run a single test file
 
 ```sh

docker/run_tests.py

@@ -83,6 +83,7 @@ def main(argv: list[str] | None = None) -> int:
         failfast=args.failfast,
         reload_package_on_testing=args.reload_package_on_testing,
         dry_run=args.dry_run,
+        color=args.color,
         tests_dir=tests_dir,
         pattern=pattern,
     )
@@ -139,6 +140,12 @@ def parse_args(argv: list[str] | None) -> argparse.Namespace:
         action="store_true",
         help="Only print runtime metadata and schedule.",
     )
+    test_group.add_argument(
+        "--color",
+        choices=("auto", "always", "never"),
+        default="auto",
+        help="Colorize test output (default: auto).",
+    )
 
     docker_group = parser.add_argument_group("docker options")
     docker_group.add_argument(
@@ -340,6 +347,7 @@ def build_docker_run_command(
     failfast: bool,
     reload_package_on_testing: bool,
     dry_run: bool,
+    color: str,
     tests_dir: str | None,
     pattern: str | None,
 ) -> list[str]:
@@ -374,6 +382,8 @@ def build_docker_run_command(
     if dry_run:
         command.append("--dry-run")
 
+    command.extend(["--color", color])
+
     if tests_dir:
         command.extend(["--tests-dir", tests_dir])
 

sbin/run_tests.py

@@ -28,6 +28,15 @@
 SCHEDULE_RUNNER_TARGET = os.path.join(UT_DIR_PATH, "zzz_run_scheduler.py")
 RX_RESULT = re.compile(r'^(?P<result>OK|FAILED|ERROR)', re.MULTILINE)
 RX_DONE = re.compile(r'^UnitTesting: Done\.$', re.MULTILINE)
+RX_TEST_STATUS = re.compile(r'\.\.\. (ok|FAIL|ERROR|skipped)(\b.*)$')
+RX_SUMMARY_OK = re.compile(r'^OK(?:\b.*)?$')
+RX_SUMMARY_FAIL = re.compile(r'^(FAILED|ERROR)(?:\b.*)?$')
+
+ANSI_RESET = "\033[0m"
+ANSI_GREEN = "\033[32m"
+ANSI_RED = "\033[31m"
+ANSI_YELLOW = "\033[33m"
+ANSI_CYAN = "\033[36m"
 
 _is_windows = sys.platform == 'win32'
 
@@ -113,9 +122,11 @@ def kill_sublime_text():
     subprocess.Popen("pkill plugin_host || true", shell=True)
 
 
-def read_output(path):
+def read_output(path, color='auto'):
     # todo: use notification instead of polling
     success = None
+    use_color = should_use_color(color)
+    pending = ""
 
     def check_is_success(result):
         try:
@@ -131,8 +142,13 @@ def check_is_done(result):
             offset = f.tell()
             result = f.read()
 
-            print(result, end="")
-            sys.stdout.flush()
+            if result:
+                if use_color:
+                    rendered, pending = colorize_output_chunk(result, pending)
+                    print(rendered, end="")
+                else:
+                    print(result, end="")
+                sys.stdout.flush()
 
             # Keep checking while we don't have a definite result.
             success = check_is_success(result)
@@ -145,9 +161,93 @@ def check_is_done(result):
 
             time.sleep(0.2)
 
+    if use_color and pending:
+        print(colorize_output_line(pending), end="")
+        sys.stdout.flush()
+
     return success
 
 
+def should_use_color(mode):
+    if mode == 'always':
+        return True
+
+    if mode == 'never':
+        return False
+
+    if os.environ.get('NO_COLOR') is not None:
+        return False
+
+    if os.environ.get('CLICOLOR_FORCE') not in (None, '', '0'):
+        return True
+
+    if os.environ.get('FORCE_COLOR') not in (None, '', '0'):
+        return True
+
+    try:
+        return sys.stdout.isatty()
+    except Exception:
+        return False
+
+
+def colorize_output_chunk(chunk, pending):
+    if not chunk:
+        return '', pending
+
+    text = pending + chunk
+    lines = text.splitlines(True)
+
+    if lines and not lines[-1].endswith(('\n', '\r')):
+        pending = lines.pop()
+    else:
+        pending = ''
+
+    rendered = ''.join(colorize_output_line(line) for line in lines)
+    return rendered, pending
+
+
+def colorize_output_line(line):
+    newline = ''
+    body = line
+
+    if body.endswith('\r\n'):
+        body = body[:-2]
+        newline = '\r\n'
+    elif body.endswith('\n') or body.endswith('\r'):
+        body = body[:-1]
+        newline = line[-1]
+
+    test_status_match = RX_TEST_STATUS.search(body)
+    if test_status_match:
+        status = test_status_match.group(1)
+        suffix = test_status_match.group(2)
+        body = (
+            body[:test_status_match.start()]
+            + '... '
+            + colorize_status(status)
+            + suffix
+        )
+
+    if RX_SUMMARY_OK.match(body):
+        body = ANSI_GREEN + body + ANSI_RESET
+    elif RX_SUMMARY_FAIL.match(body):
+        body = ANSI_RED + body + ANSI_RESET
+    elif body.startswith('Ran '):
+        body = ANSI_CYAN + body + ANSI_RESET
+
+    return body + newline
+
+
+def colorize_status(status):
+    if status == 'ok':
+        return ANSI_GREEN + status + ANSI_RESET
+
+    if status == 'skipped':
+        return ANSI_YELLOW + status + ANSI_RESET
+
+    return ANSI_RED + status + ANSI_RESET
+
+
 def restore_coverage_file(path, package):
     # restore .coverage if it exists, needed for coveralls
     if os.path.exists(path):
@@ -207,7 +307,7 @@ def detect_package_control_version():
     return str(version) if version else None
 
 
-def main(default_schedule_info, dry_run=False):
+def main(default_schedule_info, dry_run=False, color='auto'):
     package_under_test = default_schedule_info['package']
     output_dir = os.path.join(UT_OUTPUT_DIR_PATH, package_under_test)
     output_file = os.path.join(output_dir, "result")
@@ -247,7 +347,7 @@ def main(default_schedule_info, dry_run=False):
             time.sleep(2)
 
     print("Start to read output...")
-    if not read_output(output_file):
+    if not read_output(output_file, color=color):
         sys.exit(1)
     restore_coverage_file(coverage_file, package_under_test)
     delete_file_if_exists(SCHEDULE_RUNNER_TARGET)
@@ -264,6 +364,13 @@ def main(default_schedule_info, dry_run=False):
     parser.add_option('--failfast', action='store_true')
     parser.add_option('--reload-package-on-testing', action='store_true')
     parser.add_option('--dry-run', action='store_true')
+    parser.add_option(
+        '--color',
+        type='choice',
+        choices=['auto', 'always', 'never'],
+        default='auto',
+        help='Colorize test output (auto, always, never).',
+    )
 
     options, remainder = parser.parse_args()
 
@@ -294,4 +401,4 @@ def main(default_schedule_info, dry_run=False):
     if options.reload_package_on_testing:
         default_schedule_info['reload_package_on_testing'] = True
 
-    main(default_schedule_info, dry_run=options.dry_run)
+    main(default_schedule_info, dry_run=options.dry_run, color=options.color)