Commit 1b73fa22 authored by Dom Sekotill's avatar Dom Sekotill
Browse files

Merge branch '22-fix-proc-stdin' into 'main' 

Fix behave_utils.proc stdin arguments

Closes #22

See merge request !12
parents 1a316f70 3c666795

behave_utils/proc.py

+2 −1
Original line number Diff line number Diff line
@@ -170,10 +170,11 @@ async def _exec_io( 
	kwargs: dict[str, Any],

) -> trio.Process:

	async with trio.open_nursery() as nursery:

		stdin = kwargs.pop('stdin') if 'stdin' in kwargs else PIPE if data else DEVNULL

		proc: trio.Process = await nursery.start(

			partial(

				trio.run_process, [*coerce_args(cmd)],

				stdin=PIPE if data else DEVNULL,

				stdin=stdin,

				stdout=PIPE, stderr=PIPE,

				check=False,

				**kwargs,

tests/unit/proc/test_async.py

+56 −0
Original line number Diff line number Diff line
@@ -71,6 +71,33 @@ class ExecIOTests(TestCase): 
		with self.assertRaises(TypeError):

			await proc.aexec_io(ECHO_OUTPUT, input=TEST_BYTES, stdin=sys.stdin)



	@trio_test()

	async def test_stdin(self) -> None:

		"""

		Check that calling with the "stdin" argument works as expected



		Regression test for #22

		"""

		with self.subTest(stdin="IO[bytes]"):

			rfd, wfd = os.pipe()

			with open(wfd, "wb") as pipe:

				pipe.write(TEST_BYTES)



			with open(rfd, "rb") as pipe:

				recv = await proc.aexec_io(ECHO_OUTPUT, stdin=pipe, deserialiser=bytes)



			self.assertEqual(recv, TEST_BYTES)



		with self.subTest(stdin="IO[str]"):

			rfd, wfd = os.pipe()

			with open(wfd, "wb") as pipe:

				pipe.write(TEST_BYTES)



			with open(rfd, "r") as pipe:

				recv = await proc.aexec_io(ECHO_OUTPUT, stdin=pipe, deserialiser=bytes)



			self.assertEqual(recv, TEST_BYTES)



	@trio_test()

	async def test_stdout(self) -> None:

		"""
@@ -136,6 +163,35 @@ class ExecutorTests(TestCase): 


		self.assertEqual(output, TEST_BYTES)



	@trio_test()

	async def test_stdin(self) -> None:

		"""

		Check that calling with the "stdin" argument works as expected



		Regression test for #22

		"""

		exe = proc.AsyncExecutor(*FIXTURE_CMD)



		with self.subTest(stdin="IO[bytes]"):

			rfd, wfd = os.pipe()

			with open(wfd, "wb") as pipe:

				pipe.write(TEST_BYTES)



			with open(rfd, "rb") as pipe:

				recv = await exe("echo", stdin=pipe, deserialiser=bytes)



			self.assertEqual(recv, TEST_BYTES)



		with self.subTest(stdin="IO[str]"):

			rfd, wfd = os.pipe()

			with open(wfd, "wb") as pipe:

				pipe.write(TEST_BYTES)



			with open(rfd, "r") as pipe:

				recv = await exe("echo", stdin=pipe, deserialiser=bytes)



			self.assertEqual(recv, TEST_BYTES)



	@trio_test()

	async def test_stdout(self) -> None:

		"""

tests/unit/proc/test_proc.py

+54 −0
Original line number Diff line number Diff line
@@ -88,6 +88,32 @@ class ExecIOTests(TestCase): 
		with self.assertRaises(TypeError):

			proc.exec_io(ECHO_OUTPUT, input=TEST_BYTES, stdin=sys.stdin)



	def test_stdin(self) -> None:

		"""

		Check that calling with the "stdin" argument works as expected



		Regression test for #22

		"""

		with self.subTest(stdin="IO[bytes]"):

			rfd, wfd = os.pipe()

			with open(wfd, "wb") as pipe:

				pipe.write(TEST_BYTES)



			with open(rfd, "rb") as pipe:

				recv = proc.exec_io(ECHO_OUTPUT, stdin=pipe, deserialiser=bytes)



			self.assertEqual(recv, TEST_BYTES)



		with self.subTest(stdin="IO[str]"):

			rfd, wfd = os.pipe()

			with open(wfd, "wb") as pipe:

				pipe.write(TEST_BYTES)



			with open(rfd, "r") as pipe:

				recv = proc.exec_io(ECHO_OUTPUT, stdin=pipe, deserialiser=bytes)



			self.assertEqual(recv, TEST_BYTES)



	def test_stdout(self) -> None:

		"""

		Check that calling with the "stdout" argument receives bytes from stdout
@@ -158,6 +184,34 @@ class ExecutorTests(TestCase): 


		self.assertEqual(output, TEST_BYTES)



	def test_stdin(self) -> None:

		"""

		Check that calling with the "stdin" argument works as expected



		Regression test for #22

		"""

		exe = proc.Executor(*FIXTURE_CMD)



		with self.subTest(stdin="IO[bytes]"):

			rfd, wfd = os.pipe()

			with open(wfd, "wb") as pipe:

				pipe.write(TEST_BYTES)



			with open(rfd, "rb") as pipe:

				recv = exe("echo", stdin=pipe, deserialiser=bytes)



			self.assertEqual(recv, TEST_BYTES)



		with self.subTest(stdin="IO[str]"):

			rfd, wfd = os.pipe()

			with open(wfd, "wb") as pipe:

				pipe.write(TEST_BYTES)



			with open(rfd, "r") as pipe:

				recv = exe("echo", stdin=pipe, deserialiser=bytes)



			self.assertEqual(recv, TEST_BYTES)



	def test_stdout(self) -> None:

		"""

		Check that calling with the "stdout" argument receives bytes from stdout