# Copyright 2025 The Chromium Authors # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. from __future__ import annotations import datetime as dt import textwrap from unittest import mock from pyfakefs.fake_filesystem import OSType from typing_extensions import override from crossbench.helper.version import VersionParseError from crossbench.plt.linux import (SCRIPTS_DIR, LinuxPlatform, parse_display_xrandr) from crossbench.plt.posix import PosixVersion from crossbench.plt.process_meminfo import ProcessMeminfo from tests import test_helper from tests.crossbench.mock_helper import (LinuxMockPlatform, RemoteLinuxMockPlatform, ShResult) from tests.crossbench.plt.helper import (BaseLocalMockPlatformTestMixin, BasePosixMockPlatformTestCase) NOW_EPOCH = dt.datetime.now() class _LinuxMockPlatformTestCase(BasePosixMockPlatformTestCase): @override def setUp(self) -> None: super().setUp() self.fs.os = OSType.LINUX @override def mock_platform_setup(self) -> None: self.mock_platform = LinuxMockPlatform() self.platform = self.mock_platform def test_name(self): self.assertEqual(self.platform.name, "mock.linux") def test_is_linux(self): self.assertTrue(self.platform.is_linux) def test_version(self): self.mock_platform.mock_version_str = None self.expect_sh("uname", "-r", result="5.4.0-104-generic") self.assertEqual(self.platform.version_str, "5.4.0-104-generic") version = self.platform.version self.assertEqual(version.parts, (5, 4, 0)) self.assertEqual(version.version_str, "5.4.0-104-generic") @mock.patch("psutil.cpu_count") def test_cpu_cores(self, mock_cpu_count): mock_cpu_count.return_value = 12 self.assertEqual(self.platform.cpu_cores(logical=True), 12) mock_cpu_count.assert_called_once() mock_cpu_count.return_value = 6 self.assertEqual(self.platform.cpu_cores(logical=False), 6) self.assertEqual(mock_cpu_count.call_count, 2) def test_parse_display_xrandr(self): xrandr_output = textwrap.dedent(""" Screen 0: minimum 64 x 64, current 1728 x 946, maximum 32767 x 32767 DUMMY0 connected primary 1728x946+0+0 456mm x 249mm 1024x768 60.00 800x600 60.32 56.25 640x480 59.94 1600x1200_60 60.00 1600x1200_120 120.00 CRD_78 120.00*+ DUMMY1 disconnected 5120x1440_120 120.00 2160x3840_120 120.00 """) parsed = tuple(parse_display_xrandr(xrandr_output)) self.assertEqual(len(parsed), 1) self.assertDictEqual(parsed[0], { "resolution": (1728, 946), "refresh_rate": 120.0 }) def test_meminfo_no_proc(self): path = SCRIPTS_DIR / "meminfo.sh" self.fs.create_file(path, contents="meminfo") self.mock_platform.sh_results = [ShResult()] meminfo = LinuxPlatform.process_meminfo(self.mock_platform, "some_process") self.assertEqual(len(meminfo), 0) def test_platform_version_cls(self): version = PosixVersion.parse("5.4.0-104-generic") self.assertEqual(version.parts, (5, 4, 0)) self.assertEqual(version.version_str, "5.4.0-104-generic") with self.assertRaises(VersionParseError): PosixVersion.parse("foo") _MEMINFO_SCRIPT_OUTPUT = """ ==== process 926961 ==== /usr/bin/some_process -a ==== smaps_rollup ==== 00008000-7ffd0f5fe000 ---p 00000000 00:00 0 [rollup] Rss: 80364 kB Pss: 17815 kB Pss_Dirty: 8715 kB Swap: 0 kB SwapPss: 0 kB ==== process 930293 ==== /usr/bin/some_process -b ==== smaps_rollup ==== 85800000000-7ffd0f5fe000 ---p 00000000 00:00 0 [rollup] Rss: 44860 kB Pss: 5074 kB Private_Hugetlb: 0 kB Swap: 0 kB Locked: 0 kB ==== process 930304 ==== /usr/bin/some_process -c ==== smaps_rollup ==== Rss: 2680 kB Pss: 521 kB Swap: 400 kB """ def test_meminfo(self): path = SCRIPTS_DIR / "meminfo.sh" self.fs.create_file(path, contents="meminfo") self.mock_platform.sh_results = [ShResult(self._MEMINFO_SCRIPT_OUTPUT)] meminfo = LinuxPlatform.process_meminfo(self.mock_platform, "some_process") self.assertListEqual(meminfo, [ ProcessMeminfo(926961, "/usr/bin/some_process -a", 17815, 80364, 0), ProcessMeminfo(930293, "/usr/bin/some_process -b", 5074, 44860, 0), ProcessMeminfo(930304, "/usr/bin/some_process -c", 521, 2680, 400), ]) class LocalLinuxMockPlatformTestCase(BaseLocalMockPlatformTestMixin, _LinuxMockPlatformTestCase): __test__ = True class RemoteLinuxMockPlatformTestCase(_LinuxMockPlatformTestCase): @override def mock_platform_setup(self) -> None: self.host_platform = LinuxMockPlatform() self.mock_platform = RemoteLinuxMockPlatform(self.host_platform) self.platform = self.mock_platform def cpu_info(self, processor_id, physical_id, core_id): return textwrap.dedent(f""" processor : {processor_id} vendor_id : GenuineIntel cpu family : 1 model : 12 model name : Intel(R) Xeon(R) 3456 CPU @ 7.80GHz stepping : 9 microcode : 0x123456 cpu MHz : 1234.000 cache size : 3456 KB physical id : {physical_id} core id : {core_id} fpu : yes fpu_exception : yes cpuid level : 12 wp : yes """) def expect_sh_cpu_info(self, cpu_info): self.expect_sh( "grep", "-E", "processor|core id|physical id", "/proc/cpuinfo", result=cpu_info) @override def test_cpu_cores(self): single_core_info = self.cpu_info(0, 0, 0) self.expect_sh_cpu_info(single_core_info) self.assertEqual(self.platform.cpu_cores(logical=True), 1) self.assertFalse(self.platform.expected_sh_cmds) # Check that caching works. self.assertEqual(self.platform.cpu_cores(logical=True), 1) self.expect_sh_cpu_info(single_core_info) self.assertEqual(self.platform.cpu_cores(logical=False), 1) self.assertFalse(self.platform.expected_sh_cmds) self.assertEqual(self.platform.cpu_cores(logical=False), 1) def test_cpu_cores_2_cpu_single_core(self): # 2 physical chips, 1 core, 2 threads dual_chip_result = ( self.cpu_info(0, 0, 0) + self.cpu_info(1, 0, 0) + self.cpu_info(2, 1, 0) + self.cpu_info(3, 1, 0)) self.expect_sh_cpu_info(dual_chip_result) self.assertEqual(self.platform.cpu_cores(logical=True), 4) self.assertFalse(self.platform.expected_sh_cmds) self.assertEqual(self.platform.cpu_cores(logical=True), 4) self.expect_sh_cpu_info(dual_chip_result) self.assertEqual(self.platform.cpu_cores(logical=False), 2) self.assertFalse(self.platform.expected_sh_cmds) self.assertEqual(self.platform.cpu_cores(logical=False), 2) def test_cpu_cores_1_cpu_dual_core(self): # 1 physical chips, 2 cores, 2 threads dual_core_result = ( self.cpu_info(0, 0, 0) + self.cpu_info(1, 0, 1) + self.cpu_info(2, 0, 0) + self.cpu_info(3, 0, 1)) self.expect_sh_cpu_info(dual_core_result) self.assertEqual(self.platform.cpu_cores(logical=True), 4) self.assertFalse(self.platform.expected_sh_cmds) self.assertEqual(self.platform.cpu_cores(logical=True), 4) self.expect_sh_cpu_info(dual_core_result) self.assertEqual(self.platform.cpu_cores(logical=False), 2) self.assertFalse(self.platform.expected_sh_cmds) self.assertEqual(self.platform.cpu_cores(logical=False), 2) # TODO: implement more mock tests def test_local_reverse_port_forward_invalid(self): pass def test_local_reverse_port_forward(self): pass def test_local_port_forward(self): pass del _LinuxMockPlatformTestCase if __name__ == "__main__": test_helper.run_pytest(__file__)