| #!/usr/bin/env python3 |
| # |
| # Copyright 2014 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Unit tests for sys_utils module.""" |
| |
| import os |
| import subprocess |
| import tempfile |
| import unittest |
| from unittest import mock |
| |
| from cros.factory.device import device_utils |
| from cros.factory.utils.process_utils import Spawn |
| from cros.factory.utils import sys_utils |
| |
| |
| SAMPLE_INTERRUPTS = """ CPU0 CPU1 CPU2 CPU3 |
| 0: 124 0 0 0 IO-APIC-edge timer |
| 1: 19 683 7 14 IO-APIC-edge i8042 |
| NMI: 612 631 661 401 Non-maskable interrupts |
| SPU: 0 0 0 0 Spurious interrupts""" |
| |
| |
| class GetInterruptsTest(unittest.TestCase): |
| """Unit tests for GetInterrupts.""" |
| |
| @mock.patch('cros.factory.utils.file_utils.ReadLines') |
| def testGetCount(self, read_lines_mock): |
| read_lines_mock.return_value = SAMPLE_INTERRUPTS.split('\n') |
| |
| ints = sys_utils.GetInterrupts() |
| # Digit-type interrupts: |
| self.assertEqual(ints['0'], 124) |
| self.assertEqual(ints['1'], 19 + 683 + 7 + 14) |
| |
| # String-type interrupts: |
| self.assertEqual(ints['NMI'], 612 + 631 + 661 + 401) |
| self.assertEqual(ints['SPU'], 0) |
| |
| read_lines_mock.assert_called_once_with('/proc/interrupts') |
| |
| @mock.patch('cros.factory.utils.file_utils.ReadLines') |
| def testFailToReadProcInterrupts(self, read_lines_mock): |
| read_lines_mock.return_value = None |
| |
| self.assertRaisesRegex( |
| OSError, r'Unable to read /proc/interrupts', |
| sys_utils.GetInterrupts) |
| |
| read_lines_mock.assert_called_once_with('/proc/interrupts') |
| |
| |
| SAMPLE_INPUT_DEVICES = """I: Bus=0019 Vendor=0000 Product=0005 Version=0000 |
| N: Name="Lid Switch" |
| P: Phys=PNP0C0D/button/input0 |
| S: Sysfs=/devices/LNXSYSTM:00/device:00/PNP0C0D:00/input/input0 |
| U: Uniq= |
| H: Handlers=lid_event_handler lid_event_handler event0 |
| B: PROP=0 |
| B: EV=21 |
| B: SW=1 |
| |
| I: Bus=0019 Vendor=0000 Product=0001 Version=0000 |
| N: Name="Power Button" |
| P: Phys=PNP0C0C/button/input0 |
| S: Sysfs=/devices/LNXSYSTM:00/device:00/PNP0C0C:00/input/input1 |
| U: Uniq= |
| H: Handlers=kbd event1 |
| B: PROP=0 |
| B: EV=3 |
| B: KEY=10000000000000 0 |
| |
| I: Bus=0019 Vendor=0000 Product=0001 Version=0000 |
| N: Name="Power Button" |
| P: Phys=LNXPWRBN/button/input0 |
| S: Sysfs=/devices/LNXSYSTM:00/LNXPWRBN:00/input/input2 |
| U: Uniq= |
| H: Handlers=kbd event2 |
| B: PROP=0 |
| B: EV=3 |
| B: KEY=10000000000000 0 |
| |
| I: Bus=0011 Vendor=0001 Product=0001 Version=ab83 |
| N: Name="AT Translated Set 2 keyboard" |
| P: Phys=isa0060/serio0/input0 |
| S: Sysfs=/devices/platform/i8042/serio0/input/input3 |
| U: Uniq= |
| H: Handlers=sysrq kbd event3 |
| B: PROP=0 |
| B: EV=120013 |
| B: KEY=400402000000 3803078f800d001 feffffdfffefffff fffffffffffffffe |
| B: MSC=10 |
| B: LED=7 |
| |
| I: Bus=0018 Vendor=0000 Product=0000 Version=0000 |
| N: Name="Atmel maXTouch Touchscreen" |
| P: Phys=i2c-3-004a/input0 |
| S: Sysfs=/devices/platform/80860F41:03/i2c-3/3-004a/input/input4 |
| U: Uniq= |
| H: Handlers=event4 |
| B: PROP=0 |
| B: EV=b |
| B: KEY=400 0 0 0 0 0 |
| B: ABS=661800001000003 |
| |
| I: Bus=0018 Vendor=0000 Product=0000 Version=0000 |
| N: Name="Atmel maXTouch Touchpad" |
| P: Phys=i2c-0-004b/input0 |
| S: Sysfs=/devices/platform/80860F41:00/i2c-0/0-004b/input/input5 |
| U: Uniq= |
| H: Handlers=event5 |
| B: PROP=5 |
| B: EV=b |
| B: KEY=e520 10000 0 0 0 0 |
| B: ABS=661800001000003 |
| |
| I: Bus=0000 Vendor=0000 Product=0000 Version=0000 |
| N: Name="HDA Intel HDMI" |
| P: Phys=ALSA |
| S: Sysfs=/devices/pci0000:00/0000:00:1b.0/sound/card1/input6 |
| U: Uniq= |
| H: Handlers=event6 |
| B: PROP=0 |
| B: EV=21 |
| B: SW=140 |
| |
| I: Bus=0000 Vendor=0000 Product=0000 Version=0000 |
| N: Name="HDA Intel HDMI" |
| P: Phys=ALSA |
| S: Sysfs=/devices/pci0000:00/0000:00:1b.0/sound/card1/input7 |
| U: Uniq= |
| H: Handlers=event7 |
| B: PROP=0 |
| B: EV=21 |
| B: SW=140 |
| """ |
| |
| |
| class GetI2CBusTest(unittest.TestCase): |
| """Unit tests for GetI2CBus.""" |
| |
| @mock.patch('cros.factory.utils.file_utils.ReadFile') |
| def testTouchpad(self, read_file_mock): |
| read_file_mock.return_value = SAMPLE_INPUT_DEVICES |
| self.assertEqual(sys_utils.GetI2CBus(['Dummy device name', |
| 'Atmel maXTouch Touchpad']), 0) |
| read_file_mock.assert_called_once_with('/proc/bus/input/devices') |
| |
| @mock.patch('cros.factory.utils.file_utils.ReadFile') |
| def testTouchscreen(self, read_file_mock): |
| read_file_mock.return_value = SAMPLE_INPUT_DEVICES |
| self.assertEqual(sys_utils.GetI2CBus(['Atmel maXTouch Touchscreen']), 3) |
| read_file_mock.assert_called_once_with('/proc/bus/input/devices') |
| |
| @mock.patch('cros.factory.utils.file_utils.ReadFile') |
| def testNoMatch(self, read_file_mock): |
| read_file_mock.return_value = SAMPLE_INPUT_DEVICES |
| self.assertEqual(sys_utils.GetI2CBus(['Unknown Device']), None) |
| read_file_mock.assert_called_once_with('/proc/bus/input/devices') |
| |
| @mock.patch('cros.factory.utils.file_utils.ReadFile') |
| def testNonI2CDevice(self, read_file_mock): |
| read_file_mock.return_value = SAMPLE_INPUT_DEVICES |
| self.assertEqual(sys_utils.GetI2CBus(['Lid Switch']), None) |
| read_file_mock.assert_called_once_with('/proc/bus/input/devices') |
| |
| |
| SAMPLE_PARTITIONS = """major minor #blocks name |
| |
| 7 0 313564 loop0 |
| 179 0 15388672 mmcblk0 |
| 179 1 11036672 mmcblk0p1 |
| 179 2 16384 mmcblk0p2 |
| 179 3 2097152 mmcblk0p3 |
| 179 4 16384 mmcblk0p4 |
| 179 5 2097152 mmcblk0p5 |
| 179 6 0 mmcblk0p6 |
| 179 7 0 mmcblk0p7 |
| 179 8 16384 mmcblk0p8 |
| 179 9 0 mmcblk0p9 |
| 179 10 0 mmcblk0p10 |
| 179 11 8192 mmcblk0p11 |
| 179 12 16384 mmcblk0p12 |
| 179 32 4096 mmcblk0boot1 |
| 179 16 4096 mmcblk0boot0 |
| 254 0 1048576 dm-0 |
| 254 1 313564 dm-1 |
| 253 0 1953128 zram0""" |
| |
| |
| class GetPartitionsTest(unittest.TestCase): |
| """Unit tests for GetInterrupts.""" |
| |
| @mock.patch('cros.factory.utils.file_utils.ReadLines') |
| def testGetPartitions(self, read_lines_mock): |
| read_lines_mock.return_value = SAMPLE_PARTITIONS.split('\n') |
| |
| partitions = sys_utils.GetPartitions() |
| for p in partitions: |
| print(str(p)) |
| |
| read_lines_mock.assert_called_once_with('/proc/partitions') |
| |
| |
| class MountDeviceAndReadFileTest(unittest.TestCase): |
| """Unittest for MountDeviceAndReadFile.""" |
| |
| def setUp(self): |
| # Creates a temp file and create file system on it as a mock device. |
| self.device = tempfile.NamedTemporaryFile(prefix='MountDeviceAndReadFile') |
| Spawn(['truncate', '-s', '1M', self.device.name], log=True, |
| check_call=True) |
| |
| # In CrOS chroot, mkfs and mkfs.extX may live in different locations that |
| # normal user can't run without adding /sbin and /usr/sbin. |
| env = os.environ.copy() |
| env['PATH'] = '/sbin:/usr/sbin:' + env['PATH'] |
| Spawn(['/sbin/mkfs', '-E', 'root_owner=%d:%d' % (os.getuid(), os.getgid()), |
| '-F', '-t', 'ext3', self.device.name], log=True, check_call=True, |
| env=env) |
| |
| # Creates a file with some content on the device. |
| mount_point = tempfile.mkdtemp(prefix='MountDeviceAndReadFileSetup') |
| Spawn(['mount', self.device.name, mount_point], sudo=True, check_call=True, |
| log=True) |
| self.content = 'file content' |
| self.file_name = 'file' |
| with open(os.path.join(mount_point, self.file_name), 'w') as f: |
| f.write(self.content) |
| Spawn(['umount', '-l', mount_point], sudo=True, check_call=True, log=True) |
| os.rmdir(mount_point) |
| self.dut = device_utils.CreateDUTInterface() |
| |
| # Since 'mount', 'umount' requires root privilege, make link.Shell function |
| # executes commands as root. |
| # The reason why we don't use 'sudo' in MountPartition() is to make our code |
| # more general. Because some Android devices don't have 'sudo'. |
| def _SudoShell(command, stdin=None, stdout=None, stderr=None, cwd=None, |
| encoding='utf-8'): |
| if isinstance(command, str): |
| command = ['sudo', 'sh', '-c', command] |
| else: |
| command = ['sudo'] + command |
| return subprocess.Popen(command, cwd=cwd, close_fds=True, stdin=stdin, |
| stdout=stdout, stderr=stderr, encoding=encoding) |
| self.dut.link.Shell = _SudoShell |
| |
| def tearDown(self): |
| self.device.close() |
| |
| def testMountDeviceAndReadFile(self): |
| self.assertEqual( |
| self.content, |
| sys_utils.MountDeviceAndReadFile( |
| self.device.name, self.file_name)) |
| |
| def testMountDeviceAndReadFileWrongFile(self): |
| with self.assertRaises(IOError): |
| sys_utils.MountDeviceAndReadFile(self.device.name, 'no_file') |
| |
| def testMountDeviceAndReadFileWrongDevice(self): |
| with self.assertRaises(Exception): |
| sys_utils.MountDeviceAndReadFile('no_device', self.file_name) |
| |
| def testMountDeviceAndReadFileWithDUT(self): |
| self.assertEqual( |
| self.content, |
| sys_utils.MountDeviceAndReadFile( |
| self.device.name, self.file_name, self.dut)) |
| |
| def testMountDeviceAndReadFileWrongFileWithDUT(self): |
| with self.assertRaises(IOError): |
| sys_utils.MountDeviceAndReadFile(self.device.name, 'no_file', self.dut) |
| |
| def testMountDeviceAndReadFileWrongDeviceWithDUT(self): |
| with self.assertRaises(Exception): |
| sys_utils.MountDeviceAndReadFile('no_device', self.file_name, self.dut) |
| |
| |
| class TestLogMessagesTest(unittest.TestCase): |
| |
| def testGetVarLogMessages(self): |
| with tempfile.NamedTemporaryFile('w', encoding='utf-8') as f: |
| data = ("Captain's log.\xFF\n" # \xFF = invalid UTF-8 |
| 'We are in pursuit of a starship of Ferengi design.\n') |
| f.write(('X' * 100) + '\n' + data) |
| f.flush() |
| # Use max_length=len(data) + 5 so that we'll end up reading |
| # (and discarding) the last 5 bytes of garbage X's. |
| self.assertEqual( |
| '<truncated 101 bytes>\n' |
| "Captain's log.\xFF\n" |
| 'We are in pursuit of a starship of Ferengi design.\n', |
| sys_utils.GetVarLogMessages(max_length=(len(data) + 5), path=f.name)) |
| |
| dut = device_utils.CreateDUTInterface(board_class='LinuxBoard') |
| self.assertEqual( |
| '<truncated 101 bytes>\n' |
| "Captain's log.\xFF\n" |
| 'We are in pursuit of a starship of Ferengi design.\n', |
| sys_utils.GetVarLogMessages( |
| max_length=(len(data) + 5), path=f.name, dut=dut)) |
| |
| def testGetVarLogMessagesBeforeReboot(self): |
| EARLIER_VAR_LOG_MESSAGES = ( |
| "19:26:17 kernel: That's all, folks.\n" |
| "19:26:56 kernel: [ 0.000000] Initializing cgroup subsys cpuset\n" |
| "19:26:56 kernel: [ 0.000000] Initializing cgroup subsys cpu\n" |
| "19:26:56 kernel: [ 0.000000] Linux version blahblahblah\n") |
| |
| VAR_LOG_MESSAGES = ( |
| "19:00:00 kernel: 7 p.m. and all's well.\n" |
| "19:27:17 kernel: That's all, folks.\n" |
| "19:27:17 kernel: Kernel logging (proc) stopped.\n" |
| "19:27:56 kernel: imklog 4.6.2, log source = /proc/kmsg started.\n" |
| "19:27:56 rsyslogd: " |
| '[origin software="rsyslogd" blahblahblah] (re)start\n' |
| "19:27:56 kernel: [ 0.000000] Initializing cgroup subsys cpuset\n" |
| "19:27:56 kernel: [ 0.000000] Initializing cgroup subsys cpu\n" |
| "19:27:56 kernel: [ 0.000000] Linux version blahblahblah\n" |
| "19:27:56 kernel: [ 0.000000] Command line: blahblahblah\n") |
| |
| dut = device_utils.CreateDUTInterface(board_class='LinuxBoard') |
| |
| with tempfile.NamedTemporaryFile('w') as f: |
| f.write(VAR_LOG_MESSAGES) |
| f.flush() |
| |
| self.assertEqual( |
| ("19:27:17 kernel: That's all, folks.\n" |
| "19:27:17 kernel: Kernel logging (proc) stopped.\n" |
| "<after reboot, kernel came up at 19:27:56>\n"), |
| sys_utils.GetVarLogMessagesBeforeReboot( |
| path=f.name, lines=2, dut=dut)) |
| |
| self.assertEqual( |
| ("19:27:17 kernel: That's all, folks.\n" |
| "19:27:17 kernel: Kernel logging (proc) stopped.\n" |
| "<after reboot, kernel came up at 19:27:56>\n"), |
| sys_utils.GetVarLogMessagesBeforeReboot(path=f.name, lines=2)) |
| |
| self.assertEqual( |
| ("19:27:17 kernel: Kernel logging (proc) stopped.\n" |
| "<after reboot, kernel came up at 19:27:56>\n"), |
| sys_utils.GetVarLogMessagesBeforeReboot(path=f.name, lines=1)) |
| |
| self.assertEqual( |
| ("19:00:00 kernel: 7 p.m. and all's well.\n" |
| "19:27:17 kernel: That's all, folks.\n" |
| "19:27:17 kernel: Kernel logging (proc) stopped.\n" |
| "<after reboot, kernel came up at 19:27:56>\n"), |
| sys_utils.GetVarLogMessagesBeforeReboot(path=f.name, lines=100)) |
| |
| with tempfile.NamedTemporaryFile('w') as f: |
| f.write(EARLIER_VAR_LOG_MESSAGES) |
| f.flush() |
| self.assertEqual( |
| ("19:26:17 kernel: That's all, folks.\n" |
| "<after reboot, kernel came up at 19:26:56>\n"), |
| sys_utils.GetVarLogMessagesBeforeReboot(path=f.name, lines=1)) |
| |
| class TestGetRunningFactoryPythonArchivePath(unittest.TestCase): |
| |
| def setUp(self): |
| self.path_exists_mapping = {} |
| self.original_path_exists_func = os.path.exists |
| |
| def path_exists_mock_side_effect(self, *args, **unused_kwargs): |
| if args[0] in self.path_exists_mapping: |
| return self.path_exists_mapping[args[0]] |
| return self.original_path_exists_func(args[0]) |
| |
| def testNotInPythonArchive(self): |
| sys_utils.__file__ = '/path/to/factory/utils/sys_utils.py' |
| self.path_exists_mapping = { |
| sys_utils.__file__: True |
| } |
| |
| with mock.patch('os.path.exists') as path_exists_mock: |
| path_exists_mock.side_effect = self.path_exists_mock_side_effect |
| self.assertEqual(sys_utils.GetRunningFactoryPythonArchivePath(), None) |
| |
| def testInPythonFactoryArchive(self): |
| factory_par = '/path/to/factory.par' |
| sys_utils.__file__ = '/path/to/factory.par/cros/factory/utils/sys_utils.py' |
| self.path_exists_mapping = { |
| sys_utils.__file__: False, |
| factory_par: True |
| } |
| |
| with mock.patch('os.path.exists') as path_exists_mock: |
| path_exists_mock.side_effect = self.path_exists_mock_side_effect |
| self.assertEqual(sys_utils.GetRunningFactoryPythonArchivePath(), |
| factory_par) |
| |
| def testNonExistingFileWithoutCrosFactoryPrefix(self): |
| sys_utils.__file__ = '/path/to/nowhere/utils/sys_utils.py' |
| self.path_exists_mapping = { |
| sys_utils.__file__: False |
| } |
| |
| with mock.patch('os.path.exists') as path_exists_mock: |
| path_exists_mock.side_effect = self.path_exists_mock_side_effect |
| self.assertEqual(sys_utils.GetRunningFactoryPythonArchivePath(), None) |
| |
| def testNonExistingFactoryPythonArchive(self): |
| factory_par = '/path/to/factory.par' |
| sys_utils.__file__ = '/path/to/factory.par/cros/factory/utils/sys_utils.py' |
| self.path_exists_mapping = { |
| sys_utils.__file__: False, |
| factory_par: False |
| } |
| |
| with mock.patch('os.path.exists') as path_exists_mock: |
| path_exists_mock.side_effect = self.path_exists_mock_side_effect |
| self.assertEqual(sys_utils.GetRunningFactoryPythonArchivePath(), None) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
