diff --git a/tools/test-backend b/tools/test-backend
index dc7bc6fded..794a306aee 100755
--- a/tools/test-backend
+++ b/tools/test-backend
@@ -108,6 +108,7 @@ not_yet_fully_covered = {path for target in [
     'zerver/lib/server_initialization.py',
     'zerver/lib/test_fixtures.py',
     'zerver/lib/test_runner.py',
+    'zerver/lib/test_console_output.py',
     'zerver/openapi/python_examples.py',
     # Tornado should ideally have full coverage, but we're not there.
     'zerver/tornado/autoreload.py',
@@ -260,8 +261,14 @@ def main() -> None:
                         default=False,
                         help=("Generate Stripe test fixtures by making requests to Stripe test network"))
     parser.add_argument('args', nargs='*')
+    parser.add_argument('--ban-console-output', dest='ban_console_output',
+                        action="store_true",
+                        default=False, help='Require stdout and stderr to be clean of unexpected output.')
 
     options = parser.parse_args()
+    if options.ban_console_output:
+        os.environ["BAN_CONSOLE_OUTPUT"] = "1"
+
     args = options.args
     parallel = options.processes
     include_webhooks = options.coverage or options.include_webhooks
diff --git a/zerver/lib/test_classes.py b/zerver/lib/test_classes.py
index 4c524886a0..8a6172ea5d 100644
--- a/zerver/lib/test_classes.py
+++ b/zerver/lib/test_classes.py
@@ -6,7 +6,7 @@ import tempfile
 import urllib
 from contextlib import contextmanager
 from typing import Any, Dict, Iterable, Iterator, List, Optional, Sequence, Set, Tuple, Union
-from unittest import mock
+from unittest import TestResult, mock
 
 import orjson
 from django.apps import apps
@@ -41,6 +41,12 @@ from zerver.lib.streams import (
     create_stream_if_needed,
     get_default_value_for_history_public_to_subscribers,
 )
+from zerver.lib.test_console_output import (
+    ExtraConsoleOutputFinder,
+    ExtraConsoleOutputInTestException,
+    TeeStderrAndFindExtraConsoleOutput,
+    TeeStdoutAndFindExtraConsoleOutput,
+)
 from zerver.lib.test_helpers import find_key_by_email, instrument_url
 from zerver.lib.users import get_api_key
 from zerver.lib.validator import check_string
@@ -113,6 +119,35 @@ class ZulipTestCase(TestCase):
                 self.mock_ldap.reset()
             self.mock_initialize.stop()
 
+    def run(self, result: Optional[TestResult]=None) -> Optional[TestResult]:  # nocoverage
+        if not settings.BAN_CONSOLE_OUTPUT:
+            return super(ZulipTestCase, self).run(result)
+        extra_output_finder = ExtraConsoleOutputFinder()
+        with TeeStderrAndFindExtraConsoleOutput(extra_output_finder), TeeStdoutAndFindExtraConsoleOutput(extra_output_finder):
+            test_result = super(ZulipTestCase, self).run(result)
+        if extra_output_finder.full_extra_output:
+            exception_message = f"""
+---- UNEXPECTED CONSOLE OUTPUT DETECTED ----
+
+To ensure that we never miss important error output/warnings,
+we require test-backend to have clean console output.
+
+This message usually is triggered by forgotten debugging print()
+statements or new logging statements.  For the latter, you can
+use `with self.assertLogs()` to capture and verify the log output;
+use `git grep assertLogs` to see dozens of correct examples.
+
+You should be able to quickly reproduce this failure with:
+
+test-backend --ban-console-output {self.id()}
+
+Output:
+{extra_output_finder.full_extra_output}
+--------------------------------------------
+"""
+            raise ExtraConsoleOutputInTestException(exception_message)
+        return test_result
+
     '''
     WRAPPER_COMMENT:
 
diff --git a/zerver/lib/test_console_output.py b/zerver/lib/test_console_output.py
new file mode 100644
index 0000000000..58b452e20f
--- /dev/null
+++ b/zerver/lib/test_console_output.py
@@ -0,0 +1,114 @@
+import logging
+import re
+import sys
+from types import TracebackType
+from typing import Iterable, Optional, Type, cast
+
+
+class ExtraConsoleOutputInTestException(Exception):
+    pass
+
+class ExtraConsoleOutputFinder:
+    def __init__(self) -> None:
+        self.latest_test_name = ""
+        valid_line_patterns = [
+            # Example: Running zerver.tests.test_attachments.AttachmentsTests.test_delete_unauthenticated
+            "^Running ",
+
+            # Example: ** Test is TOO slow: analytics.tests.test_counts.TestRealmActiveHumans.test_end_to_end (0.581 s)
+            "^\\*\\* Test is TOO slow: ",
+            "^----------------------------------------------------------------------",
+
+            # Example: INFO: URL coverage report is in var/url_coverage.txt
+            "^INFO: URL coverage report is in",
+
+            # Example: INFO: Try running: ./tools/create-test-api-docs
+            "^INFO: Try running:",
+
+            # Example: -- Running tests in parallel mode with 4 processes
+            "^-- Running tests in",
+            "^OK",
+
+            # Example: Ran 2139 tests in 115.659s
+            "^Ran [0-9]+ tests in",
+
+            # Destroying test database for alias 'default'...
+            "^Destroying test database for alias ",
+            "^Using existing clone",
+            "^\\*\\* Skipping ",
+        ]
+        self.compiled_line_patterns = []
+        for pattern in valid_line_patterns:
+            self.compiled_line_patterns.append(re.compile(pattern))
+        self.full_extra_output = ""
+
+    def find_extra_output(self, data: str) -> None:
+        lines = data.split('\n')
+        for line in lines:
+            if not line:
+                continue
+            found_extra_output = True
+            for compiled_pattern in self.compiled_line_patterns:
+                if compiled_pattern.match(line):
+                    found_extra_output = False
+                    break
+            if found_extra_output:
+                self.full_extra_output += f'{line}\n'
+
+class TeeStderrAndFindExtraConsoleOutput():
+    def __init__(self, extra_output_finder: ExtraConsoleOutputFinder) -> None:
+        self.stderr_stream = sys.stderr
+
+        # get shared console handler instance from any logger that have it
+        self.console_log_handler = cast(logging.StreamHandler, logging.getLogger('django.server').handlers[0])
+
+        assert isinstance(self.console_log_handler, logging.StreamHandler)
+        assert self.console_log_handler.stream == sys.stderr
+        self.extra_output_finder = extra_output_finder
+
+    def __enter__(self) -> None:
+        sys.stderr = self  # type: ignore[assignment] # Doing tee by swapping stderr stream with custom file like class
+        self.console_log_handler.stream = self  # type: ignore[assignment] # Doing tee by swapping stderr stream with custom file like class
+
+    def __exit__(self, exc_type: Optional[Type[BaseException]],
+                 exc_value: Optional[BaseException],
+                 traceback: Optional[TracebackType]) -> None:
+        sys.stderr = self.stderr_stream
+        self.console_log_handler.stream = sys.stderr
+
+    def write(self, data: str) -> None:
+        self.stderr_stream.write(data)
+        self.extra_output_finder.find_extra_output(data)
+
+    def writelines(self, data: Iterable[str]) -> None:
+        self.stderr_stream.writelines(data)
+        lines = "".join(data)
+        self.extra_output_finder.find_extra_output(lines)
+
+    def flush(self) -> None:
+        self.stderr_stream.flush()
+
+class TeeStdoutAndFindExtraConsoleOutput():
+    def __init__(self, extra_output_finder: ExtraConsoleOutputFinder) -> None:
+        self.stdout_stream = sys.stdout
+        self.extra_output_finder = extra_output_finder
+
+    def __enter__(self) -> None:
+        sys.stdout = self  # type: ignore[assignment] # Doing tee by swapping stderr stream with custom file like class
+
+    def __exit__(self, exc_type: Optional[Type[BaseException]],
+                 exc_value: Optional[BaseException],
+                 traceback: Optional[TracebackType]) -> None:
+        sys.stdout = self.stdout_stream
+
+    def write(self, data: str) -> None:
+        self.stdout_stream.write(data)
+        self.extra_output_finder.find_extra_output(data)
+
+    def writelines(self, data: Iterable[str]) -> None:
+        self.stdout_stream.writelines(data)
+        lines = "".join(data)
+        self.extra_output_finder.find_extra_output(lines)
+
+    def flush(self) -> None:
+        self.stdout_stream.flush()
diff --git a/zproject/computed_settings.py b/zproject/computed_settings.py
index 08ba32bc31..63bdd1dcd6 100644
--- a/zproject/computed_settings.py
+++ b/zproject/computed_settings.py
@@ -104,6 +104,8 @@ CASPER_TESTS = False
 RUNNING_OPENAPI_CURL_TEST = False
 # This is overridden in test_settings.py for the test suites
 GENERATE_STRIPE_FIXTURES = False
+# This is overridden in test_settings.py for the test suites
+BAN_CONSOLE_OUTPUT = False
 
 # Google Compute Engine has an /etc/boto.cfg that is "nicely
 # configured" to work with GCE's storage service.  However, their
diff --git a/zproject/test_extra_settings.py b/zproject/test_extra_settings.py
index 1389729db3..9433e7bde2 100644
--- a/zproject/test_extra_settings.py
+++ b/zproject/test_extra_settings.py
@@ -56,6 +56,9 @@ if "RUNNING_OPENAPI_CURL_TEST" in os.environ:
 if "GENERATE_STRIPE_FIXTURES" in os.environ:
     GENERATE_STRIPE_FIXTURES = True
 
+if "BAN_CONSOLE_OUTPUT" in os.environ:
+    BAN_CONSOLE_OUTPUT = True
+
 # Decrease the get_updates timeout to 1 second.
 # This allows CasperJS to proceed quickly to the next test step.
 POLL_TIMEOUT = 1000