• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env vpython3
2# Copyright 2022 The Chromium Authors
3# Use of this source code is governed by a BSD-style license that can be
4# found in the LICENSE file.
5"""File for testing test_server.py."""
6
7import unittest
8import unittest.mock as mock
9
10from subprocess import CalledProcessError
11
12import test_server
13
14# TODO(crbug.com/40935291): Specifying fuchsia-port is unsupported yet.
15_FUCHSIA_PORT = 0
16_TARGET_ADDR = '127.0.0.1:33333'
17_HOST_PORT = 55555
18
19port_pair = (_FUCHSIA_PORT, _HOST_PORT)
20
21# Test names should be self-explained.
22# pylint: disable=missing-function-docstring
23
24class TestServerTest(unittest.TestCase):
25    """Unittests for test_server.py."""
26
27    def setUp(self) -> None:
28        self._subprocess_patcher = mock.patch('test_server.subprocess.run')
29        self._log_patcher = mock.patch('test_server.logging.debug')
30        self._subprocess_mock = self._subprocess_patcher.start()
31        self._log_mock = self._log_patcher.start()
32        self.addCleanup(self._log_mock.stop)
33        self.addCleanup(self._subprocess_mock.stop)
34
35        self._cmd_mock = mock.Mock()
36        self._cmd_mock.returncode = 0
37        self._cmd_mock.stdout = str(port_pair[0])
38        self._subprocess_mock.return_value = self._cmd_mock
39
40        self._forwarder = test_server.SSHPortForwarder(_TARGET_ADDR)
41        self._forwarder.Map([port_pair])
42
43    def test_ssh_port_forwarder(self) -> None:
44        self.assertEqual(self._subprocess_mock.call_count, 2)
45        self.assertEqual(
46            self._forwarder.GetDevicePortForHostPort(port_pair[1]),
47            port_pair[0])
48
49    def test_ssh_port_forwarder_unmapped(self) -> None:
50        self._forwarder.Unmap(port_pair[0])
51        # Unmap should raise an exception if no ports are mapped.
52        with self.assertRaises(Exception):
53            self._forwarder.Unmap(port_pair[0])
54
55    def test_ssh_port_forwarder_ssh_exception(self) -> None:
56        # Unmap should also raise an exception if the unmap command fails.
57        self._subprocess_mock.side_effect = [
58            self._cmd_mock,
59            CalledProcessError(cmd='ssh', returncode=1)
60        ]
61        with self.assertRaises(Exception):
62            self._forwarder.Unmap(port_pair[0])
63        self.assertEqual(self._subprocess_mock.call_count, 4)
64
65    def test_ssh_port_forward_unmap(self) -> None:
66        self._cmd_mock.returncode = 0
67        self._forwarder.Unmap(port_pair[0])
68        self.assertEqual(self._subprocess_mock.call_count, 4)
69
70    def test_port_forward_exception(self) -> None:
71        """Tests that exception is raised if |port_forward| command fails."""
72        cmd_mock = mock.Mock()
73        cmd_mock.returncode = 1
74        self._subprocess_mock.return_value = cmd_mock
75        with self.assertRaises(Exception):
76            test_server.port_forward(_TARGET_ADDR, _FUCHSIA_PORT)
77
78    @mock.patch('test_server.chrome_test_server_spawner.SpawningServer')
79    @mock.patch('test_server.port_forward')
80    def test_setup_test_server(self, forward_mock, server_mock) -> None:
81        """Test |setup_test_server|."""
82
83        forward_mock.return_value = _FUCHSIA_PORT
84        server = test_server.chrome_test_server_spawner.SpawningServer
85        server.Start = mock.Mock()
86        server_mock.return_value = server
87        with mock.patch('test_server.get_ssh_address'):
88            _, url = test_server.setup_test_server(_TARGET_ADDR, 4)
89        self.assertTrue(str(_FUCHSIA_PORT) in url)
90
91
92if __name__ == '__main__':
93    unittest.main()
94