Repository: sqlfluff/sqlfluff

Issue:
"--nocolor" option is not honored
### Search before asking

- [X] I searched the [issues](https://github.com/sqlfluff/sqlfluff/issues) and found no similar issues.


### What Happened

` sqlfluff lint --verbose --nocolor --dialect ansi --disable_progress_bar queries.sql --write-output result.txt `
 
content on result.txt file has ansi chars:
```
 ==== sqlfluff ====
[30m[1msqlfluff[0m:                1.0.0 [30m[1mpython[0m:                  3.9.7
....
```
` sqlfluff lint --verbose --nocolor --dialect ansi --disable_progress_bar queries.sql >result2.txt`

content onresult2.txt hasn't any ansi chars:
 ```
==== sqlfluff ====
sqlfluff:                1.0.0 python:                  3.9.7
....
```
but worse....
` sqlfluff lint --verbose --dialect ansi --disable_progress_bar queries.sql`
returns plain output to console
 ```
==== sqlfluff ====
sqlfluff:                1.0.0 python:                  3.9.7
....
```

### Expected Behaviour

1) Both syntax "--write-output result.txt" and "> result.txt" should return the same content

2) the "--nocolor" option  should be always honored



### Observed Behaviour

(see above)

### How to reproduce

```
$ sqlfluff lint --verbose --nocolor --dialect ansi --disable_progress_bar queries.sql --write-output result.txt 
$ cat result.txt 
$ sqlfluff lint --verbose --nocolor --dialect ansi --disable_progress_bar queries.sql >result2.txt
$ cat result2.txt 
```

### Dialect

ansi

### Version

1.0.0

### Configuration

No user config file

### Are you willing to work on and submit a PR to address the issue?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/sqlfluff/sqlfluff/blob/main/CODE_OF_CONDUCT.md)


Discussion:
codecov[bot]: # [Codecov](https://codecov.io/gh/sqlfluff/sqlfluff/pull/3536?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff) Report
> Merging [#3536](https://codecov.io/gh/sqlfluff/sqlfluff/pull/3536?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff) (1306836) into [main](https://codecov.io/gh/sqlfluff/sqlfluff/commit/153c438e5a73bf8701e254cfee33c97175854e33?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff) (153c438) will **decrease** coverage by `0.00%`.
> The diff coverage is `99.44%`.

```diff
@@             Coverage Diff             @@
##              main    #3536      +/-   ##
===========================================
- Coverage   100.00%   99.99%   -0.01%     
===========================================
  Files          172      172              
  Lines        13047    13055       +8     
===========================================
+ Hits         13047    13054       +7     
- Misses           0        1       +1     
```


| [Impacted Files](https://codecov.io/gh/sqlfluff/sqlfluff/pull/3536?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff) | Coverage Δ | |
|---|---|---|
| [src/sqlfluff/cli/commands.py](https://codecov.io/gh/sqlfluff/sqlfluff/pull/3536/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff#diff-c3JjL3NxbGZsdWZmL2NsaS9jb21tYW5kcy5weQ==) | `99.71% <96.96%> (-0.29%)` | :arrow_down: |
| [src/sqlfluff/cli/formatters.py](https://codecov.io/gh/sqlfluff/sqlfluff/pull/3536/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff#diff-c3JjL3NxbGZsdWZmL2NsaS9mb3JtYXR0ZXJzLnB5) | `100.00% <100.00%> (ø)` | |
| [src/sqlfluff/cli/helpers.py](https://codecov.io/gh/sqlfluff/sqlfluff/pull/3536/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff#diff-c3JjL3NxbGZsdWZmL2NsaS9oZWxwZXJzLnB5) | `100.00% <100.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/sqlfluff/sqlfluff/pull/3536?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/sqlfluff/sqlfluff/pull/3536?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff). Last update [153c438...1306836](https://codecov.io/gh/sqlfluff/sqlfluff/pull/3536?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff).

barrywhart: Thanks! There are very few actual additions or changes, just moving things around.

Overnight, I thought of a couple things I want to look at:
* Add a specific test for the user's issue
* Look at some related code re: output format that we may be able to combine or eliminate
barrywhart: @tunetheweb: I added a new test, `test/cli/commands_test.py::test__cli__command_lint_nocolor`, to check specifically for the user's issue. This general test structure (plus possibly adding more cases someday) will help ensure colors don't sneak into the output.
tunetheweb: SGTM!

Patch:
diff --git a/src/sqlfluff/cli/commands.py b/src/sqlfluff/cli/commands.py
index e2b7766c8..fe3f54284 100644
--- a/src/sqlfluff/cli/commands.py
+++ b/src/sqlfluff/cli/commands.py
@@ -7,7 +7,7 @@ import json
 import logging
 import time
 from logging import LogRecord
-from typing import Callable, Tuple, Optional, List, cast
+from typing import Callable, Tuple, Optional, cast
 
 import yaml
 
@@ -23,16 +23,10 @@ from tqdm import tqdm
 from sqlfluff.cli.autocomplete import shell_completion_enabled, dialect_shell_complete
 
 from sqlfluff.cli.formatters import (
-    format_rules,
-    format_violation,
     format_linting_result_header,
-    format_linting_stats,
-    colorize,
-    format_dialect_warning,
-    format_dialects,
     OutputStreamFormatter,
 )
-from sqlfluff.cli.helpers import cli_table, get_package_version
+from sqlfluff.cli.helpers import get_package_version
 from sqlfluff.cli.outputstream import make_output_stream, OutputStream
 
 # Import from sqlfluff core.
@@ -44,23 +38,24 @@ from sqlfluff.core import (
     SQLFluffUserError,
     dialect_selector,
     dialect_readout,
-    TimingSummary,
 )
 from sqlfluff.core.config import progress_bar_configuration
 
 from sqlfluff.core.enums import FormatType, Color
-from sqlfluff.core.linter import ParsedString
 from sqlfluff.core.plugin.host import get_plugin_manager
-from sqlfluff.core.linter import LintingResult
 
 
 class RedWarningsFilter(logging.Filter):
     """This filter makes all warnings or above red."""
 
+    def __init__(self, formatter: OutputStreamFormatter):
+        super().__init__()
+        self.formatter = formatter
+
     def filter(self, record: logging.LogRecord) -> bool:
         """Filter any warnings (or above) to turn them red."""
         if record.levelno >= logging.WARNING:
-            record.msg = f"{colorize(record.msg, Color.red)} "
+            record.msg = f"{self.formatter.colorize(record.msg, Color.red)} "
         return True
 
 
@@ -83,7 +78,10 @@ class StreamHandlerTqdm(logging.StreamHandler):
 
 
 def set_logging_level(
-    verbosity: int, logger: Optional[logging.Logger] = None, stderr_output: bool = False
+    verbosity: int,
+    formatter: OutputStreamFormatter,
+    logger: Optional[logging.Logger] = None,
+    stderr_output: bool = False,
 ) -> None:
     """Set up logging for the CLI.
 
@@ -111,7 +109,7 @@ def set_logging_level(
     # tamed ANSI colour statements, and return us to normality.
     handler.setFormatter(logging.Formatter("\u001b[0m%(levelname)-10s %(message)s"))
     # Set up a handler to colour warnings red.
-    handler.addFilter(RedWarningsFilter())
+    handler.addFilter(RedWarningsFilter(formatter))
     if logger:
         focus_logger = logging.getLogger(f"sqlfluff.{logger}")
         focus_logger.addHandler(handler)
@@ -292,6 +290,7 @@ def get_config(
     **kwargs,
 ) -> FluffConfig:
     """Get a config object from kwargs."""
+    plain_output = OutputStreamFormatter.should_produce_plain_output(kwargs["nocolor"])
     if kwargs.get("dialect"):
         try:
             # We're just making sure it exists at this stage.
@@ -299,7 +298,8 @@ def get_config(
             dialect_selector(kwargs["dialect"])
         except SQLFluffUserError as err:
             click.echo(
-                colorize(
+                OutputStreamFormatter.colorize_helper(
+                    plain_output,
                     f"Error loading dialect '{kwargs['dialect']}': {str(err)}",
                     color=Color.red,
                 )
@@ -307,8 +307,10 @@ def get_config(
             sys.exit(66)
         except KeyError:
             click.echo(
-                colorize(
-                    f"Error: Unknown dialect '{kwargs['dialect']}'", color=Color.red
+                OutputStreamFormatter.colorize_helper(
+                    plain_output,
+                    f"Error: Unknown dialect '{kwargs['dialect']}'",
+                    color=Color.red,
                 )
             )
             sys.exit(66)
@@ -326,7 +328,8 @@ def get_config(
         )
     except SQLFluffUserError as err:  # pragma: no cover
         click.echo(
-            colorize(
+            OutputStreamFormatter.colorize_helper(
+                plain_output,
                 f"Error loading config: {str(err)}",
                 color=Color.red,
             )
@@ -349,6 +352,7 @@ def get_linter_and_formatter(
         sys.exit(66)
     formatter = OutputStreamFormatter(
         output_stream=output_stream or make_output_stream(cfg),
+        nocolor=cfg.get("nocolor"),
         verbosity=cfg.get("verbose"),
         output_line_length=cfg.get("output_line_length"),
     )
@@ -381,8 +385,8 @@ def version(**kwargs) -> None:
 def rules(**kwargs) -> None:
     """Show the current rules in use."""
     c = get_config(**kwargs, dialect="ansi")
-    lnt, _ = get_linter_and_formatter(c)
-    click.echo(format_rules(lnt), color=c.get("color"))
+    lnt, formatter = get_linter_and_formatter(c)
+    click.echo(formatter.format_rules(lnt), color=c.get("color"))
 
 
 @cli.command()
@@ -390,7 +394,8 @@ def rules(**kwargs) -> None:
 def dialects(**kwargs) -> None:
     """Show the current dialects available."""
     c = get_config(**kwargs, require_dialect=False)
-    click.echo(format_dialects(dialect_readout), color=c.get("color"))
+    _, formatter = get_linter_and_formatter(c)
+    click.echo(formatter.format_dialects(dialect_readout), color=c.get("color"))
 
 
 def dump_file_payload(filename: Optional[str], payload: str):
@@ -505,7 +510,12 @@ def lint(
     formatter.dispatch_config(lnt)
 
     # Set up logging.
-    set_logging_level(verbosity=verbose, logger=logger, stderr_output=non_human_output)
+    set_logging_level(
+        verbosity=verbose,
+        formatter=formatter,
+        logger=logger,
+        stderr_output=non_human_output,
+    )
     # add stdin if specified via lone '-'
     if ("-",) == paths:
         result = lnt.lint_string_wrapped(sys.stdin.read(), fname="stdin")
@@ -522,7 +532,7 @@ def lint(
             )
         except OSError:
             click.echo(
-                colorize(
+                formatter.colorize(
                     f"The path(s) '{paths}' could not be accessed. Check it/they "
                     "exist(s).",
                     Color.red,
@@ -531,7 +541,7 @@ def lint(
             sys.exit(1)
         # Output the final stats
         if verbose >= 1:
-            click.echo(format_linting_stats(result, verbose=verbose))
+            click.echo(formatter.format_linting_stats(result, verbose=verbose))
 
     if format == FormatType.json.value:
         file_output = json.dumps(result.as_records())
@@ -589,44 +599,20 @@ def lint(
     output_stream.close()
     if bench:
         click.echo("==== overall timings ====")
-        click.echo(cli_table([("Clock time", result.total_time)]))
+        click.echo(formatter.cli_table([("Clock time", result.total_time)]))
         timing_summary = result.timing_summary()
         for step in timing_summary:
             click.echo(f"=== {step} ===")
-            click.echo(cli_table(timing_summary[step].items()))
+            click.echo(formatter.cli_table(timing_summary[step].items()))
 
     if not nofail:
         if not non_human_output:
-            _completion_message(config)
+            formatter.completion_message()
         sys.exit(result.stats()["exit code"])
     else:
         sys.exit(0)
 
 
-def _handle_files_with_tmp_or_prs_errors(lint_result: LintingResult) -> int:
-    """Discard lint fixes for files with templating or parse errors.
-
-    Returns 1 if there are any files with templating or parse errors after
-    filtering, else 0. (Intended as a process exit code.)
-    """
-    total_errors, num_filtered_errors = lint_result.count_tmp_prs_errors()
-    lint_result.discard_fixes_for_lint_errors_in_files_with_tmp_or_prs_errors()
-    if total_errors:
-        click.echo(
-            colorize(f"  [{total_errors} templating/parsing errors found]", Color.red)
-        )
-        if num_filtered_errors < total_errors:
-            color = Color.red if num_filtered_errors else Color.green
-            click.echo(
-                colorize(
-                    f"  [{num_filtered_errors} templating/parsing errors "
-                    f'remaining after "ignore"]',
-                    color,
-                )
-            )
-    return 1 if num_filtered_errors else 0
-
-
 def do_fixes(lnt, result, formatter=None, **kwargs):
     """Actually do the fixes."""
     click.echo("Persisting Changes...")
@@ -724,7 +710,12 @@ def fix(
     formatter.dispatch_config(lnt)
 
     # Set up logging.
-    set_logging_level(verbosity=verbose, logger=logger, stderr_output=fixing_stdin)
+    set_logging_level(
+        verbosity=verbose,
+        formatter=formatter,
+        logger=logger,
+        stderr_output=fixing_stdin,
+    )
 
     # handle stdin case. should output formatted sql to stdout and nothing else.
     if fixing_stdin:
@@ -734,7 +725,7 @@ def fix(
         templater_error = result.num_violations(types=SQLTemplaterError) > 0
         unfixable_error = result.num_violations(types=SQLLintError, fixable=False) > 0
         if not fix_even_unparsable:
-            exit_code = _handle_files_with_tmp_or_prs_errors(result)
+            exit_code = formatter.handle_files_with_tmp_or_prs_errors(result)
 
         if result.num_violations(types=SQLLintError, fixable=True) > 0:
             stdout = result.paths[0].files[0].fix_string()[0]
@@ -743,14 +734,14 @@ def fix(
 
         if templater_error:
             click.echo(
-                colorize(
+                formatter.colorize(
                     "Fix aborted due to unparseable template variables.",
                     Color.red,
                 ),
                 err=True,
             )
             click.echo(
-                colorize(
+                formatter.colorize(
                     "Use --FIX-EVEN-UNPARSABLE' to attempt to fix the SQL anyway.",
                     Color.red,
                 ),
@@ -758,7 +749,10 @@ def fix(
             )
 
         if unfixable_error:
-            click.echo(colorize("Unfixable violations detected.", Color.red), err=True)
+            click.echo(
+                formatter.colorize("Unfixable violations detected.", Color.red),
+                err=True,
+            )
 
         click.echo(stdout, nl=False)
         sys.exit(1 if templater_error or unfixable_error else exit_code)
@@ -774,7 +768,7 @@ def fix(
         )
     except OSError:
         click.echo(
-            colorize(
+            formatter.colorize(
                 f"The path(s) '{paths}' could not be accessed. Check it/they exist(s).",
                 Color.red,
             ),
@@ -783,7 +777,7 @@ def fix(
         sys.exit(1)
 
     if not fix_even_unparsable:
-        exit_code = _handle_files_with_tmp_or_prs_errors(result)
+        exit_code = formatter.handle_files_with_tmp_or_prs_errors(result)
 
     # NB: We filter to linting violations here, because they're
     # the only ones which can be potentially fixed.
@@ -794,7 +788,9 @@ def fix(
             "linting violations found"
         )
         if force:
-            click.echo(f"{colorize('FORCE MODE', Color.red)}: Attempting fixes...")
+            click.echo(
+                f"{formatter.colorize('FORCE MODE', Color.red)}: Attempting fixes..."
+            )
             success = do_fixes(
                 lnt,
                 result,
@@ -822,7 +818,7 @@ def fix(
                 if not success:
                     sys.exit(1)  # pragma: no cover
                 else:
-                    _completion_message(config)
+                    formatter.completion_message()
             elif c == "n":
                 click.echo("Aborting...")
                 exit_code = 1
@@ -832,7 +828,7 @@ def fix(
                 exit_code = 1
     else:
         click.echo("==== no fixable linting violations found ====")
-        _completion_message(config)
+        formatter.completion_message()
 
     error_types = [
         (
@@ -849,22 +845,15 @@ def fix(
 
     if bench:
         click.echo("==== overall timings ====")
-        click.echo(cli_table([("Clock time", result.total_time)]))
+        click.echo(formatter.cli_table([("Clock time", result.total_time)]))
         timing_summary = result.timing_summary()
         for step in timing_summary:
             click.echo(f"=== {step} ===")
-            click.echo(cli_table(timing_summary[step].items()))
+            click.echo(formatter.cli_table(timing_summary[step].items()))
 
     sys.exit(exit_code)
 
 
-def _completion_message(config: FluffConfig) -> None:
-    click.echo(
-        "All Finished"
-        f"{'' if (config.get('nocolor') or not sys.stdout.isatty()) else ' 📜 🎉'}!"
-    )
-
-
 def quoted_presenter(dumper, data):
     """Re-presenter which always double quotes string values needing escapes."""
     if "\n" in data or "\t" in data or "'" in data:
@@ -965,7 +954,12 @@ def parse(
     formatter.dispatch_config(lnt)
 
     # Set up logging.
-    set_logging_level(verbosity=verbose, logger=logger, stderr_output=non_human_output)
+    set_logging_level(
+        verbosity=verbose,
+        formatter=formatter,
+        logger=logger,
+        stderr_output=non_human_output,
+    )
 
     # TODO: do this better
 
@@ -1001,7 +995,7 @@ def parse(
 
         # iterative print for human readout
         if format == FormatType.human.value:
-            violations_count = _print_out_violations_and_timing(
+            violations_count = formatter.print_out_violations_and_timing(
                 output_stream, bench, code_only, total_time, verbose, parsed_strings
             )
         else:
@@ -1030,7 +1024,7 @@ def parse(
 
     except OSError:  # pragma: no cover
         click.echo(
-            colorize(
+            formatter.colorize(
                 f"The path '{path}' could not be accessed. Check it exists.",
                 Color.red,
             ),
@@ -1053,52 +1047,6 @@ def parse(
         sys.exit(0)
 
 
-def _print_out_violations_and_timing(
-    output_stream: OutputStream,
-    bench: bool,
-    code_only: bool,
-    total_time: float,
-    verbose: int,
-    parsed_strings: List[ParsedString],
-) -> int:
-    """Used by human formatting during the parse."""
-    violations_count = 0
-    timing = TimingSummary()
-
-    for parsed_string in parsed_strings:
-        timing.add(parsed_string.time_dict)
-
-        if parsed_string.tree:
-            output_stream.write(parsed_string.tree.stringify(code_only=code_only))
-        else:
-            # TODO: Make this prettier
-            output_stream.write("...Failed to Parse...")  # pragma: no cover
-
-        violations_count += len(parsed_string.violations)
-        if parsed_string.violations:
-            output_stream.write("==== parsing violations ====")  # pragma: no cover
-        for v in parsed_string.violations:
-            output_stream.write(format_violation(v))  # pragma: no cover
-        if parsed_string.violations:
-            output_stream.write(
-                format_dialect_warning(parsed_string.config.get("dialect"))
-            )
-
-        if verbose >= 2:
-            output_stream.write("==== timings ====")
-            output_stream.write(cli_table(parsed_string.time_dict.items()))
-
-    if verbose >= 2 or bench:
-        output_stream.write("==== overall timings ====")
-        output_stream.write(cli_table([("Clock time", total_time)]))
-        timing_summary = timing.summary()
-        for step in timing_summary:
-            output_stream.write(f"=== {step} ===")
-            output_stream.write(cli_table(timing_summary[step].items()))
-
-    return violations_count
-
-
 # This "__main__" handler allows invoking SQLFluff using "python -m", which
 # simplifies the use of cProfile, e.g.:
 # python -m cProfile -s cumtime -m sqlfluff.cli.commands lint slow_file.sql
diff --git a/src/sqlfluff/cli/formatters.py b/src/sqlfluff/cli/formatters.py
index e72ae84df..d17ae3962 100644
--- a/src/sqlfluff/cli/formatters.py
+++ b/src/sqlfluff/cli/formatters.py
@@ -1,35 +1,22 @@
 """Defines the formatters for the CLI."""
-
-
 from io import StringIO
-from typing import List, Union
+import sys
+from typing import List, Optional, Tuple, Union
+
+import click
+from colorama import Style
 
 from sqlfluff.cli.helpers import (
-    colorize,
-    cli_table,
     get_package_version,
     get_python_version,
     get_python_implementation,
     pad_line,
+    wrap_field,
 )
 from sqlfluff.cli.outputstream import OutputStream
-from sqlfluff.core import SQLBaseError, FluffConfig, Linter
+from sqlfluff.core import SQLBaseError, FluffConfig, Linter, TimingSummary
 from sqlfluff.core.enums import Color
-from sqlfluff.core.linter import LintedFile
-
-
-def format_filename(
-    filename: str, success: Union[str, bool] = False, success_text: str = "PASS"
-) -> str:
-    """Format filenames."""
-    if isinstance(success, str):
-        status_string = success
-    else:
-        status_string = colorize(
-            success_text if success else "FAIL",
-            Color.green if success else Color.red,
-        )
-    return f"== [{colorize(filename, Color.lightgrey)}] {status_string}"
+from sqlfluff.core.linter import LintedFile, LintingResult, ParsedString
 
 
 def split_string_on_spaces(s: str, line_length: int = 100) -> List[str]:
@@ -54,79 +41,6 @@ def split_string_on_spaces(s: str, line_length: int = 100) -> List[str]:
     return line_buff
 
 
-def format_violation(violation: SQLBaseError, max_line_length: int = 90) -> str:
-    """Format a violation."""
-    if not isinstance(violation, SQLBaseError):  # pragma: no cover
-        raise ValueError(f"Unexpected violation format: {violation}")
-
-    desc: str = violation.desc()
-    line_elem = "   -" if violation.line_no is None else f"{violation.line_no:4d}"
-    pos_elem = "   -" if violation.line_pos is None else f"{violation.line_pos:4d}"
-
-    if violation.ignore:
-        desc = "IGNORE: " + desc  # pragma: no cover
-
-    split_desc = split_string_on_spaces(desc, line_length=max_line_length - 25)
-
-    out_buff = ""
-    # Grey out the violation if we're ignoring it.
-    section_color: Color = Color.lightgrey if violation.ignore else Color.blue
-    for idx, line in enumerate(split_desc):
-        if idx == 0:
-            rule_code = violation.rule_code().rjust(4)
-            if "PRS" in rule_code:
-                section_color = Color.red
-            out_buff += colorize(
-                f"L:{line_elem} | P:{pos_elem} | {rule_code} | ",
-                section_color,
-            )
-        else:
-            out_buff += (
-                "\n"
-                + (" " * 23)
-                + colorize(
-                    "| ",
-                    section_color,
-                )
-            )
-        out_buff += line
-    return out_buff
-
-
-def format_linting_stats(result, verbose=0):
-    """Format a set of stats given a `LintingResult`."""
-    text_buffer = StringIO()
-    all_stats = result.stats()
-    text_buffer.write("==== summary ====\n")
-    if verbose >= 2:
-        output_fields = [
-            "files",
-            "violations",
-            "clean files",
-            "unclean files",
-            "avg per file",
-            "unclean rate",
-            "status",
-        ]
-        special_formats = {"unclean rate": "{0:.0%}"}
-    else:
-        output_fields = ["violations", "status"]
-        special_formats = {}
-    # Generate content tuples, applying special formats for some fields
-    summary_content = [
-        (
-            key,
-            special_formats[key].format(all_stats[key])
-            if key in special_formats
-            else all_stats[key],
-        )
-        for key in output_fields
-    ]
-    # Render it all as a table
-    text_buffer.write(cli_table(summary_content, max_label_width=14))
-    return text_buffer.getvalue()
-
-
 def format_linting_result_header():
     """Format the header of a linting result output."""
     text_buffer = StringIO()
@@ -134,70 +48,6 @@ def format_linting_result_header():
     return text_buffer.getvalue()
 
 
-def format_config_vals(config_vals):
-    """Format an iterable of config values from a config object."""
-    text_buffer = StringIO()
-    for i, k, v in config_vals:
-        val = "" if v is None else str(v)
-        text_buffer.write(
-            ("    " * i)
-            + colorize(pad_line(str(k) + ":", 20, "left"), color=Color.lightgrey)
-            + pad_line(val, 20, "left")
-            + "\n"
-        )
-    return text_buffer.getvalue()
-
-
-def format_rules(linter: Linter, verbose: int = 0) -> str:
-    """Format the a set of rules given a `Linter`."""
-    text_buffer = StringIO()
-    text_buffer.write("==== sqlfluff - rules ====\n")
-    text_buffer.write(
-        cli_table(
-            linter.rule_tuples(),
-            col_width=80,
-            cols=1,
-            label_color=Color.blue,
-            val_align="left",
-        )
-    )
-    return text_buffer.getvalue()
-
-
-def format_dialects(dialect_readout, verbose=0):
-    """Format the dialects yielded by `dialect_readout`."""
-    text_buffer = StringIO()
-    text_buffer.write("==== sqlfluff - dialects ====\n")
-    readouts = [
-        (
-            dialect.label,
-            f"{dialect.name} dialect [inherits from '{dialect.inherits_from}']",
-        )
-        for dialect in dialect_readout()
-    ]
-    text_buffer.write(
-        cli_table(
-            readouts,
-            col_width=60,
-            cols=1,
-            label_color=Color.blue,
-            val_align="right",
-        )
-    )
-    return text_buffer.getvalue()
-
-
-def format_dialect_warning(dialect):
-    """Output a warning for parsing errors."""
-    return colorize(
-        (
-            "WARNING: Parsing errors found and dialect is set to "
-            f"'{dialect}'. Have you configured your dialect correctly?"
-        ),
-        Color.lightgrey,
-    )
-
-
 class OutputStreamFormatter:
     """Formatter which writes to an OutputStream.
 
@@ -220,15 +70,22 @@ class OutputStreamFormatter:
     def __init__(
         self,
         output_stream: OutputStream,
+        nocolor: bool,
         verbosity: int = 0,
         filter_empty: bool = True,
         output_line_length: int = 80,
     ):
         self._output_stream = output_stream
+        self.plain_output = self.should_produce_plain_output(nocolor)
         self._verbosity = verbosity
         self._filter_empty = filter_empty
         self.output_line_length = output_line_length
 
+    @staticmethod
+    def should_produce_plain_output(nocolor: bool) -> bool:
+        """Returns True if text output should be plain (not colored)."""
+        return nocolor or not sys.stdout.isatty()
+
     def _dispatch(self, s: str) -> None:
         """Dispatch a string to the callback.
 
@@ -254,19 +111,19 @@ class OutputStreamFormatter:
                 config_content.append(("dialect", linter.dialect.name))
             config_content += linter.templater.config_pairs()
             text_buffer.write(
-                cli_table(config_content, col_width=30, max_label_width=15)
+                self.cli_table(config_content, col_width=30, max_label_width=15)
             )
             text_buffer.write("\n")
             if linter.config.get("rule_allowlist"):
                 text_buffer.write(
-                    cli_table(
+                    self.cli_table(
                         [("rules", ", ".join(linter.config.get("rule_allowlist")))],
                         col_width=41,
                     )
                 )
             if self._verbosity > 1:
                 text_buffer.write("\n== Raw Config:\n")
-                text_buffer.write(format_config_vals(linter.config.iter_vals()))
+                text_buffer.write(self.format_config_vals(linter.config.iter_vals()))
         return text_buffer.getvalue()
 
     def dispatch_config(self, linter: Linter) -> None:
@@ -277,12 +134,11 @@ class OutputStreamFormatter:
         """Dispatch filenames during a persist operation."""
         # Only show the skip records at higher levels of verbosity
         if self._verbosity >= 2 or result != "SKIP":
-            self._dispatch(format_filename(filename=filename, success=result))
+            self._dispatch(self.format_filename(filename=filename, success=result))
 
-    @staticmethod
-    def _format_path(path: str) -> str:
+    def _format_path(self, path: str) -> str:
         """Format paths."""
-        return f"=== [ path: {colorize(path, Color.lightgrey)} ] ===\n"
+        return f"=== [ path: {self.colorize(path, Color.lightgrey)} ] ===\n"
 
     def dispatch_path(self, path: str) -> None:
         """Dispatch paths for display."""
@@ -294,7 +150,7 @@ class OutputStreamFormatter:
     ) -> None:
         """Dispatch the header displayed before templating."""
         if self._verbosity > 1:
-            self._dispatch(format_filename(filename=fname, success="TEMPLATING"))
+            self._dispatch(self.format_filename(filename=fname, success="TEMPLATING"))
             # This is where we output config diffs if they exist.
             if file_config:
                 # Only output config diffs if there is a config to diff to.
@@ -302,28 +158,30 @@ class OutputStreamFormatter:
                 if config_diff:  # pragma: no cover
                     self._dispatch("   Config Diff:")
                     self._dispatch(
-                        format_config_vals(linter_config.iter_vals(cfg=config_diff))
+                        self.format_config_vals(
+                            linter_config.iter_vals(cfg=config_diff)
+                        )
                     )
 
     def dispatch_parse_header(self, fname: str) -> None:
         """Dispatch the header displayed before parsing."""
         if self._verbosity > 1:
-            self._dispatch(format_filename(filename=fname, success="PARSING"))
+            self._dispatch(self.format_filename(filename=fname, success="PARSING"))
 
     def dispatch_lint_header(self, fname: str) -> None:
         """Dispatch the header displayed before linting."""
         if self._verbosity > 1:
-            self._dispatch(format_filename(filename=fname, success="LINTING"))
+            self._dispatch(self.format_filename(filename=fname, success="LINTING"))
 
     def dispatch_compilation_header(self, templater, message):
         """Dispatch the header displayed before linting."""
         self._dispatch(
-            f"=== [{colorize(templater, Color.lightgrey)}] {message}"
+            f"=== [{self.colorize(templater, Color.lightgrey)}] {message}"
         )  # pragma: no cover
 
     def dispatch_dialect_warning(self, dialect) -> None:
         """Dispatch a warning for dialects."""
-        self._dispatch(format_dialect_warning(dialect))  # pragma: no cover
+        self._dispatch(self.format_dialect_warning(dialect))  # pragma: no cover
 
     def _format_file_violations(
         self, fname: str, violations: List[SQLBaseError]
@@ -335,7 +193,7 @@ class OutputStreamFormatter:
 
         # Only print the filename if it's either a failure or verbosity > 1
         if self._verbosity > 0 or not success:
-            text_buffer.write(format_filename(fname, success=success))
+            text_buffer.write(self.format_filename(fname, success=success))
             text_buffer.write("\n")
 
         # If we have violations, print them
@@ -344,7 +202,9 @@ class OutputStreamFormatter:
             s = sorted(violations, key=lambda v: (v.line_no, v.line_pos))
             for violation in s:
                 text_buffer.write(
-                    format_violation(violation, max_line_length=self.output_line_length)
+                    self.format_violation(
+                        violation, max_line_length=self.output_line_length
+                    )
                 )
                 text_buffer.write("\n")
         str_buffer = text_buffer.getvalue()
@@ -361,3 +221,350 @@ class OutputStreamFormatter:
             fname, linted_file.get_violations(fixable=True if only_fixable else None)
         )
         self._dispatch(s)
+
+    def colorize(self, s: str, color: Optional[Color] = None) -> str:
+        """Optionally use ANSI colour codes to colour a string."""
+        return self.colorize_helper(self.plain_output, s, color)
+
+    @staticmethod
+    def colorize_helper(
+        plain_output: bool, s: str, color: Optional[Color] = None
+    ) -> str:
+        """Static version of colorize() method."""
+        if not color or plain_output:
+            return s
+        else:
+            return f"{color.value}{s}{Style.RESET_ALL}"
+
+    def cli_table_row(
+        self,
+        fields: List[Tuple[str, str]],
+        col_width,
+        max_label_width=10,
+        sep_char=": ",
+        divider_char=" ",
+        label_color=Color.lightgrey,
+        val_align="right",
+    ) -> str:
+        """Make a row of a CLI table, using wrapped values."""
+        # Do some intel first
+        cols = len(fields)
+        last_col_idx = cols - 1
+        wrapped_fields = [
+            wrap_field(
+                field[0],
+                field[1],
+                width=col_width,
+                max_label_width=max_label_width,
+                sep_char=sep_char,
+            )
+            for field in fields
+        ]
+        max_lines = max(fld["lines"] for fld in wrapped_fields)
+        last_line_idx = max_lines - 1
+        # Make some text
+        buff = StringIO()
+        for line_idx in range(max_lines):
+            for col_idx in range(cols):
+                # Assume we pad labels left and values right
+                fld = wrapped_fields[col_idx]
+                ll = fld["label_list"]
+                vl = fld["val_list"]
+                buff.write(
+                    self.colorize(
+                        pad_line(
+                            ll[line_idx] if line_idx < len(ll) else "",
+                            width=fld["label_width"],
+                        ),
+                        color=label_color,
+                    )
+                )
+                if line_idx == 0:
+                    buff.write(sep_char)
+                else:
+                    buff.write(" " * len(sep_char))
+                buff.write(
+                    pad_line(
+                        vl[line_idx] if line_idx < len(vl) else "",
+                        width=fld["val_width"],
+                        align=val_align,
+                    )
+                )
+                if col_idx != last_col_idx:
+                    buff.write(divider_char)
+                elif line_idx != last_line_idx:
+                    buff.write("\n")
+        return buff.getvalue()
+
+    def cli_table(
+        self,
+        fields,
+        col_width=20,
+        cols=2,
+        divider_char=" ",
+        sep_char=": ",
+        label_color=Color.lightgrey,
+        float_format="{0:.2f}",
+        max_label_width=10,
+        val_align="right",
+    ) -> str:
+        """Make a crude ascii table.
+
+        Assume that `fields` is an iterable of (label, value) pairs.
+        """
+        # First format all the values into strings
+        formatted_fields = []
+        for label, value in fields:
+            label = str(label)
+            if isinstance(value, float):
+                value = float_format.format(value)
+            else:
+                value = str(value)
+            formatted_fields.append((label, value))
+
+        # Set up a buffer to hold the whole table
+        buff = StringIO()
+        while len(formatted_fields) > 0:
+            row_buff: List[Tuple[str, str]] = []
+            while len(row_buff) < cols and len(formatted_fields) > 0:
+                row_buff.append(formatted_fields.pop(0))
+            buff.write(
+                self.cli_table_row(
+                    row_buff,
+                    col_width=col_width,
+                    max_label_width=max_label_width,
+                    sep_char=sep_char,
+                    divider_char=divider_char,
+                    label_color=label_color,
+                    val_align=val_align,
+                )
+            )
+            if len(formatted_fields) > 0:
+                buff.write("\n")
+        return buff.getvalue()
+
+    def format_filename(
+        self,
+        filename: str,
+        success: Union[str, bool] = False,
+        success_text: str = "PASS",
+    ) -> str:
+        """Format filenames."""
+        if isinstance(success, str):
+            status_string = success
+        else:
+            status_string = self.colorize(
+                success_text if success else "FAIL",
+                Color.green if success else Color.red,
+            )
+        return f"== [{self.colorize(filename, Color.lightgrey)}] {status_string}"
+
+    def format_violation(
+        self, violation: SQLBaseError, max_line_length: int = 90
+    ) -> str:
+        """Format a violation."""
+        if not isinstance(violation, SQLBaseError):  # pragma: no cover
+            raise ValueError(f"Unexpected violation format: {violation}")
+
+        desc: str = violation.desc()
+        line_elem = "   -" if violation.line_no is None else f"{violation.line_no:4d}"
+        pos_elem = "   -" if violation.line_pos is None else f"{violation.line_pos:4d}"
+
+        if violation.ignore:
+            desc = "IGNORE: " + desc  # pragma: no cover
+
+        split_desc = split_string_on_spaces(desc, line_length=max_line_length - 25)
+
+        out_buff = ""
+        # Grey out the violation if we're ignoring it.
+        section_color: Color = Color.lightgrey if violation.ignore else Color.blue
+        for idx, line in enumerate(split_desc):
+            if idx == 0:
+                rule_code = violation.rule_code().rjust(4)
+                if "PRS" in rule_code:
+                    section_color = Color.red
+                out_buff += self.colorize(
+                    f"L:{line_elem} | P:{pos_elem} | {rule_code} | ",
+                    section_color,
+                )
+            else:
+                out_buff += (
+                    "\n"
+                    + (" " * 23)
+                    + self.colorize(
+                        "| ",
+                        section_color,
+                    )
+                )
+            out_buff += line
+        return out_buff
+
+    def format_linting_stats(self, result, verbose=0):
+        """Format a set of stats given a `LintingResult`."""
+        text_buffer = StringIO()
+        all_stats = result.stats()
+        text_buffer.write("==== summary ====\n")
+        if verbose >= 2:
+            output_fields = [
+                "files",
+                "violations",
+                "clean files",
+                "unclean files",
+                "avg per file",
+                "unclean rate",
+                "status",
+            ]
+            special_formats = {"unclean rate": "{0:.0%}"}
+        else:
+            output_fields = ["violations", "status"]
+            special_formats = {}
+        # Generate content tuples, applying special formats for some fields
+        summary_content = [
+            (
+                key,
+                special_formats[key].format(all_stats[key])
+                if key in special_formats
+                else all_stats[key],
+            )
+            for key in output_fields
+        ]
+        # Render it all as a table
+        text_buffer.write(self.cli_table(summary_content, max_label_width=14))
+        return text_buffer.getvalue()
+
+    def format_config_vals(self, config_vals):
+        """Format an iterable of config values from a config object."""
+        text_buffer = StringIO()
+        for i, k, v in config_vals:
+            val = "" if v is None else str(v)
+            text_buffer.write(
+                ("    " * i)
+                + self.colorize(
+                    pad_line(str(k) + ":", 20, "left"), color=Color.lightgrey
+                )
+                + pad_line(val, 20, "left")
+                + "\n"
+            )
+        return text_buffer.getvalue()
+
+    def format_rules(self, linter: Linter, verbose: int = 0) -> str:
+        """Format the a set of rules given a `Linter`."""
+        text_buffer = StringIO()
+        text_buffer.write("==== sqlfluff - rules ====\n")
+        text_buffer.write(
+            self.cli_table(
+                linter.rule_tuples(),
+                col_width=80,
+                cols=1,
+                label_color=Color.blue,
+                val_align="left",
+            )
+        )
+        return text_buffer.getvalue()
+
+    def format_dialects(self, dialect_readout, verbose=0):
+        """Format the dialects yielded by `dialect_readout`."""
+        text_buffer = StringIO()
+        text_buffer.write("==== sqlfluff - dialects ====\n")
+        readouts = [
+            (
+                dialect.label,
+                f"{dialect.name} dialect [inherits from '{dialect.inherits_from}']",
+            )
+            for dialect in dialect_readout()
+        ]
+        text_buffer.write(
+            self.cli_table(
+                readouts,
+                col_width=60,
+                cols=1,
+                label_color=Color.blue,
+                val_align="right",
+            )
+        )
+        return text_buffer.getvalue()
+
+    def format_dialect_warning(self, dialect):
+        """Output a warning for parsing errors."""
+        return self.colorize(
+            (
+                "WARNING: Parsing errors found and dialect is set to "
+                f"'{dialect}'. Have you configured your dialect correctly?"
+            ),
+            Color.lightgrey,
+        )
+
+    def handle_files_with_tmp_or_prs_errors(self, lint_result: LintingResult) -> int:
+        """Discard lint fixes for files with templating or parse errors.
+
+        Returns 1 if there are any files with templating or parse errors after
+        filtering, else 0. (Intended as a process exit code.)
+        """
+        total_errors, num_filtered_errors = lint_result.count_tmp_prs_errors()
+        lint_result.discard_fixes_for_lint_errors_in_files_with_tmp_or_prs_errors()
+        if total_errors:
+            click.echo(
+                self.colorize(
+                    f"  [{total_errors} templating/parsing errors found]", Color.red
+                )
+            )
+            if num_filtered_errors < total_errors:
+                color = Color.red if num_filtered_errors else Color.green
+                click.echo(
+                    self.colorize(
+                        f"  [{num_filtered_errors} templating/parsing errors "
+                        f'remaining after "ignore"]',
+                        color,
+                    )
+                )
+        return 1 if num_filtered_errors else 0
+
+    def print_out_violations_and_timing(
+        self,
+        output_stream: OutputStream,
+        bench: bool,
+        code_only: bool,
+        total_time: float,
+        verbose: int,
+        parsed_strings: List[ParsedString],
+    ) -> int:
+        """Used by human formatting during the parse."""
+        violations_count = 0
+        timing = TimingSummary()
+
+        for parsed_string in parsed_strings:
+            timing.add(parsed_string.time_dict)
+
+            if parsed_string.tree:
+                output_stream.write(parsed_string.tree.stringify(code_only=code_only))
+            else:
+                # TODO: Make this prettier
+                output_stream.write("...Failed to Parse...")  # pragma: no cover
+
+            violations_count += len(parsed_string.violations)
+            if parsed_string.violations:
+                output_stream.write("==== parsing violations ====")  # pragma: no cover
+            for v in parsed_string.violations:
+                output_stream.write(self.format_violation(v))  # pragma: no cover
+            if parsed_string.violations:
+                output_stream.write(
+                    self.format_dialect_warning(parsed_string.config.get("dialect"))
+                )
+
+            if verbose >= 2:
+                output_stream.write("==== timings ====")
+                output_stream.write(self.cli_table(parsed_string.time_dict.items()))
+
+        if verbose >= 2 or bench:
+            output_stream.write("==== overall timings ====")
+            output_stream.write(self.cli_table([("Clock time", total_time)]))
+            timing_summary = timing.summary()
+            for step in timing_summary:
+                output_stream.write(f"=== {step} ===")
+                output_stream.write(self.cli_table(timing_summary[step].items()))
+
+        return violations_count
+
+    def completion_message(self) -> None:
+        """Prints message when SQLFluff is finished."""
+        click.echo("All Finished" f"{'' if self.plain_output else ' 📜 🎉'}!")
diff --git a/src/sqlfluff/cli/helpers.py b/src/sqlfluff/cli/helpers.py
index 7ea51c888..102c925bc 100644
--- a/src/sqlfluff/cli/helpers.py
+++ b/src/sqlfluff/cli/helpers.py
@@ -1,25 +1,10 @@
 """CLI helper utilities."""
 
-from io import StringIO
 import sys
 import textwrap
-from typing import Optional, List, Dict, Tuple
-
-from colorama import Style
+from typing import Dict, List
 
 from sqlfluff import __version__ as pkg_version
-from sqlfluff.core.enums import Color
-
-
-def colorize(s: str, color: Optional[Color] = None) -> str:
-    """Use ANSI colour codes to colour a string.
-
-    The name of this function is in American. I'm sorry :(.
-    """
-    if color:
-        return f"{color.value}{s}{Style.RESET_ALL}"
-    else:
-        return s
 
 
 def get_python_version() -> str:
@@ -84,110 +69,3 @@ def pad_line(s: str, width: int, align: str = "left") -> str:
         return (" " * gap) + s
     else:
         raise ValueError(f"Unknown alignment: {align}")  # pragma: no cover
-
-
-def cli_table_row(
-    fields: List[Tuple[str, str]],
-    col_width,
-    max_label_width=10,
-    sep_char=": ",
-    divider_char=" ",
-    label_color=Color.lightgrey,
-    val_align="right",
-) -> str:
-    """Make a row of a CLI table, using wrapped values."""
-    # Do some intel first
-    cols = len(fields)
-    last_col_idx = cols - 1
-    wrapped_fields = [
-        wrap_field(
-            field[0],
-            field[1],
-            width=col_width,
-            max_label_width=max_label_width,
-            sep_char=sep_char,
-        )
-        for field in fields
-    ]
-    max_lines = max(fld["lines"] for fld in wrapped_fields)
-    last_line_idx = max_lines - 1
-    # Make some text
-    buff = StringIO()
-    for line_idx in range(max_lines):
-        for col_idx in range(cols):
-            # Assume we pad labels left and values right
-            fld = wrapped_fields[col_idx]
-            ll = fld["label_list"]
-            vl = fld["val_list"]
-            buff.write(
-                colorize(
-                    pad_line(
-                        ll[line_idx] if line_idx < len(ll) else "",
-                        width=fld["label_width"],
-                    ),
-                    color=label_color,
-                )
-            )
-            if line_idx == 0:
-                buff.write(sep_char)
-            else:
-                buff.write(" " * len(sep_char))
-            buff.write(
-                pad_line(
-                    vl[line_idx] if line_idx < len(vl) else "",
-                    width=fld["val_width"],
-                    align=val_align,
-                )
-            )
-            if col_idx != last_col_idx:
-                buff.write(divider_char)
-            elif line_idx != last_line_idx:
-                buff.write("\n")
-    return buff.getvalue()
-
-
-def cli_table(
-    fields,
-    col_width=20,
-    cols=2,
-    divider_char=" ",
-    sep_char=": ",
-    label_color=Color.lightgrey,
-    float_format="{0:.2f}",
-    max_label_width=10,
-    val_align="right",
-) -> str:
-    """Make a crude ascii table.
-
-    Assume that `fields` is an iterable of (label, value) pairs.
-    """
-    # First format all the values into strings
-    formatted_fields = []
-    for label, value in fields:
-        label = str(label)
-        if isinstance(value, float):
-            value = float_format.format(value)
-        else:
-            value = str(value)
-        formatted_fields.append((label, value))
-
-    # Set up a buffer to hold the whole table
-    buff = StringIO()
-    while len(formatted_fields) > 0:
-        row_buff: List[Tuple[str, str]] = []
-        while len(row_buff) < cols and len(formatted_fields) > 0:
-            row_buff.append(formatted_fields.pop(0))
-        buff.write(
-            cli_table_row(
-                row_buff,
-                col_width=col_width,
-                max_label_width=max_label_width,
-                sep_char=sep_char,
-                divider_char=divider_char,
-                label_color=label_color,
-                val_align=val_align,
-            )
-        )
-        if len(formatted_fields) > 0:
-            buff.write("\n")
-    return buff.getvalue()


Test patch:
diff --git a/test/cli/commands_test.py b/test/cli/commands_test.py
index c15bd9d93..8a87f7e2f 100644
--- a/test/cli/commands_test.py
+++ b/test/cli/commands_test.py
@@ -4,6 +4,7 @@ import configparser
 import json
 import os
 import pathlib
+import re
 import shutil
 import stat
 import tempfile
@@ -25,6 +26,13 @@ from sqlfluff.cli.commands import lint, version, rules, fix, parse, dialects, ge
 from sqlfluff.core.rules.base import BaseRule, LintFix, LintResult
 from sqlfluff.core.parser.segments.raw import CommentSegment
 
+re_ansi_escape = re.compile(r"\x1b[^m]*m")
+
+
+def contains_ansi_escape(s: str) -> bool:
+    """Does the string contain ANSI escape codes (e.g. color)?"""
+    return re_ansi_escape.search(s) is not None
+
 
 def invoke_assert_code(
     ret_code=0,
@@ -1096,6 +1104,36 @@ def test__cli__command_fail_nice_not_found(command):
     assert "could not be accessed" in result.output
 
 
+@patch("click.utils.should_strip_ansi")
+@patch("sys.stdout.isatty")
+def test__cli__command_lint_nocolor(isatty, should_strip_ansi, capsys, tmpdir):
+    """Test the --nocolor option prevents color output."""
+    # Patch these two functions to make it think every output stream is a TTY.
+    # In spite of this, the output should not contain ANSI color codes because
+    # we specify "--nocolor" below.
+    isatty.return_value = True
+    should_strip_ansi.return_value = False
+    fpath = "test/fixtures/linter/indentation_errors.sql"
+    output_file = str(tmpdir / "result.txt")
+    cmd_args = [
+        "--verbose",
+        "--nocolor",
+        "--dialect",
+        "ansi",
+        "--disable_progress_bar",
+        fpath,
+        "--write-output",
+        output_file,
+    ]
+    with pytest.raises(SystemExit):
+        lint(cmd_args)
+    out = capsys.readouterr()[0]
+    assert not contains_ansi_escape(out)
+    with open(output_file, "r") as f:
+        file_contents = f.read()
+    assert not contains_ansi_escape(file_contents)
+
+
 @pytest.mark.parametrize(
     "serialize",
     ["human", "yaml", "json", "github-annotation", "github-annotation-native"],
diff --git a/test/cli/formatters_test.py b/test/cli/formatters_test.py
index 861551658..5ab615469 100644
--- a/test/cli/formatters_test.py
+++ b/test/cli/formatters_test.py
@@ -2,12 +2,15 @@
 
 import re
 
+from sqlfluff.core import FluffConfig
+from sqlfluff.core.enums import Color
 from sqlfluff.core.rules.base import RuleGhost
 from sqlfluff.core.parser import RawSegment
 from sqlfluff.core.parser.markers import PositionMarker
 from sqlfluff.core.errors import SQLLintError
 from sqlfluff.core.templaters.base import TemplatedFile
-from sqlfluff.cli.formatters import format_filename, format_violation
+from sqlfluff.cli.formatters import OutputStreamFormatter
+from sqlfluff.cli.outputstream import FileOutput
 
 
 def escape_ansi(line):
@@ -16,13 +19,16 @@ def escape_ansi(line):
     return ansi_escape.sub("", line)
 
 
-def test__cli__formatters__filename_nocol():
+def test__cli__formatters__filename_nocol(tmpdir):
     """Test formatting filenames."""
-    res = format_filename("blahblah", success=True)
+    formatter = OutputStreamFormatter(
+        FileOutput(FluffConfig(require_dialect=False), str(tmpdir / "out.txt")), False
+    )
+    res = formatter.format_filename("blahblah", success=True)
     assert escape_ansi(res) == "== [blahblah] PASS"
 
 
-def test__cli__formatters__violation():
+def test__cli__formatters__violation(tmpdir):
     """Test formatting violations.
 
     NB Position is 1 + start_pos.
@@ -37,10 +43,34 @@ def test__cli__formatters__violation():
     )
     r = RuleGhost("A", "DESC")
     v = SQLLintError(segment=s, rule=r)
-    f = format_violation(v)
+    formatter = OutputStreamFormatter(
+        FileOutput(FluffConfig(require_dialect=False), str(tmpdir / "out.txt")), False
+    )
+    f = formatter.format_violation(v)
     # Position is 3, 3 becase foobarbar is on the third
     # line (i.e. it has two newlines preceding it) and
     # it's at the third position in that line (i.e. there
     # are two characters between it and the preceding
     # newline).
     assert escape_ansi(f) == "L:   3 | P:   3 |    A | DESC"
+
+
+def test__cli__helpers__colorize(tmpdir):
+    """Test ANSI colouring."""
+    formatter = OutputStreamFormatter(
+        FileOutput(FluffConfig(require_dialect=False), str(tmpdir / "out.txt")), False
+    )
+    # Force color output for this test.
+    formatter.plain_output = False
+    assert formatter.colorize("foo", Color.red) == "\u001b[31mfoo\u001b[0m"
+
+
+def test__cli__helpers__cli_table(tmpdir):
+    """Test making tables."""
+    vals = [("a", 3), ("b", "c"), ("d", 4.7654), ("e", 9)]
+    formatter = OutputStreamFormatter(
+        FileOutput(FluffConfig(require_dialect=False), str(tmpdir / "out.txt")), False
+    )
+    txt = formatter.cli_table(vals, col_width=7, divider_char="|", label_color=None)
+    # NB: No trailing newline
+    assert txt == "a:    3|b:    c\nd: 4.77|e:    9"
diff --git a/test/cli/helpers_test.py b/test/cli/helpers_test.py
index b70a07c6e..47bd9756a 100644
--- a/test/cli/helpers_test.py
+++ b/test/cli/helpers_test.py
@@ -2,21 +2,7 @@
 
 import pytest
 
-from sqlfluff.cli.helpers import colorize, cli_table, wrap_elem, wrap_field, pad_line
-from sqlfluff.core.enums import Color
-
-
-def test__cli__helpers__colorize():
-    """Test ANSI colouring."""
-    assert colorize("foo", Color.red) == "\u001b[31mfoo\u001b[0m"
-
-
-def test__cli__helpers__cli_table():
-    """Test making tables."""
-    vals = [("a", 3), ("b", "c"), ("d", 4.7654), ("e", 9)]
-    txt = cli_table(vals, col_width=7, divider_char="|", label_color=None)
-    # NB: No trailing newline
-    assert txt == "a:    3|b:    c\nd: 4.77|e:    9"
+from sqlfluff.cli.helpers import wrap_elem, wrap_field, pad_line
 
 
 @pytest.mark.parametrize(
diff --git a/test/core/linter_test.py b/test/core/linter_test.py
index f520533c4..bd1a667d1 100644
--- a/test/core/linter_test.py
+++ b/test/core/linter_test.py
@@ -260,7 +260,7 @@ def test__linter__linting_parallel_thread(force_error, monkeypatch):
     config = FluffConfig(overrides={"dialect": "ansi"})
     output_stream = make_output_stream(config, None, os.devnull)
     lntr = Linter(
-        formatter=OutputStreamFormatter(output_stream, verbosity=0),
+        formatter=OutputStreamFormatter(output_stream, False, verbosity=0),
         dialect="ansi",
     )
     result = lntr.lint_paths(