path: root/pw_emu/py/tests/qemu_test.py

#!/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.
"""QEMU emulator tests."""

import json
import os
import socket
import sys
import tempfile
import time
import unittest

from pathlib import Path
from typing import Any, Dict, Optional

from pw_emu.core import InvalidChannelName, InvalidChannelType
from tests.common import check_prog, ConfigHelperWithEmulator


# 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 arm_gdb.py wrapper directly
_arm_none_eabi_gdb_path = Path(
    os.path.join(
        os.environ['PW_ROOT'],
        'pw_env_setup',
        'py',
        'pw_env_setup',
        'entry_points',
        'arm_gdb.py',
    )
).resolve()


class TestQemu(ConfigHelperWithEmulator):
    """Tests for a valid qemu configuration."""

    _config = {
        'gdb': ['python', str(_arm_none_eabi_gdb_path)],
        'qemu': {
            'executable': 'qemu-system-arm',
        },
        'targets': {
            'test-target': {
                'ignore1': None,
                'qemu': {
                    'machine': 'lm3s6965evb',
                    'channels': {
                        'chardevs': {
                            'test_uart': {
                                'id': 'serial0',
                            }
                        }
                    },
                },
                'ignore2': None,
            }
        },
    }

    def setUp(self) -> None:
        super().setUp()
        # No image so start paused to avoid crashing.
        self._emu.start(target='test-target', pause=True)

    def tearDown(self) -> None:
        self._emu.stop()
        super().tearDown()

    def test_running(self) -> None:
        self.assertTrue(self._emu.running())

    def test_list_properties(self) -> None:
        self.assertIsNotNone(self._emu.list_properties('/machine'))

    def test_get_property(self) -> None:
        self.assertEqual(
            self._emu.get_property('/machine', 'type'), 'lm3s6965evb-machine'
        )

    def test_set_property(self) -> None:
        self._emu.set_property('/machine', 'graphics', False)
        self.assertFalse(self._emu.get_property('/machine', 'graphics'))

    def test_bad_channel_name(self) -> None:
        with self.assertRaises(InvalidChannelName):
            self._emu.get_channel_addr('serial1')

    def get_reg(self, addr: int) -> bytes:
        temp = tempfile.NamedTemporaryFile(delete=False)
        temp.close()

        res = self._emu.run_gdb_cmds(
            [
                f'dump val {temp.name} *(char*){addr}',
                'disconnect',
            ]
        )
        self.assertEqual(res.returncode, 0, res.stderr.decode('ascii'))

        with open(temp.name, 'rb') as file:
            ret = file.read(1)

        self.assertNotEqual(ret, b'', res.stderr.decode('ascii'))

        os.unlink(temp.name)

        return ret

    def poll_data(self, timeout: int) -> Optional[bytes]:
        uartris = 0x4000C03C
        uartrd = 0x4000C000

        deadline = time.monotonic() + timeout
        while self.get_reg(uartris) == b'\x00':
            time.sleep(0.1)
            if time.monotonic() > deadline:
                return None
        return self.get_reg(uartrd)

    def test_channel_stream(self) -> None:
        ok, msg = check_prog('arm-none-eabi-gdb')
        if not ok:
            self.skipTest(msg)

        stream = self._emu.get_channel_stream('test_uart')
        stream.write('test\n'.encode('ascii'))

        self.assertEqual(self.poll_data(5), b't')
        self.assertEqual(self.poll_data(5), b'e')
        self.assertEqual(self.poll_data(5), b's')
        self.assertEqual(self.poll_data(5), b't')

    def test_gdb(self) -> None:
        self._emu.run_gdb_cmds(['c'])
        deadline = time.monotonic() + 5
        while self._emu.running():
            if time.monotonic() > deadline:
                return
        self.assertFalse(self._emu.running())


class TestQemuChannelsTcp(TestQemu):
    """Tests for configurations using TCP channels."""

    _config: Dict[str, Any] = {}
    _config.update(json.loads(json.dumps(TestQemu._config)))
    _config['qemu']['channels'] = {'type': 'tcp'}

    def test_get_channel_addr(self) -> None:
        host, port = self._emu.get_channel_addr('test_uart')
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.connect((host, port))
        sock.close()


class TestQemuChannelsPty(TestQemu):
    """Tests for configurations using PTY channels."""

    _config: Dict[str, Any] = {}
    _config.update(json.loads(json.dumps(TestQemu._config)))
    _config['qemu']['channels'] = {'type': 'pty'}

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

    def test_get_path(self) -> None:
        self.assertTrue(os.path.exists(self._emu.get_channel_path('test_uart')))


class TestQemuInvalidChannelType(ConfigHelperWithEmulator):
    """Test invalid channel type configuration."""

    _config = {
        'qemu': {
            'executable': 'qemu-system-arm',
            'channels': {'type': 'invalid'},
        },
        'targets': {
            'test-target': {
                'qemu': {
                    'machine': 'lm3s6965evb',
                }
            }
        },
    }

    def test_start(self) -> None:
        with self.assertRaises(InvalidChannelType):
            self._emu.start('test-target', pause=True)


class TestQemuTargetChannelsMixed(ConfigHelperWithEmulator):
    """Test configuration with mixed channels types."""

    _config = {
        'qemu': {
            'executable': 'qemu-system-arm',
        },
        'targets': {
            'test-target': {
                'qemu': {
                    'machine': 'lm3s6965evb',
                    'channels': {
                        'chardevs': {
                            'test_uart0': {
                                'id': 'serial0',
                            },
                            'test_uart1': {
                                'id': 'serial1',
                                'type': 'tcp',
                            },
                            'test_uart2': {
                                'id': 'serial2',
                                'type': 'pty',
                            },
                        }
                    },
                }
            }
        },
    }

    def setUp(self) -> None:
        if sys.platform == 'win32':
            self.skipTest('pty not supported on win32')
        super().setUp()
        # no image to run so start paused
        self._emu.start('test-target', pause=True)

    def tearDown(self) -> None:
        self._emu.stop()
        super().tearDown()

    def test_uart0_addr(self) -> None:
        host, port = self._emu.get_channel_addr('test_uart0')
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.connect((host, port))
        sock.close()

    def test_uart1_addr(self) -> None:
        host, port = self._emu.get_channel_addr('test_uart1')
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.connect((host, port))
        sock.close()

    def test_uart2_path(self) -> None:
        self.assertTrue(
            os.path.exists(self._emu.get_channel_path('test_uart2'))
        )


def main() -> None:
    ok, msg = check_prog('qemu-system-arm')
    if not ok:
        print(f'skipping tests: {msg}')
        sys.exit(0)

    unittest.main()


if __name__ == '__main__':
    main()