| # Copyright 2024 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 contextlib |
| import os |
| from unittest import mock, skipIf |
| |
| import pyfakefs |
| from typing_extensions import override |
| |
| from crossbench import path as pth |
| from crossbench import plt |
| from crossbench.plt.port_manager import PortForwardException |
| from tests import test_helper |
| from tests.crossbench.plt.helper import BasePosixMockPlatformTestCase |
| |
| |
| class LinuxSshMockPlatformTestCase(BasePosixMockPlatformTestCase): |
| __test__ = True |
| HOST = "host" |
| PORT = 9515 |
| SSH_PORT = 22 |
| SSH_USER = "user" |
| platform: plt.LinuxSshPlatform |
| |
| @override |
| def setUp(self) -> None: |
| super().setUp() |
| self.host_platform = self.mock_platform |
| self.platform = plt.LinuxSshPlatform( |
| self.mock_platform, |
| host=self.HOST, |
| port=self.PORT, |
| ssh_port=self.SSH_PORT, |
| ssh_user=self.SSH_USER) |
| self.mock_platform_str(self.platform, "linux_ssh_mock_platform") |
| |
| def _expect_sh_ssh(self, *args, result=""): |
| self.mock_platform.expect_sh( |
| "ssh", |
| "-p", |
| str(self.SSH_PORT), |
| f"{self.SSH_USER}@{self.HOST}", |
| *args, |
| result=result) |
| |
| def _expect_sh_ssh_shell(self, *args, result=""): |
| cmd_string = f"ssh -p {str(self.SSH_PORT)} {self.SSH_USER}@{self.HOST} " |
| |
| for arg in args: |
| cmd_string += arg + " " |
| |
| self.mock_platform.expect_sh(cmd_string, result=result) |
| |
| def expect_sh(self, *args, result="") -> None: |
| self._expect_sh_ssh(*args, result=result) |
| |
| def test_is_linux(self): |
| self.assertTrue(self.platform.is_linux) |
| |
| def test_is_remote_ssh(self): |
| self.assertTrue(self.platform.is_remote_ssh) |
| |
| def test_basic_properties(self): |
| self.assertTrue(self.platform.is_remote) |
| self.assertEqual(self.platform.host, self.HOST) |
| self.assertEqual(self.platform.port, self.PORT) |
| self.assertIs(self.platform.host_platform, self.mock_platform) |
| self.assertTrue(self.platform.is_posix) |
| |
| def test_name(self): |
| self.assertEqual(self.platform.name, "linux_ssh") |
| |
| def test_version(self): |
| self._expect_sh_ssh("uname -r", result="999") |
| self.assertEqual(self.platform.version_str, "999") |
| # Subsequent calls are cached. |
| self.assertEqual(self.platform.version_str, "999") |
| |
| @skipIf( |
| tuple(map(int, pyfakefs.__version__.split("."))) < (5, 5), |
| "pth.AnyWindowsPath does not work correctly with older pyfakefs") |
| def test_iterdir(self): |
| self._expect_sh_ssh("'[' -d parent_dir/child_dir ']'") |
| self._expect_sh_ssh("ls -1 parent_dir/child_dir", result="file1\nfile2\n") |
| |
| self.assertSetEqual( |
| set(self.platform.iterdir(pth.AnyWindowsPath("parent_dir\\child_dir"))), |
| { |
| pth.AnyPosixPath("parent_dir/child_dir/file1"), |
| pth.AnyPosixPath("parent_dir/child_dir/file2") |
| }) |
| |
| def test_cat_file(self): |
| self._expect_sh_ssh("cat path/to/a/file") |
| self.platform.cat(self.platform.path("path/to/a/file")) |
| self._expect_sh_ssh("cat 'path/with a space/to/a/file'") |
| self.platform.cat(self.platform.path("path/with a space/to/a/file")) |
| |
| def test_sh_shell_invalid(self): |
| with self.assertRaisesRegex(ValueError, "shell=True"): |
| self.platform.sh_stdout("ls", "folder with space", shell=True) |
| |
| def test_sh_shell(self): |
| self._expect_sh_ssh("ls sdcard", result="FILE1\nFILE2\n") |
| self.assertEqual(self.platform.sh_stdout("ls", "sdcard"), "FILE1\nFILE2\n") |
| |
| self._expect_sh_ssh("ls 'folder with space'", result="FOLDER\n") |
| self.assertEqual( |
| self.platform.sh_stdout("ls", "folder with space"), "FOLDER\n") |
| |
| self._expect_sh_ssh("'ls foo && ls bar'", result="FILE1\nFILE2\n") |
| self.assertEqual( |
| self.platform.sh_stdout("ls foo && ls bar"), "FILE1\nFILE2\n") |
| |
| self._expect_sh_ssh_shell("'ls foo && ls bar'", result="FILE1\nFILE2\n") |
| self.assertEqual( |
| self.platform.sh_stdout("ls foo && ls bar", shell=True), |
| "FILE1\nFILE2\n") |
| |
| self._expect_sh_ssh("ls foo '&&' ls bar", result="FILE1\nFILE2\n") |
| self.assertEqual( |
| self.platform.sh_stdout("ls", "foo", "&&", "ls", "bar"), |
| "FILE1\nFILE2\n") |
| |
| @contextlib.contextmanager |
| def mock_popen(self, platform): |
| with mock.patch.object(type(platform), "popen") as patcher: |
| yield patcher |
| |
| @contextlib.contextmanager |
| def mock_get_free_port(self, platform, port): |
| with mock.patch.object( |
| type(platform), "get_free_port", return_value=port) as patcher: |
| yield patcher |
| |
| @contextlib.contextmanager |
| def mock_wait_for_port(self, platform): |
| with mock.patch.object(type(platform), "wait_for_port") as patcher: |
| yield patcher |
| |
| def test_port_forward(self): |
| with self.platform.ports.nested() as ports: |
| with self.mock_popen( |
| self.host_platform) as mock_popen, self.mock_wait_for_port( |
| self.host_platform) as mock_wait_for_port: |
| port = ports.forward(666, 33221) |
| mock_popen.assert_called_once() |
| mock_wait_for_port.assert_called_once() |
| self.assertEqual(port, 666) |
| with self.assertRaisesRegex(PortForwardException, "twice"): |
| port = ports.forward(666, 33221) |
| ports.stop_forward(port) |
| |
| def test_port_forward_auto_port(self): |
| with self.platform.ports.nested() as ports: |
| with self.mock_get_free_port(self.host_platform, 666) as mock_free_port: |
| with self.mock_popen(self.host_platform) as mock_popen: |
| with self.mock_wait_for_port( |
| self.host_platform) as mock_wait_for_port: |
| port = ports.forward(0, 33221) |
| mock_popen.assert_called_once() |
| mock_wait_for_port.assert_called_once() |
| mock_free_port.assert_called_once() |
| self.assertEqual(port, 666) |
| with self.assertRaisesRegex(PortForwardException, "twice"): |
| port = ports.forward(666, 33221) |
| ports.stop_forward(port) |
| |
| def test_reverse_port_forward(self): |
| with self.platform.ports.nested() as ports: |
| self._expect_sh_ssh("ss -HOlnt sport = 666", result="666") |
| with self.mock_popen(self.host_platform) as mock_popen: |
| port = ports.reverse_forward(666, 33221) |
| mock_popen.assert_called_once() |
| with self.assertRaisesRegex(PortForwardException, "twice"): |
| ports.reverse_forward(666, 33221) |
| self.assertEqual(port, 666) |
| ports.stop_reverse_forward(port) |
| |
| def test_push_creates_dest_dir(self): |
| self._expect_sh_ssh("mkdir -p remote/dest/path") |
| self.mock_platform.expect_sh( |
| "scp", "-P", self.SSH_PORT, "source/path/file", |
| f"{self.SSH_USER}@{self.HOST}:remote/dest/path/file") |
| self.platform.push( |
| self.host_platform.path("source/path/file"), |
| self.platform.path("remote/dest/path/file")) |
| |
| def test_push_dir(self): |
| self._expect_sh_ssh("mkdir -p remote/dest/path") |
| self.mock_platform.expect_sh( |
| "scp", "-P", self.SSH_PORT, "-r", "source/path/dir", |
| f"{self.SSH_USER}@{self.HOST}:remote/dest/path/dir") |
| source_dir = self.host_platform.path("source/path/dir") |
| self.fs.create_dir(source_dir) |
| self.platform.push(source_dir, self.platform.path("remote/dest/path/dir")) |
| |
| def test_pull_creates_dest_dir(self): |
| self.mock_platform.expect_sh( |
| "scp", "-P", self.SSH_PORT, |
| f"{self.SSH_USER}@{self.HOST}:remote/source/path/file", |
| "local/dest/path/file") |
| self.platform.pull( |
| self.platform.path("remote/source/path/file"), |
| self.platform.path("local/dest/path/file")) |
| |
| self.assertEqual(self.mock_platform.mkdir_calls, 1) |
| self.assertTrue(os.path.exists("local/dest/path")) |
| |
| if __name__ == "__main__": |
| test_helper.run_pytest(__file__) |
