tests: Check stdio is blocking in sudo contexts

refs #712
pull/1307/head
Alex Willmer 4 months ago
parent 76f6eb741d
commit c508bfb58b

@ -25,6 +25,7 @@ In progress (unreleased)
* :gh:issue:`1306` CI: Move sudo test users defaults into ``/etc/sudoers.d`` * :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` preamble_size: Fix variability of measured command size
* :gh:issue:`1306` tests: Count bytes written in ``stdio_test.StdIOTest`` * :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) v0.3.27 (2025-08-20)

@ -1,28 +1,46 @@
import unittest
import testlib import testlib
import stdio_checks import stdio_checks
class StdIOTest(testlib.RouterMixin, testlib.TestCase): class StdIOMixin(testlib.RouterMixin):
""" """
Test that stdin, stdout, and stderr conform to common expectations, Test that stdin, stdout, and stderr conform to common expectations,
such as blocking IO. 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 Writing to stdout should not raise EAGAIN. Regression test for
https://github.com/mitogen-hq/mitogen/issues/712. https://github.com/mitogen-hq/mitogen/issues/712.
""" """
size = 1 * 2**20 size = 1 * 2**20
context = self.router.local()
nwritten = context.call(stdio_checks.shout_stdout, size) nwritten = context.call(stdio_checks.shout_stdout, size)
self.assertEqual(nwritten, size) self.assertEqual(nwritten, size)
def test_stdio_is_blocking(self): def check_stdio_is_blocking(self, context):
context = self.router.local()
stdin_blocking, stdout_blocking, stderr_blocking = context.call( stdin_blocking, stdout_blocking, stderr_blocking = context.call(
stdio_checks.stdio_is_blocking, stdio_checks.stdio_is_blocking,
) )
self.assertTrue(stdin_blocking) self.assertTrue(stdin_blocking)
self.assertTrue(stdout_blocking) self.assertTrue(stdout_blocking)
self.assertTrue(stderr_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())

@ -179,6 +179,15 @@ def have_python3():
return _have_cmd(['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): def retry(fn, on, max_attempts, delay):
for i in range(max_attempts): for i in range(max_attempts):
try: try:

Loading…
Cancel
Save