# Copyright 2019 - The Android Open Source Project # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Tests for pull.""" import unittest import os import tempfile from unittest import mock from acloud import errors from acloud.internal import constants from acloud.internal.lib import driver_test_lib from acloud.internal.lib import ssh from acloud.internal.lib import utils from acloud.list import list as list_instances from acloud.public import config from acloud.pull import pull class PullTest(driver_test_lib.BaseDriverTest): """Test pull.""" # pylint: disable=no-member def testPullFileFromInstance(self): """test PullFileFromInstance.""" cfg = mock.MagicMock() cfg.ssh_private_key_path = "fake_ssh_path" cfg.extra_args_ssh_tunnel = "" instance = mock.MagicMock() instance.ip = "1.1.1.1" # Multiple selected files case. selected_files = ["file1.log", "file2.log"] self.Patch(pull, "SelectLogFileToPull", return_value=selected_files) self.Patch(pull, "GetDownloadLogFolder", return_value="fake_folder") self.Patch(pull, "PullLogs") self.Patch(pull, "DisplayLog") pull.PullFileFromInstance(cfg, instance) self.assertEqual(pull.DisplayLog.call_count, 0) # Only one file selected case. selected_files = ["file1.log"] self.Patch(pull, "SelectLogFileToPull", return_value=selected_files) pull.PullFileFromInstance(cfg, instance) self.assertEqual(pull.DisplayLog.call_count, 1) # pylint: disable=no-member def testPullLogs(self): """test PullLogs.""" _ssh = mock.MagicMock() self.Patch(utils, "PrintColorString") log_files = ["file1.log", "file2.log"] download_folder = "/fake_folder" pull.PullLogs(_ssh, log_files, download_folder) self.assertEqual(_ssh.ScpPullFile.call_count, 2) utils.PrintColorString.assert_called_once() @mock.patch.object(ssh.Ssh, "Run") def testDisplayLog(self, mock_ssh_run): """Test DisplayLog.""" fake_ip = ssh.IP(external="1.1.1.1", internal="10.1.1.1") _ssh = ssh.Ssh(ip=fake_ip, user=constants.GCE_USER, ssh_private_key_path="/fake/acloud_rea") self.Patch(utils, "GetUserAnswerYes", return_value="Y") log_file = "file1.log" pull.DisplayLog(_ssh, log_file) expected_cmd = "tail -f -n +1 %s" % log_file mock_ssh_run.assert_has_calls([ mock.call(expected_cmd, show_output=True)]) def testGetDownloadLogFolder(self): """test GetDownloadLogFolder.""" self.Patch(tempfile, "gettempdir", return_value="/tmp") self.Patch(os.path, "exists", return_value=True) instance = "instance" expected_path = "/tmp/instance" self.assertEqual(pull.GetDownloadLogFolder(instance), expected_path) def testSelectLogFileToPull(self): """test choose log files from the remote instance.""" _ssh = mock.MagicMock() # Test only one log file case log_files = ["file1.log"] self.Patch(pull, "GetAllLogFilePaths", return_value=log_files) expected_result = ["file1.log"] self.assertEqual(pull.SelectLogFileToPull(_ssh), expected_result) # Test no log files case self.Patch(pull, "GetAllLogFilePaths", return_value=[]) with self.assertRaises(errors.CheckPathError): pull.SelectLogFileToPull(_ssh) # Test two log files case. log_files = ["file1.log", "file2.log"] choose_log = ["file2.log"] self.Patch(pull, "GetAllLogFilePaths", return_value=log_files) self.Patch(utils, "GetAnswerFromList", return_value=choose_log) expected_result = ["file2.log"] self.assertEqual(pull.SelectLogFileToPull(_ssh), expected_result) # Test user provided file name exist. log_files = ["/home/vsoc-01/cuttlefish_runtime/file1.log", "/home/vsoc-01/cuttlefish_runtime/file2.log"] input_file = "file1.log" self.Patch(pull, "GetAllLogFilePaths", return_value=log_files) expected_result = ["/home/vsoc-01/cuttlefish_runtime/file1.log"] self.assertEqual(pull.SelectLogFileToPull(_ssh, input_file), expected_result) # Test user provided file name not exist. log_files = ["/home/vsoc-01/cuttlefish_runtime/file1.log", "/home/vsoc-01/cuttlefish_runtime/file2.log"] input_file = "not_exist.log" self.Patch(pull, "GetAllLogFilePaths", return_value=log_files) with self.assertRaises(errors.CheckPathError): pull.SelectLogFileToPull(_ssh, input_file) def testFilterLogfiles(self): """test filer log file from black list.""" # Filter out file name is "kernel". files = ["kernel.log", "logcat", "kernel"] expected_result = ["kernel.log", "logcat"] self.assertEqual(pull.FilterLogfiles(files), expected_result) # Filter out file extension is ".img". files = ["kernel.log", "system.img", "userdata.img", "launcher.log"] expected_result = ["kernel.log", "launcher.log"] self.assertEqual(pull.FilterLogfiles(files), expected_result) @mock.patch.object(pull, "PullFileFromInstance") def testRun(self, mock_pull_file): """test Run.""" cfg = mock.MagicMock() args = mock.MagicMock() instance_obj = mock.MagicMock() # Test case with provided instance name. args.instance_name = "instance_1" args.file_name = "file1.log" args.no_prompt = True self.Patch(config, "GetAcloudConfig", return_value=cfg) self.Patch(list_instances, "GetInstancesFromInstanceNames", return_value=[instance_obj]) pull.Run(args) mock_pull_file.assert_has_calls([ mock.call(cfg, instance_obj, args.file_name, args.no_prompt)]) # Test case for user select one instance to pull log. selected_instance = mock.MagicMock() self.Patch(list_instances, "ChooseOneRemoteInstance", return_value=selected_instance) args.instance_name = None pull.Run(args) mock_pull_file.assert_has_calls([ mock.call(cfg, selected_instance, args.file_name, args.no_prompt)]) if __name__ == '__main__': unittest.main()