Add unit tests for the proc module (07f8a7cb) · Commits · Dom Sekotill / Behave Utils

tests/init.py

0 → 100644

+0 −0

Empty file added.

tests/unit/init.py

0 → 100644

+26 −0

Original line number	Diff line number	Diff line
		# Copyright 2022 Dominik Sekotill <dom.sekotill@kodo.org.uk>
		#
		# This Source Code Form is subject to the terms of the Mozilla Public
		# License, v. 2.0. If a copy of the MPL was not distributed with this
		# file, You can obtain one at http://mozilla.org/MPL/2.0/.

		"""
		Unit tests for behave_utils
		"""

		import unittest
		import warnings


		class TestCase(unittest.TestCase):
		"""
		Base class for all project test cases

		Extends the base class provided by `unittest`
		"""

		def setUp(self) -> None: # noqa: D102
		warnings.simplefilter("error", category=DeprecationWarning)

		def tearDown(self) -> None: # noqa: D102
		warnings.resetwarnings()

tests/unit/proc/init.py

0 → 100644

+0 −0

Empty file added.

tests/unit/proc/fixture_output.py

0 → 100755

+69 −0

Original line number	Diff line number	Diff line
		#!/usr/bin/env python3
		# Copyright 2022 Dominik Sekotill <dom.sekotill@kodo.org.uk>
		#
		# This Source Code Form is subject to the terms of the Mozilla Public
		# License, v. 2.0. If a copy of the MPL was not distributed with this
		# file, You can obtain one at http://mozilla.org/MPL/2.0/.

		"""
		A fixture script for producing example outputs
		"""

		import argparse
		import json
		import shutil
		import sys
		from typing import Callable
		from typing import NoReturn

		ErrorC = Callable[[str], NoReturn]


		def main() -> None:
		"""
		Produce various example outputs; CLI entrypoint
		"""
		argp = argparse.ArgumentParser()
		subs = argp.add_subparsers(required=True)

		jsonp = subs.add_parser("json", description=json_cmd.__doc__)
		jsonp.set_defaults(func=json_cmd, parser=jsonp)

		echop = subs.add_parser("echo", description=echo_cmd.__doc__)
		echop.set_defaults(func=echo_cmd, parser=echop)

		rcodep = subs.add_parser("rcode", description=rcode_cmd.__doc__)
		rcodep.add_argument("--code", type=int, default=1)
		rcodep.set_defaults(func=rcode_cmd, parser=rcodep)

		args = argp.parse_args()
		args.func(args, args.parser.error)


		def json_cmd(args: argparse.Namespace, error: ErrorC) -> None:
		"""
		Output a sample JSON string
		"""
		json.dump(
		{"example-output": True},
		sys.stdout,
		)


		def echo_cmd(args: argparse.Namespace, error: ErrorC) -> None:
		"""
		Echo everything from stdin to stdout
		"""
		shutil.copyfileobj(sys.stdin, sys.stdout)


		def rcode_cmd(args: argparse.Namespace, error: ErrorC) -> None:
		"""
		Return a non-zero return code
		"""
		if 0 >= args.code or args.code >= 128:
		raise error(f"bad value for --code: {args.code}")
		raise SystemExit(args.code)


		main()

tests/unit/proc/test_proc.py

0 → 100644

+214 −0

Original line number	Diff line number	Diff line
		# Copyright 2022 Dominik Sekotill <dom.sekotill@kodo.org.uk>
		#
		# This Source Code Form is subject to the terms of the Mozilla Public
		# License, v. 2.0. If a copy of the MPL was not distributed with this
		# file, You can obtain one at http://mozilla.org/MPL/2.0/.

		"""
		Unit tests for behave_utils.proc
		"""

		import io
		import os
		import subprocess
		import sys
		import warnings
		from sys import executable as python
		from typing import Any

		from behave_utils import json
		from behave_utils import proc

		from .. import TestCase

		FIXTURE_CMD = [python, "-BESsm", "tests.unit.proc.fixture_output"]
		JSON_OUTPUT = [*FIXTURE_CMD, "json"]
		ECHO_OUTPUT = [*FIXTURE_CMD, "echo"]

		TEST_BYTES = b"""lorem ipsum dolorum"""


		class ExecIOTests(TestCase):
		"""
		Tests for the behave_utils.proc.exec_io function
		"""

		def test_deserialiser(self) -> None:
		"""
		Check that calling with a deserialiser correctly deserialises output
		"""
		with self.subTest(deserialiser=bytes):
		output: Any = proc.exec_io(JSON_OUTPUT, deserialiser=bytes)
		self.assertIsInstance(output, bytes)

		with self.subTest(deserialiser=json.JSONObject):
		output = proc.exec_io(JSON_OUTPUT, deserialiser=json.JSONObject.from_string)
		self.assertIsInstance(output, json.JSONObject)

		def test_deserialiser_with_stdout(self) -> None:
		"""
		Check that calling with both deserialiser and stdout raises TypeError
		"""
		with self.assertRaises(TypeError):
		proc.exec_io(ECHO_OUTPUT, stdout=sys.stdout, deserialiser=bytes)

		def test_input(self) -> None:
		"""
		Check that calling with the "input" argument passes bytes to stdin
		"""
		output = proc.exec_io(ECHO_OUTPUT, input=TEST_BYTES, deserialiser=bytes)

		self.assertEqual(output, TEST_BYTES)

		def test_data(self) -> None:
		"""
		Check that calling with the deprecated "data" argument passes bytes to stdin
		"""
		with warnings.catch_warnings(record=True) as messages:
		warnings.simplefilter("always")

		output = proc.exec_io(ECHO_OUTPUT, data=TEST_BYTES, deserialiser=bytes)

		self.assertEqual(output, TEST_BYTES)
		assert len(messages) == 1 and issubclass(messages[0].category, DeprecationWarning)

		def test_input_with_data(self) -> None:
		"""
		Check that calling with both "input" and "data" arguments raises TypeError
		"""
		with self.assertRaises(TypeError):
		proc.exec_io(ECHO_OUTPUT, input=TEST_BYTES, data=TEST_BYTES)

		def test_input_with_stdin(self) -> None:
		"""
		Check that calling with both "input" and "stdin" arguments raises TypeError
		"""
		with self.assertRaises(TypeError):
		proc.exec_io(ECHO_OUTPUT, input=TEST_BYTES, stdin=sys.stdin)

		def test_stdout(self) -> None:
		"""
		Check that calling with the "stdout" argument receives bytes from stdout
		"""
		with self.subTest(stdout="BytesIO"):
		bbuff = io.BytesIO()

		code = proc.exec_io(ECHO_OUTPUT, input=TEST_BYTES, stdout=bbuff)

		bbuff.seek(0)
		self.assertEqual(code, 0)
		self.assertEqual(bbuff.read(), TEST_BYTES)

		with self.subTest(stdout="StringIO"):
		sbuff = io.StringIO()

		code = proc.exec_io(ECHO_OUTPUT, input=TEST_BYTES, stdout=sbuff)

		sbuff.seek(0)
		self.assertEqual(code, 0)
		self.assertEqual(sbuff.read(), TEST_BYTES.decode())

		with self.subTest(stdout="pipe"):
		read_fd, write_fd = os.pipe()

		code = proc.exec_io(ECHO_OUTPUT, input=TEST_BYTES, stdout=write_fd)

		os.close(write_fd)
		with io.open(read_fd, mode="rb") as pipe:
		self.assertEqual(pipe.read(), TEST_BYTES)
		self.assertEqual(code, 0)


		class ExecutorTests(TestCase):
		"""
		Tests for the behave_utils.proc.Executor class
		"""

		def test_deserialiser(self) -> None:
		"""
		Check that calling with a deserialiser correctly deserialises output
		"""
		exe = proc.Executor(*FIXTURE_CMD)

		with self.subTest(deserialiser=bytes):
		output: Any = exe("json", deserialiser=bytes)
		self.assertIsInstance(output, bytes)

		with self.subTest(deserialiser=json.JSONObject):
		output = exe("json", deserialiser=json.JSONObject.from_string)
		self.assertIsInstance(output, json.JSONObject)

		def test_input(self) -> None:
		"""
		Check that calling with the "input" argument passes bytes to stdin
		"""
		exe = proc.Executor(*FIXTURE_CMD)

		output = exe("echo", input=TEST_BYTES, deserialiser=bytes)

		self.assertEqual(output, TEST_BYTES)

		def test_stdout(self) -> None:
		"""
		Check that calling with the "stdout" argument receives bytes from stdout
		"""
		exe = proc.Executor(*FIXTURE_CMD)

		with self.subTest(stdout="BytesIO"):
		bbuff = io.BytesIO()

		exe("echo", input=TEST_BYTES, stdout=bbuff)

		bbuff.seek(0)
		self.assertEqual(bbuff.read(), TEST_BYTES)

		with self.subTest(stdout="StringIO"):
		sbuff = io.StringIO()

		exe("echo", input=TEST_BYTES, stdout=sbuff)

		sbuff.seek(0)
		self.assertEqual(sbuff.read(), TEST_BYTES.decode())

		with self.subTest(stdout="pipe"):
		read_fd, write_fd = os.pipe()

		exe("echo", input=TEST_BYTES, stdout=write_fd)

		os.close(write_fd)
		with io.open(read_fd, mode="rb") as pipe:
		self.assertEqual(pipe.read(), TEST_BYTES)

		def test_return_code(self) -> None:
		"""
		Check that the "query" argument behaves as expected
		"""
		exe = proc.Executor(*FIXTURE_CMD)

		with self.subTest(query=True):
		exe("rcode", "--code=3", query=True)

		with self.subTest(query=False), self.assertRaises(subprocess.CalledProcessError):
		exe("rcode", "--code=3", query=False)

		with self.subTest(query=None), self.assertRaises(subprocess.CalledProcessError):
		exe("rcode", "--code=3")

		def test_subcommand(self) -> None:
		"""
		Check that the subcommand method returns a new instance with appended arguments
		"""
		class NewExecutor(proc.Executor):
		...

		with self.subTest(cls=proc.Executor):
		exe = proc.Executor("foo", "bar").subcommand("baz")

		self.assertIsInstance(exe, proc.Executor)
		self.assertListEqual(exe, ["foo", "bar", "baz"])

		with self.subTest(cls=NewExecutor):
		exe = NewExecutor("foo", "bar").subcommand("baz")

		self.assertIsInstance(exe, NewExecutor)
		self.assertListEqual(exe, ["foo", "bar", "baz"])