Add a mechanism to silence noisy s3 simulator logs

python/arcticdb/storage_fixtures/s3.py

@@ -24,6 +24,7 @@
 from typing import Optional, Any, Type
 
 import werkzeug
+from werkzeug.serving import WSGIRequestHandler
 import botocore.exceptions
 from moto.server import DomainDispatcherApplication, create_backend_app
 
@@ -56,6 +57,42 @@
 logger = logging.getLogger("S3 Storage Fixture")
 
 
+def _is_truthy_env(var_name: str) -> bool:
+    return os.getenv(var_name, "").strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _configure_moto_server_logging() -> None:
+    # Disabled by default to keep test output clean. Enable with ARCTICDB_MOTO_VERBOSE=1.
+    if _is_truthy_env("ARCTICDB_MOTO_VERBOSE"):
+        return
+
+    werkzeug_logger = logging.getLogger("werkzeug")
+    werkzeug_logger.propagate = False
+    werkzeug_logger.setLevel(logging.ERROR)
+
+
+def _suppress_moto_server_stdio() -> None:
+    if _is_truthy_env("ARCTICDB_MOTO_VERBOSE"):
+        return
+
+    with open(os.devnull, "w") as devnull, contextlib.redirect_stdout(devnull), contextlib.redirect_stderr(devnull):
+        pass  # caller must restructure to wrap the werkzeug.run_simple call
+    sys.stdout = devnull
+    sys.stderr = devnull
+
+
+class _QuietMotoRequestHandler(WSGIRequestHandler):
+    def log(self, type, message, *args):
+        return
+
+    def log_request(self, code="-", size="-"):
+        return
+
+    def log_error(self, format, *args):
+        super().log_error(format, *args)
+        return
+
+
 class Key:
     def __init__(self, *, id: str, secret: str, user_name: str):
         self.id = id
@@ -769,22 +806,33 @@ def __call__(self, environ, start_response):
 
 
 def run_s3_server(port, key_file, cert_file):
+    _configure_moto_server_logging()
+    _suppress_moto_server_stdio()
+    verbose = _is_truthy_env("ARCTICDB_MOTO_VERBOSE")
+    _configure_moto_server_logging(verbose)
+    _suppress_moto_server_stdio(verbose)
+    request_handler = None if verbose else _QuietMotoRequestHandler
     werkzeug.run_simple(
         "0.0.0.0",
         port,
         HostDispatcherApplication(create_backend_app),
         threaded=True,
         ssl_context=(cert_file, key_file) if cert_file and key_file else None,
+        request_handler=request_handler,
     )
 
 
 def run_gcp_server(port, key_file, cert_file):
+    _configure_moto_server_logging()
+    _suppress_moto_server_stdio()
+    request_handler = None if _is_truthy_env("ARCTICDB_MOTO_VERBOSE") else _QuietMotoRequestHandler
     werkzeug.run_simple(
         "0.0.0.0",
         port,
         GcpHostDispatcherApplication(create_backend_app),
         threaded=True,
         ssl_context=(cert_file, key_file) if cert_file and key_file else None,
+        request_handler=request_handler,
     )