pw_emu/py/tests/cli_test.py - pigweed/pigweed - Git at Google

 #!/usr/bin/env python
 # Copyright 2023 The Pigweed Authors
 #
 # Licensed under the Apache License, Version 2.0 (the "License"); you may not
 # use this file except in compliance with the License. You may obtain a copy of
 # the License at
 #
 #     https://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 # License for the specific language governing permissions and limitations under
 # the License.
 """Tests for the command line interface"""

 import os
 import signal
 import subprocess
 import sys
 import time
 import unittest

 from pathlib import Path

 from mock_emu_frontend import _mock_emu
 from config_helper import ConfigHelper


 # TODO: b/301382004 - The Python Pigweed package install (into python-venv)
 # races with running this test and there is no way to add that package as a test
 # depedency without creating circular depedencies. This means we can't rely on
 # using Pigweed tools like pw cli or the arm-none-eabi-gdb wrapper.
 #
 # Run the CLI directly instead of going through pw cli.
 _cli_path = Path(
     os.path.join(os.environ['PW_ROOT'], 'pw_emu', 'py', 'pw_emu', '__main__.py')
 ).resolve()


 class TestCli(ConfigHelper):
     """Test non-interactive commands"""

     _config = {
         'emulators': {
             'mock-emu': {
                 'launcher': 'mock_emu_frontend.MockEmuLauncher',
                 'connector': 'mock_emu_frontend.MockEmuConnector',
             }
         },
         'mock-emu': {
             'tcp_channel': True,
             'gdb_channel': True,
         },
         'gdb': _mock_emu + ['--exit', '--'],
         'targets': {'test-target': {'mock-emu': {}}},
     }

     def _build_cmd(self, args: list[str]) -> list[str]:
         cmd = [
             'python',
             str(_cli_path),
             '--working-dir',
             self._wdir.name,
             '--config',
             self._config_file,
         ] + args
         return cmd

     def _run(self, args: list[str], **kwargs) -> subprocess.CompletedProcess:
         """Run the CLI and wait for completion"""
         return subprocess.run(self._build_cmd(args), **kwargs)

     def _popen(self, args: list[str], **kwargs) -> subprocess.Popen:
         """Run the CLI in the background"""
         return subprocess.Popen(self._build_cmd(args), **kwargs)


 class TestNonInteractive(TestCli):
     """Test non interactive commands."""

     def setUp(self) -> None:
         super().setUp()
         self.assertEqual(self._run(['start', 'test-target']).returncode, 0)

     def tearDown(self) -> None:
         self.assertEqual(self._run(['stop']).returncode, 0)
         super().tearDown()

     def test_already_running(self) -> None:
         self.assertNotEqual(self._run(['start', 'test-target']).returncode, 0)

     def test_gdb_cmds(self) -> None:
         status = self._run(
             ['gdb-cmds', 'show version'],
         )
         self.assertEqual(status.returncode, 0)

     def test_prop_ls(self) -> None:
         status = self._run(['prop-ls', 'path1'], stdout=subprocess.PIPE)
         self.assertEqual(status.returncode, 0)
         self.assertTrue('prop1' in status.stdout.decode('ascii'))
         status = self._run(['prop-ls', 'invalid path'], stdout=subprocess.PIPE)
         self.assertNotEqual(status.returncode, 0)

     def test_prop_get(self) -> None:
         status = self._run(
             ['prop-get', 'invalid path', 'prop1'],
             stdout=subprocess.PIPE,
         )
         self.assertNotEqual(status.returncode, 0)
         status = self._run(
             ['prop-get', 'path1', 'invalid prop'],
             stdout=subprocess.PIPE,
         )
         self.assertNotEqual(status.returncode, 0)
         status = self._run(
             ['prop-get', 'path1', 'prop1'],
             stdout=subprocess.PIPE,
         )
         self.assertEqual(status.returncode, 0)
         self.assertTrue('val1' in status.stdout.decode('ascii'))

     def test_prop_set(self) -> None:
         status = self._run(
             ['prop-set', 'invalid path', 'prop1', 'v'],
             stdout=subprocess.PIPE,
         )
         self.assertNotEqual(status.returncode, 0)
         status = self._run(
             ['prop-set', 'path1', 'invalid prop', 'v'],
             stdout=subprocess.PIPE,
         )
         self.assertNotEqual(status.returncode, 0)
         status = self._run(
             ['prop-set', 'path1', 'prop1', 'value'],
             stdout=subprocess.PIPE,
         )
         self.assertEqual(status.returncode, 0)
         status = self._run(
             ['prop-get', 'path1', 'prop1'],
             stdout=subprocess.PIPE,
         )
         self.assertEqual(status.returncode, 0)
         self.assertTrue('value' in status.stdout.decode('ascii'), status.stdout)

     def test_reset(self) -> None:
         self.assertEqual(self._run(['reset']).returncode, 0)
         self.assertTrue(os.path.exists(os.path.join(self._wdir.name, 'reset')))

     def test_load(self) -> None:
         self.assertEqual(self._run(['load', 'executable']).returncode, 0)

     def test_resume(self) -> None:
         self.assertEqual(self._run(['resume']).returncode, 0)


 class TestForeground(TestCli):
     """Test starting in foreground"""

     def _test_common(self, cmd) -> None:
         # Run the CLI process in a new session so that we can terminate both the
         # CLI and the mock emulator it spawns in the foreground.
         args = {}
         if sys.platform == 'win32':
             args['creationflags'] = subprocess.CREATE_NEW_PROCESS_GROUP
         else:
             args['start_new_session'] = True
         proc = self._popen(cmd, stdout=subprocess.PIPE, **args)
         assert proc.stdout
         output = proc.stdout.readline()
         self.assertTrue(
             'starting mock emulator' in output.decode('utf-8'),
             output.decode('utf-8'),
         )
         if sys.platform == 'win32':
             # See https://bugs.python.org/issue26350
             os.kill(proc.pid, signal.CTRL_BREAK_EVENT)
         else:
             os.kill(-proc.pid, signal.SIGTERM)
         proc.wait()
         proc.stdout.close()

     def test_foreground(self) -> None:
         self._test_common(['start', '--foreground', 'test-target'])

     def test_debug(self) -> None:
         self._test_common(['start', '--debug', 'test-target'])


 class TestInteractive(TestCli):
     """Test interactive commands"""

     def setUp(self) -> None:
         super().setUp()
         self.assertEqual(self._run(['start', 'test-target']).returncode, 0)

     def tearDown(self) -> None:
         self.assertEqual(self._run(['stop']).returncode, 0)
         super().tearDown()

     @staticmethod
     def _read_nonblocking(fd: int, size: int) -> bytes:
         try:
             return os.read(fd, size)
         except BlockingIOError:
             return b''

     def test_term(self) -> None:
         """Test the pw emu term command"""

         if sys.platform == 'win32':
             self.skipTest('pty not supported on win32')

         # pylint: disable=import-outside-toplevel
         # Can't import pty on win32.
         import pty

         # pylint: disable=no-member
         # Avoid pylint false positive on win32.
         pid, fd = pty.fork()
         if pid == 0:
             status = self._run(['term', 'tcp'])
             # pylint: disable=protected-access
             # Use os._exit instead of os.exit after fork.
             os._exit(status.returncode)
         else:
             expected = '--- Miniterm on tcp ---'

             # Read the expected string with a timeout.
             os.set_blocking(fd, False)
             deadline = time.monotonic() + 5
             data = self._read_nonblocking(fd, len(expected))
             while len(data) < len(expected):
                 time.sleep(0.1)
                 data += self._read_nonblocking(fd, len(expected) - len(data))
                 if time.monotonic() > deadline:
                     break
             self.assertTrue(
                 expected in data.decode('ascii'),
                 data + self._read_nonblocking(fd, 100),
             )

             # send CTRL + ']' to terminate miniterm
             os.write(fd, b'\x1d')

             # wait for the process to exit, with a timeout
             deadline = time.monotonic() + 5
             wait_pid, ret = os.waitpid(pid, os.WNOHANG)
             while wait_pid == 0:
                 time.sleep(0.1)
                 # Discard input to avoid writer hang on MacOS,
                 # see https://github.com/python/cpython/issues/97001.
                 try:
                     self._read_nonblocking(fd, 100)
                 except OSError:
                     # Avoid read errors when the child pair of the pty
                     # closes when the child terminates.
                     pass
                 wait_pid, ret = os.waitpid(pid, os.WNOHANG)
                 if time.monotonic() > deadline:
                     break
             self.assertEqual(wait_pid, pid)
             self.assertEqual(ret, 0)

     def test_gdb(self) -> None:
         res = self._run(['gdb', '-e', 'executable'], stdout=subprocess.PIPE)
         self.assertEqual(res.returncode, 0)
         output = res.stdout.decode('ascii')
         self.assertTrue('target remote' in output, output)
         self.assertTrue('executable' in output, output)


 if __name__ == '__main__':
     unittest.main()
	#!/usr/bin/env python
	# Copyright 2023 The Pigweed Authors
	#
	# Licensed under the Apache License, Version 2.0 (the "License"); you may not
	# use this file except in compliance with the License. You may obtain a copy of
	# the License at
	#
	# https://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing, software
	# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
	# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
	# License for the specific language governing permissions and limitations under
	# the License.
	"""Tests for the command line interface"""

	import os
	import signal
	import subprocess
	import sys
	import time
	import unittest

	from pathlib import Path

	from mock_emu_frontend import _mock_emu
	from config_helper import ConfigHelper


	# TODO: b/301382004 - The Python Pigweed package install (into python-venv)
	# races with running this test and there is no way to add that package as a test
	# depedency without creating circular depedencies. This means we can't rely on
	# using Pigweed tools like pw cli or the arm-none-eabi-gdb wrapper.
	#
	# Run the CLI directly instead of going through pw cli.
	_cli_path = Path(
	os.path.join(os.environ['PW_ROOT'], 'pw_emu', 'py', 'pw_emu', '__main__.py')
	).resolve()


	class TestCli(ConfigHelper):
	"""Test non-interactive commands"""

	_config = {
	'emulators': {
	'mock-emu': {
	'launcher': 'mock_emu_frontend.MockEmuLauncher',
	'connector': 'mock_emu_frontend.MockEmuConnector',
	}
	},
	'mock-emu': {
	'tcp_channel': True,
	'gdb_channel': True,
	},
	'gdb': _mock_emu + ['--exit', '--'],
	'targets': {'test-target': {'mock-emu': {}}},
	}

	def _build_cmd(self, args: list[str]) -> list[str]:
	cmd = [
	'python',
	str(_cli_path),
	'--working-dir',
	self._wdir.name,
	'--config',
	self._config_file,
	] + args
	return cmd

	def _run(self, args: list[str], **kwargs) -> subprocess.CompletedProcess:
	"""Run the CLI and wait for completion"""
	return subprocess.run(self._build_cmd(args), **kwargs)

	def _popen(self, args: list[str], **kwargs) -> subprocess.Popen:
	"""Run the CLI in the background"""
	return subprocess.Popen(self._build_cmd(args), **kwargs)


	class TestNonInteractive(TestCli):
	"""Test non interactive commands."""

	def setUp(self) -> None:
	super().setUp()
	self.assertEqual(self._run(['start', 'test-target']).returncode, 0)

	def tearDown(self) -> None:
	self.assertEqual(self._run(['stop']).returncode, 0)
	super().tearDown()

	def test_already_running(self) -> None:
	self.assertNotEqual(self._run(['start', 'test-target']).returncode, 0)

	def test_gdb_cmds(self) -> None:
	status = self._run(
	['gdb-cmds', 'show version'],
	)
	self.assertEqual(status.returncode, 0)

	def test_prop_ls(self) -> None:
	status = self._run(['prop-ls', 'path1'], stdout=subprocess.PIPE)
	self.assertEqual(status.returncode, 0)
	self.assertTrue('prop1' in status.stdout.decode('ascii'))
	status = self._run(['prop-ls', 'invalid path'], stdout=subprocess.PIPE)
	self.assertNotEqual(status.returncode, 0)

	def test_prop_get(self) -> None:
	status = self._run(
	['prop-get', 'invalid path', 'prop1'],
	stdout=subprocess.PIPE,
	)
	self.assertNotEqual(status.returncode, 0)
	status = self._run(
	['prop-get', 'path1', 'invalid prop'],
	stdout=subprocess.PIPE,
	)
	self.assertNotEqual(status.returncode, 0)
	status = self._run(
	['prop-get', 'path1', 'prop1'],
	stdout=subprocess.PIPE,
	)
	self.assertEqual(status.returncode, 0)
	self.assertTrue('val1' in status.stdout.decode('ascii'))

	def test_prop_set(self) -> None:
	status = self._run(
	['prop-set', 'invalid path', 'prop1', 'v'],
	stdout=subprocess.PIPE,
	)
	self.assertNotEqual(status.returncode, 0)
	status = self._run(
	['prop-set', 'path1', 'invalid prop', 'v'],
	stdout=subprocess.PIPE,
	)
	self.assertNotEqual(status.returncode, 0)
	status = self._run(
	['prop-set', 'path1', 'prop1', 'value'],
	stdout=subprocess.PIPE,
	)
	self.assertEqual(status.returncode, 0)
	status = self._run(
	['prop-get', 'path1', 'prop1'],
	stdout=subprocess.PIPE,
	)
	self.assertEqual(status.returncode, 0)
	self.assertTrue('value' in status.stdout.decode('ascii'), status.stdout)

	def test_reset(self) -> None:
	self.assertEqual(self._run(['reset']).returncode, 0)
	self.assertTrue(os.path.exists(os.path.join(self._wdir.name, 'reset')))

	def test_load(self) -> None:
	self.assertEqual(self._run(['load', 'executable']).returncode, 0)

	def test_resume(self) -> None:
	self.assertEqual(self._run(['resume']).returncode, 0)


	class TestForeground(TestCli):
	"""Test starting in foreground"""

	def _test_common(self, cmd) -> None:
	# Run the CLI process in a new session so that we can terminate both the
	# CLI and the mock emulator it spawns in the foreground.
	args = {}
	if sys.platform == 'win32':
	args['creationflags'] = subprocess.CREATE_NEW_PROCESS_GROUP
	else:
	args['start_new_session'] = True
	proc = self._popen(cmd, stdout=subprocess.PIPE, **args)
	assert proc.stdout
	output = proc.stdout.readline()
	self.assertTrue(
	'starting mock emulator' in output.decode('utf-8'),
	output.decode('utf-8'),
	)
	if sys.platform == 'win32':
	# See https://bugs.python.org/issue26350
	os.kill(proc.pid, signal.CTRL_BREAK_EVENT)
	else:
	os.kill(-proc.pid, signal.SIGTERM)
	proc.wait()
	proc.stdout.close()

	def test_foreground(self) -> None:
	self._test_common(['start', '--foreground', 'test-target'])

	def test_debug(self) -> None:
	self._test_common(['start', '--debug', 'test-target'])


	class TestInteractive(TestCli):
	"""Test interactive commands"""

	def setUp(self) -> None:
	super().setUp()
	self.assertEqual(self._run(['start', 'test-target']).returncode, 0)

	def tearDown(self) -> None:
	self.assertEqual(self._run(['stop']).returncode, 0)
	super().tearDown()

	@staticmethod
	def _read_nonblocking(fd: int, size: int) -> bytes:
	try:
	return os.read(fd, size)
	except BlockingIOError:
	return b''

	def test_term(self) -> None:
	"""Test the pw emu term command"""

	if sys.platform == 'win32':
	self.skipTest('pty not supported on win32')

	# pylint: disable=import-outside-toplevel
	# Can't import pty on win32.
	import pty

	# pylint: disable=no-member
	# Avoid pylint false positive on win32.
	pid, fd = pty.fork()
	if pid == 0:
	status = self._run(['term', 'tcp'])
	# pylint: disable=protected-access
	# Use os._exit instead of os.exit after fork.
	os._exit(status.returncode)
	else:
	expected = '--- Miniterm on tcp ---'

	# Read the expected string with a timeout.
	os.set_blocking(fd, False)
	deadline = time.monotonic() + 5
	data = self._read_nonblocking(fd, len(expected))
	while len(data) < len(expected):
	time.sleep(0.1)
	data += self._read_nonblocking(fd, len(expected) - len(data))
	if time.monotonic() > deadline:
	break
	self.assertTrue(
	expected in data.decode('ascii'),
	data + self._read_nonblocking(fd, 100),
	)

	# send CTRL + ']' to terminate miniterm
	os.write(fd, b'\x1d')

	# wait for the process to exit, with a timeout
	deadline = time.monotonic() + 5
	wait_pid, ret = os.waitpid(pid, os.WNOHANG)
	while wait_pid == 0:
	time.sleep(0.1)
	# Discard input to avoid writer hang on MacOS,
	# see https://github.com/python/cpython/issues/97001.
	try:
	self._read_nonblocking(fd, 100)
	except OSError:
	# Avoid read errors when the child pair of the pty
	# closes when the child terminates.
	pass
	wait_pid, ret = os.waitpid(pid, os.WNOHANG)
	if time.monotonic() > deadline:
	break
	self.assertEqual(wait_pid, pid)
	self.assertEqual(ret, 0)

	def test_gdb(self) -> None:
	res = self._run(['gdb', '-e', 'executable'], stdout=subprocess.PIPE)
	self.assertEqual(res.returncode, 0)
	output = res.stdout.decode('ascii')
	self.assertTrue('target remote' in output, output)
	self.assertTrue('executable' in output, output)


	if __name__ == '__main__':
	unittest.main()