serve_repo_unittests.py (6410B)
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 serve_repo.py.""" 6 7 import argparse 8 import json 9 import unittest 10 import unittest.mock as mock 11 12 from types import SimpleNamespace 13 14 import serve_repo 15 16 from common import REPO_ALIAS, register_device_args 17 18 _REPO_DIR = 'test_repo_dir' 19 _REPO_NAME = 'test_repo_name' 20 _TARGET = 'test_target' 21 _NO_SERVERS_LIST = json.dumps({'ok': {'data': []}}) 22 _SERVERS_LIST = json.dumps( 23 {'ok': { 24 'data': [{ 25 'name': _REPO_NAME, 26 'repo_path': _REPO_DIR 27 }] 28 }}) 29 _WRONG_SERVERS_LIST = json.dumps( 30 {'ok': { 31 'data': [{ 32 'name': 'wrong_name', 33 'repo_path': _REPO_DIR 34 }] 35 }}) 36 37 38 # Tests private functions. 39 # pylint: disable=protected-access 40 class ServeRepoTest(unittest.TestCase): 41 """Unittests for serve_repo.py.""" 42 @mock.patch('serve_repo.run_ffx_command') 43 def test_start_server(self, mock_ffx) -> None: 44 """Test |_start_serving| function for start.""" 45 46 mock_ffx.side_effect = [ 47 SimpleNamespace(returncode=1, stderr='err1', stdout=''), 48 SimpleNamespace(returncode=0, stdout=_SERVERS_LIST, stderr=''), 49 SimpleNamespace(returncode=0, stdout=_SERVERS_LIST, stderr=''), 50 SimpleNamespace(returncode=0, stdout='', stderr='') 51 ] 52 serve_repo._start_serving(_REPO_DIR, _REPO_NAME, _TARGET) 53 self.assertEqual(mock_ffx.call_count, 3) 54 first_call = mock_ffx.call_args_list[0] 55 56 self.assertEqual( 57 mock.call(cmd=[ 58 'repository', 'server', 'start', '--background', '--address', 59 '[::]:0', '--repository', _REPO_NAME, '--repo-path', _REPO_DIR, 60 '--no-device' 61 ], 62 check=False), first_call) 63 second_call = mock_ffx.call_args_list[1] 64 self.assertEqual( 65 mock.call(cmd=[ 66 '--machine', 'json', 'repository', 'server', 'list', '--name', 67 _REPO_NAME 68 ], 69 check=False, 70 capture_output=True), second_call) 71 72 third_call = mock_ffx.call_args_list[2] 73 self.assertEqual( 74 mock.call(cmd=[ 75 'target', 'repository', 'register', '-r', _REPO_NAME, 76 '--alias', REPO_ALIAS 77 ], 78 target_id=_TARGET), third_call) 79 80 @mock.patch('serve_repo.run_ffx_command') 81 def test_assert_server_running(self, mock_ffx) -> None: 82 """Test |_assert_server_running| function for start.""" 83 84 mock_ffx.side_effect = [ 85 SimpleNamespace(returncode=0, stdout=_SERVERS_LIST) 86 ] 87 # Raises an error if there is a problem, so no need to check for 88 # RuntimeError. 89 try: 90 serve_repo._assert_server_running(_REPO_NAME) 91 except RuntimeError as err: 92 self.fail(f'Unexpected error: {err}') 93 94 @mock.patch('serve_repo.run_ffx_command') 95 def test_is_server_not_running(self, mock_ffx) -> None: 96 """Test |_assert_server_running| function for start with no server.""" 97 98 mock_ffx.return_value = SimpleNamespace(returncode=0, 99 stdout=_NO_SERVERS_LIST, 100 stderr='') 101 102 with self.assertRaises(RuntimeError): 103 serve_repo._assert_server_running(_REPO_NAME) 104 105 @mock.patch('serve_repo.run_ffx_command') 106 def test_is_wrong_server_running(self, mock_ffx) -> None: 107 """Test |_assert_server_running| function for start with no server.""" 108 109 mock_ffx.return_value = SimpleNamespace(returncode=0, 110 stdout=_WRONG_SERVERS_LIST, 111 stderr='') 112 113 with self.assertRaises(RuntimeError): 114 serve_repo._assert_server_running(_REPO_NAME) 115 116 @mock.patch('serve_repo.run_ffx_command') 117 def test_is_server_not_running_bad_ffx(self, mock_ffx) -> None: 118 """Test |_assert_server_running| function for start with bad ffx.""" 119 120 mock_ffx.return_value = SimpleNamespace(returncode=1, 121 stderr='Some error', 122 stdout='') 123 124 with self.assertRaises(RuntimeError): 125 serve_repo._assert_server_running(_REPO_NAME) 126 127 @mock.patch('serve_repo.run_ffx_command') 128 def test_is_server_not_running_bad_json(self, mock_ffx) -> None: 129 """Test |_assert_server_running| function for start with bad ffx.""" 130 131 mock_ffx.return_value = SimpleNamespace(returncode=0, 132 stderr='', 133 stdout='{"some": bad...') 134 135 with self.assertRaises(RuntimeError): 136 serve_repo._assert_server_running(_REPO_NAME) 137 138 @mock.patch('serve_repo.run_ffx_command') 139 def test_stop_server(self, mock_ffx) -> None: 140 """Test |_stop_serving| function for stop.""" 141 142 serve_repo._stop_serving(_REPO_NAME, _TARGET) 143 self.assertEqual(mock_ffx.call_count, 2) 144 first_call = mock_ffx.call_args_list[0] 145 146 self.assertEqual( 147 mock.call( 148 cmd=['target', 'repository', 'deregister', '-r', _REPO_NAME], 149 target_id=_TARGET, 150 check=False), first_call) 151 second_call = mock_ffx.call_args_list[1] 152 self.assertEqual( 153 mock.call(cmd=['repository', 'server', 'stop', _REPO_NAME], 154 check=False), second_call) 155 156 @mock.patch('serve_repo._start_serving') 157 @mock.patch('serve_repo._stop_serving') 158 def test_serve_repository(self, mock_stop, mock_start) -> None: 159 """Tests |serve_repository| context manager.""" 160 161 parser = argparse.ArgumentParser() 162 serve_repo.register_serve_args(parser) 163 register_device_args(parser) 164 with serve_repo.serve_repository( 165 parser.parse_args([ 166 '--repo', _REPO_DIR, '--repo-name', _REPO_NAME, 167 '--target-id', _TARGET 168 ])): 169 self.assertEqual(mock_start.call_count, 1) 170 self.assertEqual(mock_stop.call_count, 1) 171 172 173 if __name__ == '__main__': 174 unittest.main()