diff --git a/tests/gold_tests/logging/log_pipe.test.py b/tests/gold_tests/logging/log_pipe.test.py
index e550b75b5e3..fbfcc67b6f5 100644
--- a/tests/gold_tests/logging/log_pipe.test.py
+++ b/tests/gold_tests/logging/log_pipe.test.py
@@ -27,12 +27,16 @@
 ts_counter = 1
 
 
-def get_ts(logging_config):
+def get_ts(logging_config, disable_log_checks=False):
     """
     Create a Traffic Server process.
+
+    :param disable_log_checks: Whether to disable the default diagnostics log
+        checks on systems where the kernel may reject increasing the pipe
+        buffer size for an unprivileged Traffic Server process.
     """
     global ts_counter
-    ts = Test.MakeATSProcess("ts{}".format(ts_counter))
+    ts = Test.MakeATSProcess("ts{}".format(ts_counter), disable_log_checks=disable_log_checks)
     ts_counter += 1
 
     ts.Disk.records_config.update(
@@ -122,10 +126,15 @@ def get_ts(logging_config):
       mode: ascii_pipe
       format: custom
       pipe_buffer_size: {}
-      '''.format(pipe_name, pipe_size).split("\n"))
+      '''.format(pipe_name, pipe_size).split("\n"),
+    disable_log_checks=True)
 
 pipe_path = os.path.join(ts.Variables.LOGDIR, pipe_name)
 
+ts.Disk.diags_log.Content += Testers.ExcludesExpression(
+    r"ERROR:(?! Set pipe size failed for pipe .*: Operation not permitted)", "The diagnostics should contain no unexpected errors.")
+ts.Disk.diags_log.Content += Testers.ExcludesExpression("FATAL:", "The diagnostics should contain no fatal errors.")
+
 ts.Disk.traffic_out.Content += Testers.ContainsExpression(
     "Created named pipe .*{}".format(pipe_name), "Verify that the named pipe was created")
 
@@ -147,7 +156,8 @@ def get_ts(logging_config):
     "New buffer size for pipe.*{}".format(pipe_name), "Verify that the named pipe's size was adjusted")
 buffer_verifier = "pipe_buffer_is_larger_than.py"
 tr.Setup.Copy(buffer_verifier)
-verify_buffer_size = tr.Processes.Process("verify_buffer_size", f"{sys.executable} {buffer_verifier} {pipe_path} {pipe_size}")
+verify_buffer_size = tr.Processes.Process(
+    "verify_buffer_size", f"{sys.executable} {buffer_verifier} {pipe_path} {pipe_size} {ts.Disk.diags_log.AbsPath}")
 verify_buffer_size.Return = 0
 verify_buffer_size.Streams.All += Testers.ContainsExpression("Success", "The buffer size verifier should report success.")
 
diff --git a/tests/gold_tests/logging/pipe_buffer_is_larger_than.py b/tests/gold_tests/logging/pipe_buffer_is_larger_than.py
index 194dbfa1741..00fa7db43d0 100644
--- a/tests/gold_tests/logging/pipe_buffer_is_larger_than.py
+++ b/tests/gold_tests/logging/pipe_buffer_is_larger_than.py
@@ -19,7 +19,9 @@
 
 import argparse
 import fcntl
+import os
 import sys
+import time
 
 F_SETPIPE_SZ = 1031  # Linux 2.6.35+
 F_GETPIPE_SZ = 1032  # Linux 2.6.35+
@@ -32,10 +34,12 @@ def parse_args():
 
     parser.add_argument('minimum_buffer_size', help='The minimu buffer size for the pipe to expect.')
 
+    parser.add_argument('diags_log', help='The diagnostics log in which a kernel permission error may be reported.')
+
     return parser.parse_args()
 
 
-def test_fifo(fifo, minimum_buffer_size):
+def test_fifo(fifo, minimum_buffer_size, diags_log):
     try:
         fifo_fd = open(fifo, "rb+", buffering=0)
         buffer_size = fcntl.fcntl(fifo_fd, F_GETPIPE_SZ)
@@ -43,9 +47,21 @@ def test_fifo(fifo, minimum_buffer_size):
         if buffer_size >= int(minimum_buffer_size):
             print("Success. Size is: {} which is larger than: {}".format(buffer_size, minimum_buffer_size))
             return 0
-        else:
-            print("Fail. Size is: {} which is smaller than: {}".format(buffer_size, minimum_buffer_size))
-            return 1
+
+        # Diagnostic writes are asynchronous, so briefly wait for the
+        # explicit permission error before treating an unchanged size as a
+        # failure.
+        for _ in range(50):
+            if os.path.exists(diags_log):
+                with open(diags_log, encoding='utf-8') as diags:
+                    diagnostics = diags.read()
+                    if "Set pipe size failed" in diagnostics and "Operation not permitted" in diagnostics:
+                        print("Success. The kernel denied increasing the pipe buffer for the unprivileged ATS user.")
+                        return 0
+            time.sleep(0.1)
+
+        print("Fail. Size is: {} which is smaller than: {}".format(buffer_size, minimum_buffer_size))
+        return 1
     except Exception as e:
         print("Unable to open fifo, error: {}".format(str(e)))
         return 2
@@ -53,7 +69,7 @@ def test_fifo(fifo, minimum_buffer_size):
 
 def main():
     args = parse_args()
-    return test_fifo(args.pipe_name, args.minimum_buffer_size)
+    return test_fifo(args.pipe_name, args.minimum_buffer_size, args.diags_log)
 
 
 if __name__ == '__main__':