allow CLICOLOR_FORCE env var to force CLI color output (#6270)

This adds support for using the `CLICOLOR_FORCE` environment variable to
force color output in Linux even when a tty is not used, mimicking
behavior in [CMake and other
tools](https://bixense.com/clicolors/#bug-reports). The primary intent
is to allow color to be output when cppcheck is run in CI, for example.

lib/color.cpp

@@ -18,11 +18,13 @@
 
 #include "color.h"
 
-#ifndef _WIN32
-#include <unistd.h>
 #include <cstddef>
+#include <cstdlib>
 #include <sstream>
 #include <iostream>
+
+#ifndef _WIN32
+#include <unistd.h>
 #endif
 
 bool gDisableColors = false;
@@ -40,26 +42,39 @@ static bool isStreamATty(const std::ostream & os)
 }
 #endif
 
-std::ostream& operator<<(std::ostream & os, Color c)
+static bool isColorEnabled(const std::ostream & os)
 {
-#ifndef _WIN32
-    if (!gDisableColors && isStreamATty(os))
-        return os << "\033[" << static_cast<std::size_t>(c) << "m";
+    // See https://bixense.com/clicolors/
+    static const bool color_forced_off = (nullptr != std::getenv("NO_COLOR"));
+    if (color_forced_off)
+    {
+        return false;
+    }
+    static const bool color_forced_on = (nullptr != std::getenv("CLICOLOR_FORCE"));
+    if (color_forced_on)
+    {
+        return true;
+    }
+#ifdef _WIN32
+    (void)os;
+    return false;
 #else
-    (void)c;
+    return isStreamATty(os);
 #endif
+}
+
+std::ostream& operator<<(std::ostream & os, Color c)
+{
+    if (!gDisableColors && isColorEnabled(os))
+        return os << "\033[" << static_cast<std::size_t>(c) << "m";
+
     return os;
 }
 
 std::string toString(Color c)
 {
-#ifndef _WIN32
-    std::stringstream ss;
+    std::ostringstream ss;
     ss << c;
     return ss.str();
-#else
-    (void)c;
-    return "";
-#endif
 }
 

releasenotes.txt

@@ -17,4 +17,4 @@ Deprecations:
 -
 
 Other:
--
+- Add support for 'CLICOLOR_FORCE'/'NO_COLOR' environment variables to force/disable ANSI color output for diagnostics.

test/cli/other_test.py

@@ -71,6 +71,43 @@ def test_preprocessor_error(tmpdir):
     assert exitcode != 0
 
 
+ANSI_BOLD = "\x1b[1m"
+ANSI_FG_RED = "\x1b[31m"
+ANSI_FG_DEFAULT = "\x1b[39m"
+ANSI_FG_RESET = "\x1b[0m"
+
+
+@pytest.mark.parametrize("env,color_expected", [({"CLICOLOR_FORCE":"1"}, True), ({"NO_COLOR": "1", "CLICOLOR_FORCE":"1"}, False)])
+def test_color_non_tty(tmpdir, env, color_expected):
+    test_file = os.path.join(tmpdir, 'test.c')
+    with open(test_file, 'wt') as f:
+        f.write('#error test\nx=1;\n')
+    exitcode, _, stderr = cppcheck([test_file], env=env)
+
+    assert exitcode == 0
+    assert stderr
+    assert (ANSI_BOLD in stderr) == color_expected
+    assert (ANSI_FG_RED in stderr)  == color_expected
+    assert (ANSI_FG_DEFAULT in stderr) == color_expected
+    assert (ANSI_FG_RESET in stderr) == color_expected
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="TTY not supported in Windows")
+@pytest.mark.parametrize("env,color_expected", [({}, True), ({"NO_COLOR": "1"}, False)])
+def test_color_tty(tmpdir, env, color_expected):
+    test_file = os.path.join(tmpdir, 'test.c')
+    with open(test_file, 'wt') as f:
+        f.write('#error test\nx=1;\n')
+    exitcode, _, stderr = cppcheck([test_file], env=env, tty=True)
+
+    assert exitcode == 0
+    assert stderr
+    assert (ANSI_BOLD in stderr) == color_expected
+    assert (ANSI_FG_RED in stderr) == color_expected
+    assert (ANSI_FG_DEFAULT in stderr) == color_expected
+    assert (ANSI_FG_RESET in stderr) == color_expected
+
+
 def test_invalid_library(tmpdir):
     args = ['--library=none', '--library=posix', '--library=none2', 'file.c']
 

test/cli/testutils.py

@@ -1,7 +1,10 @@
+import errno
 import logging
 import os
+import select
 import signal
 import subprocess
+import time
 
 # Create Cppcheck project file
 import sys
@@ -72,8 +75,82 @@ def __lookup_cppcheck_exe():
     return exe_path
 
 
+def __run_subprocess_tty(args, env=None, cwd=None, timeout=None):
+    import pty
+    mo, so = pty.openpty()
+    me, se = pty.openpty() 
+    p = subprocess.Popen(args, stdout=so, stderr=se, env=env, cwd=cwd)
+    for fd in [so, se]:
+        os.close(fd)
+
+    select_timeout = 0.04  # seconds
+    readable = [mo, me]
+    result = {mo: b'', me: b''}
+    try:
+        start_time = time.monotonic()
+        while readable:
+            ready, _, _ = select.select(readable, [], [], select_timeout)
+            for fd in ready:
+                try:
+                    data = os.read(fd, 512)
+                except OSError as e:
+                    if e.errno != errno.EIO:
+                        raise
+                    # EIO means EOF on some systems
+                    readable.remove(fd)
+                else:
+                    if not data: # EOF
+                        readable.remove(fd)
+                    result[fd] += data
+            if timeout is not None and (time.monotonic() - start_time):
+                break
+    finally:
+        for fd in [mo, me]:
+            os.close(fd)
+        if p.poll() is None:
+            p.kill()
+        return_code = p.wait()
+
+    stdout = result[mo]
+    stderr = result[me]        
+    return return_code, stdout, stderr
+
+
+def __run_subprocess(args, env=None, cwd=None, timeout=None):
+    p = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env, cwd=cwd)
+
+    try:
+        comm = p.communicate(timeout=timeout)
+        return_code = p.returncode
+        p = None
+    except subprocess.TimeoutExpired:
+        import psutil
+        # terminate all the child processes
+        child_procs = psutil.Process(p.pid).children(recursive=True)
+        if len(child_procs) > 0:
+            for child in child_procs:
+                child.terminate()
+            try:
+                # call with timeout since it might be stuck
+                p.communicate(timeout=5)
+                p = None
+            except subprocess.TimeoutExpired:
+                pass
+        raise
+    finally:
+        if p:
+            # sending the signal to the process groups causes the parent Python process to terminate as well
+            #os.killpg(os.getpgid(p.pid), signal.SIGTERM)  # Send the signal to all the process groups
+            p.terminate()
+            comm = p.communicate()
+
+    stdout = comm[0]
+    stderr = comm[1]    
+    return return_code, stdout, stderr
+
+
 # Run Cppcheck with args
-def cppcheck(args, env=None, remove_checkers_report=True, cwd=None, cppcheck_exe=None, timeout=None):
+def cppcheck(args, env=None, remove_checkers_report=True, cwd=None, cppcheck_exe=None, timeout=None, tty=False):
     exe = cppcheck_exe if cppcheck_exe else __lookup_cppcheck_exe()
     assert exe is not None, 'no cppcheck binary found'
 
@@ -113,33 +190,13 @@ def cppcheck(args, env=None, remove_checkers_report=True, cwd=None, cppcheck_exe
             args.append(arg_executor)
 
     logging.info(exe + ' ' + ' '.join(args))
-    p = subprocess.Popen([exe] + args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env, cwd=cwd)
-    try:
-        comm = p.communicate(timeout=timeout)
-        return_code = p.returncode
-        p = None
-    except subprocess.TimeoutExpired:
-        import psutil
-        # terminate all the child processes
-        child_procs = psutil.Process(p.pid).children(recursive=True)
-        if len(child_procs) > 0:
-            for child in child_procs:
-                child.terminate()
-            try:
-                # call with timeout since it might be stuck
-                p.communicate(timeout=5)
-                p = None
-            except subprocess.TimeoutExpired:
-                pass
-        raise
-    finally:
-        if p:
-            # sending the signal to the process groups causes the parent Python process to terminate as well
-            #os.killpg(os.getpgid(p.pid), signal.SIGTERM)  # Send the signal to all the process groups
-            p.terminate()
-            comm = p.communicate()
-    stdout = comm[0].decode(encoding='utf-8', errors='ignore').replace('\r\n', '\n')
-    stderr = comm[1].decode(encoding='utf-8', errors='ignore').replace('\r\n', '\n')
+
+    run_subprocess = __run_subprocess_tty if tty else __run_subprocess
+    return_code, stdout, stderr = run_subprocess([exe] + args, env=env, cwd=cwd, timeout=timeout)
+
+    stdout = stdout.decode(encoding='utf-8', errors='ignore').replace('\r\n', '\n')
+    stderr = stderr.decode(encoding='utf-8', errors='ignore').replace('\r\n', '\n')
+
     if remove_checkers_report:
         if stderr.find('[checkersReport]\n') > 0:
             start_id = stderr.find('[checkersReport]\n')