From c508bfb58bca496ea9e436ed2ac60b678910b829 Mon Sep 17 00:00:00 2001
From: Alex Willmer <alex@moreati.org.uk>
Date: Fri, 15 Aug 2025 17:39:40 +0100
Subject: [PATCH] tests: Check stdio is blocking in sudo contexts

refs #712
---
 docs/changelog.rst  |  1 +
 tests/stdio_test.py | 28 +++++++++++++++++++++++-----
 tests/testlib.py    |  9 +++++++++
 3 files changed, 33 insertions(+), 5 deletions(-)

diff --git a/docs/changelog.rst b/docs/changelog.rst
index dbf653dc..956e5dc9 100644
--- a/docs/changelog.rst
+++ b/docs/changelog.rst
@@ -25,6 +25,7 @@ In progress (unreleased)
 * :gh:issue:`1306` CI: Move sudo test users defaults into ``/etc/sudoers.d``
 * :gh:issue:`1306` preamble_size: Fix variability of measured command size
 * :gh:issue:`1306` tests: Count bytes written in ``stdio_test.StdIOTest``
+* :gh:issue:`1306` tests: Check stdio is blocking in sudo contexts
 
 
 v0.3.27 (2025-08-20)
diff --git a/tests/stdio_test.py b/tests/stdio_test.py
index 3c0e91d1..d60cf31b 100644
--- a/tests/stdio_test.py
+++ b/tests/stdio_test.py
@@ -1,28 +1,46 @@
+import unittest
+
 import testlib
 
 import stdio_checks
 
 
-class StdIOTest(testlib.RouterMixin, testlib.TestCase):
+class StdIOMixin(testlib.RouterMixin):
     """
     Test that stdin, stdout, and stderr conform to common expectations,
     such as blocking IO.
     """
-    def test_can_write_stdout_1_mib(self):
+    def check_can_write_stdout_1_mib(self, context):
         """
         Writing to stdout should not raise EAGAIN. Regression test for
         https://github.com/mitogen-hq/mitogen/issues/712.
         """
         size = 1 * 2**20
-        context = self.router.local()
         nwritten = context.call(stdio_checks.shout_stdout, size)
         self.assertEqual(nwritten, size)
 
-    def test_stdio_is_blocking(self):
-        context = self.router.local()
+    def check_stdio_is_blocking(self, context):
         stdin_blocking, stdout_blocking, stderr_blocking = context.call(
             stdio_checks.stdio_is_blocking,
         )
         self.assertTrue(stdin_blocking)
         self.assertTrue(stdout_blocking)
         self.assertTrue(stderr_blocking)
+
+
+class LocalTest(StdIOMixin, testlib.TestCase):
+    def test_can_write_stdout_1_mib(self):
+        self.check_can_write_stdout_1_mib(self.router.local())
+
+    def test_stdio_is_blocking(self):
+        self.check_stdio_is_blocking(self.router.local())
+
+
+class SudoTest(StdIOMixin, testlib.TestCase):
+    @unittest.skipIf(not testlib.have_sudo_nopassword(), 'Needs passwordless sudo')
+    def test_can_write_stdout_1_mib(self):
+        self.check_can_write_stdout_1_mib(self.router.sudo())
+
+    @unittest.skipIf(not testlib.have_sudo_nopassword(), 'Needs passwordless sudo')
+    def test_stdio_is_blocking(self):
+        self.check_stdio_is_blocking(self.router.sudo())
diff --git a/tests/testlib.py b/tests/testlib.py
index b91739e6..20b6e7c7 100644
--- a/tests/testlib.py
+++ b/tests/testlib.py
@@ -179,6 +179,15 @@ def have_python3():
     return _have_cmd(['python3'])
 
 
+def have_sudo_nopassword():
+    """
+    Return True if we can run `sudo` with no password, otherwise False.
+
+    Any cached credentials are ignored.
+    """
+    return _have_cmd(['sudo', '-kn', 'true'])
+
+
 def retry(fn, on, max_attempts, delay):
     for i in range(max_attempts):
         try: