xref: /aosp_15_r20/external/cronet/build/fuchsia/test/test_server_unittests.py (revision 6777b5387eb2ff775bb5750e3f5d96f37fb7352b) 
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
10import test_server
11
12_HOST_PORT = 44444
13_HOST_PORT_PAIR = '127.0.0.1:33333'
14_SERVER_PORT = 55555
15
16
17class TestServerTest(unittest.TestCase):
18    """Unittests for test_server.py."""
19
20    def setUp(self) -> None:
21        self._subprocess_patcher = mock.patch('test_server.subprocess.run')
22        self._log_patcher = mock.patch('test_server.logging.debug')
23        self._subprocess_mock = self._subprocess_patcher.start()
24        self._log_mock = self._log_patcher.start()
25        self.addCleanup(self._log_mock.stop)
26        self.addCleanup(self._subprocess_mock.stop)
27
28    def test_ssh_port_forwarder(self) -> None:
29        """Test SSHPortForwarder."""
30
31        port_pair = (_HOST_PORT, _SERVER_PORT)
32        cmd_mock = mock.Mock()
33        cmd_mock.returncode = 0
34        cmd_mock.stdout = str(port_pair[0])
35        self._subprocess_mock.return_value = cmd_mock
36
37        forwarder = test_server.SSHPortForwarder(_HOST_PORT_PAIR)
38
39        # Unmap should raise an exception if no ports are mapped.
40        with self.assertRaises(Exception):
41            forwarder.Unmap(port_pair[0])
42
43        forwarder.Map([port_pair])
44        self.assertEqual(self._subprocess_mock.call_count, 2)
45        self.assertEqual(forwarder.GetDevicePortForHostPort(port_pair[1]),
46                         port_pair[0])
47
48        # Unmap should also raise an exception if the unmap command fails.
49        self._subprocess_mock.reset_mock()
50        cmd_mock.returncode = 1
51        with self.assertRaises(Exception):
52            forwarder.Unmap(port_pair[0])
53        self.assertEqual(self._subprocess_mock.call_count, 1)
54
55        self._subprocess_mock.reset_mock()
56        cmd_mock.returncode = 0
57        forwarder.Unmap(port_pair[0])
58        self.assertEqual(self._subprocess_mock.call_count, 1)
59
60    def test_port_forward_exception(self) -> None:
61        """Tests that exception is raised if |port_forward| command fails."""
62
63        cmd_mock = mock.Mock()
64        cmd_mock.returncode = 1
65        self._subprocess_mock.return_value = cmd_mock
66        with self.assertRaises(Exception):
67            test_server.port_forward(_HOST_PORT_PAIR, _HOST_PORT)
68
69    @mock.patch('test_server.chrome_test_server_spawner.SpawningServer')
70    @mock.patch('test_server.port_forward')
71    def test_setup_test_server(self, forward_mock, server_mock) -> None:
72        """Test |setup_test_server|."""
73
74        forward_mock.return_value = _HOST_PORT
75        server = test_server.chrome_test_server_spawner.SpawningServer
76        server.Start = mock.Mock()
77        server_mock.return_value = server
78        with mock.patch('test_server.run_ffx_command'):
79            _, url = test_server.setup_test_server(_HOST_PORT_PAIR, 4)
80        self.assertTrue(str(_HOST_PORT) in url)
81
82
83if __name__ == '__main__':
84    unittest.main()
85