#!/usr/bin/env python # # Copyright (C) 2017 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. # import os import unittest try: from unittest import mock except ImportError: import mock try: import StringIO as string_io_module except ImportError: import io as string_io_module from host_controller.build import build_flasher from host_controller.tfc import command_task from host_controller.tfc import device_info from host_controller import common from host_controller import console class ConsoleTest(unittest.TestCase): """A test for console.Console. Attribute: _out_file: The console output buffer. _host_controller: A mock tfc_host_controller.HostController. _build_provider_pab: A mock build_provider_pab.BuildProviderPAB. _tfc_client: A mock tfc_client.TfcClient. _vti_client A mock vti_endpoint_client.VtiEndpointClient. _console: The console being tested. """ _DEVICES = [ device_info.DeviceInfo( device_serial="ABC001", run_target="sailfish", state="Available", build_id="111111", sdk_version="27") ] _TASKS = [ command_task.CommandTask( request_id="1", task_id="1-0", command_id="2", command_line="vts -m SampleShellTest", device_serials=["ABC001"]) ] def setUp(self): """Creates the console.""" self._out_file = string_io_module.StringIO() self._host_controller = mock.Mock() self._build_provider_pab = mock.Mock() self._tfc_client = mock.Mock() self._vti_client = mock.Mock() self._console = console.Console( self._vti_client, self._tfc_client, self._build_provider_pab, [self._host_controller], None, out_file=self._out_file) self._console.device_image_info = {} def tearDown(self): """Closes the output file.""" self._out_file.close() def _IssueCommand(self, command_line): """Issues a command in the console. Args: command_line: A string, the input to the console. Returns: A string, the output of the console. """ out_position = self._out_file.tell() self._console.onecmd(command_line) self._out_file.seek(out_position) return self._out_file.read() def testLease(self): """Tests the lease command.""" self._host_controller.LeaseCommandTasks.return_value = self._TASKS output = self._IssueCommand("lease") expected = ( "request_id command_id task_id device_serials command_line \n" "1 2 1-0 ABC001 vts -m SampleShellTest\n" ) self.assertEqual(expected, output) output = self._IssueCommand("lease --host 0") self.assertEqual(expected, output) def testRequest(self): """Tests the request command.""" user = "user0" cluster = "cluster0" run_target = "sailfish" command_line = "vts -m SampleShellTest" self._IssueCommand("request --user %s --cluster %s --run-target %s " "-- %s" % (user, cluster, run_target, command_line)) req = self._tfc_client.NewRequest.call_args[0][0] self.assertEqual(user, req.user) self.assertEqual(cluster, req.cluster) self.assertEqual(run_target, req.run_target) self.assertEqual(command_line, req.command_line) def testListHosts(self): """Tests the list command.""" self._host_controller.hostname = "host0" output = self._IssueCommand("list hosts") self.assertEqual("index name\n" "[ 0] host0\n", output) def testListDevices(self): """Tests the list command.""" self._host_controller.ListDevices.return_value = self._DEVICES self._host_controller.hostname = "host0" output = self._IssueCommand("list devices") expected = ( "[ 0] host0\n" "device_serial state run_target build_id sdk_version stub\n" "ABC001 Available sailfish 111111 27 \n" ) self.assertEqual(expected, output) output = self._IssueCommand("list devices --host 0") self.assertEqual(expected, output) def testWrongHostIndex(self): """Tests host index out of range.""" output = self._IssueCommand("list devices --host 1") expected = "IndexError: " self.assertTrue(output.startswith(expected)) output = self._IssueCommand("lease --host 1") self.assertTrue(output.startswith(expected)) @mock.patch('host_controller.build.build_flasher.BuildFlasher') def testFetchPOSTAndFlash(self, mock_class): """Tests fetching from pab and flashing.""" self._build_provider_pab.GetArtifact.return_value = ({ "system.img": "/mock/system.img", "odm.img": "/mock/odm.img" }, {}, { "build_id": "build_id" }, {}) self._build_provider_pab.GetFetchedArtifactType.return_value = common._ARTIFACT_TYPE_DEVICE self._IssueCommand( "fetch --branch=aosp-master-ndk --target=darwin_mac " "--account_id=100621237 " "--artifact_name=foo-{build_id}.tar.bz2 --method=POST") self._build_provider_pab.GetArtifact.assert_called_with( account_id='100621237', branch='aosp-master-ndk', target='darwin_mac', artifact_name='foo-{build_id}.tar.bz2', build_id='latest', method='POST', full_device_images=False) self.assertEqual(self._console.device_image_info, { "system.img": "/mock/system.img", "odm.img": "/mock/odm.img" }) flasher = mock.Mock() mock_class.return_value = flasher self._IssueCommand("flash --current system=system.img odm=odm.img") flasher.Flash.assert_called_with({ "system": "/mock/system.img", "odm": "/mock/odm.img" }, False) def testFetchAndEnvironment(self): """Tests fetching from pab and check stored os environment""" build_id_return = "4328532" target_return = "darwin_mac" expected_fetch_info = {"build_id": build_id_return} self._build_provider_pab.GetArtifact.return_value = ({ "system.img": "/mock/system.img", "odm.img": "/mock/odm.img" }, {}, expected_fetch_info, {}) self._IssueCommand( "fetch --branch=aosp-master-ndk --target=%s " "--account_id=100621237 " "--artifact_name=foo-{id}.tar.bz2 --method=POST" % target_return) self._build_provider_pab.GetArtifact.assert_called_with( account_id='100621237', branch='aosp-master-ndk', target='darwin_mac', artifact_name='foo-{id}.tar.bz2', build_id='latest', full_device_images=False, method='POST') expected = expected_fetch_info["build_id"] self.assertEqual(build_id_return, expected) @mock.patch('host_controller.build.build_flasher.BuildFlasher') def testFlashGSI(self, mock_class): flasher = mock.Mock() mock_class.return_value = flasher self._IssueCommand("flash --gsi=system.img") flasher.FlashGSI.assert_called_with( 'system.img', None, skip_vbmeta=False) @mock.patch('host_controller.build.build_flasher.BuildFlasher') def testFlashGSIWithVbmeta(self, mock_class): flasher = mock.Mock() mock_class.return_value = flasher self._IssueCommand("flash --gsi=system.img --vbmeta=vbmeta.img") flasher.FlashGSI.assert_called_with( 'system.img', 'vbmeta.img', skip_vbmeta=False) @mock.patch('host_controller.build.build_flasher.BuildFlasher') def testFlashall(self, mock_class): flasher = mock.Mock() mock_class.return_value = flasher self._IssueCommand("flash --build_dir=path/to/dir/") flasher.Flashall.assert_called_with('path/to/dir/') @mock.patch('host_controller.command_processor.command_flash.importlib') @mock.patch('host_controller.command_processor.command_flash.issubclass') def testImportFlasher(self, mock_issubclass, mock_importlib): mock_issubclass.return_value = True flasher_module = mock.Mock() flasher = mock.Mock() mock_importlib.import_module.return_value = flasher_module flasher_module.Flasher.return_value = flasher self._IssueCommand("flash --serial ABC001 " "--flasher_type test.flasher.Flasher " "--flasher_path /test/flasher " "-- --unit test") mock_issubclass.assert_called_once_with(flasher_module.Flasher, build_flasher.BuildFlasher) mock_importlib.import_module.assert_called_with("test.flasher") flasher_module.Flasher.assert_called_with("ABC001", "/test/flasher") flasher.Flash.assert_called_with({}, {}, "--unit", "test") flasher.WaitForDevice.assert_called_with() if __name__ == "__main__": unittest.main()