443 lines
18 KiB
Python
443 lines
18 KiB
Python
# Copyright 2018 - 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 delete."""
|
|
|
|
import subprocess
|
|
import unittest
|
|
|
|
from unittest import mock
|
|
|
|
from acloud import errors
|
|
from acloud.delete import delete
|
|
from acloud.internal.lib import driver_test_lib
|
|
from acloud.internal.lib import oxygen_client
|
|
from acloud.internal.lib import utils
|
|
from acloud.list import list as list_instances
|
|
from acloud.public import config
|
|
from acloud.public import device_driver
|
|
from acloud.public import report
|
|
|
|
|
|
# pylint: disable=invalid-name,protected-access,unused-argument,no-member
|
|
class DeleteTest(driver_test_lib.BaseDriverTest):
|
|
"""Test delete functions."""
|
|
|
|
def testDeleteLocalCuttlefishInstanceSuccess(self):
|
|
"""Test DeleteLocalCuttlefishInstance."""
|
|
instance_object = mock.MagicMock()
|
|
instance_object.name = "local-instance"
|
|
mock_lock = mock.Mock()
|
|
mock_lock.Lock.return_value = True
|
|
instance_object.GetLock.return_value = mock_lock
|
|
|
|
delete_report = report.Report(command="delete")
|
|
delete.DeleteLocalCuttlefishInstance(instance_object, delete_report)
|
|
self.assertEqual(delete_report.data, {
|
|
"deleted": [
|
|
{
|
|
"type": "instance",
|
|
"name": "local-instance",
|
|
},
|
|
],
|
|
})
|
|
self.assertEqual(delete_report.status, "SUCCESS")
|
|
mock_lock.SetInUse.assert_called_once_with(False)
|
|
mock_lock.Unlock.assert_called_once()
|
|
|
|
mock_lock.Lock.return_value = False
|
|
delete.DeleteLocalCuttlefishInstance(instance_object, delete_report)
|
|
self.assertEqual(delete_report.status, "FAIL")
|
|
|
|
def testDeleteLocalCuttlefishInstanceFailure(self):
|
|
"""Test DeleteLocalCuttlefishInstance with command failure."""
|
|
instance_object = mock.MagicMock()
|
|
instance_object.name = "local-instance"
|
|
instance_object.Delete.side_effect = subprocess.CalledProcessError(
|
|
1, "cmd")
|
|
mock_lock = mock.Mock()
|
|
mock_lock.Lock.return_value = True
|
|
instance_object.GetLock.return_value = mock_lock
|
|
|
|
delete_report = report.Report(command="delete")
|
|
delete.DeleteLocalCuttlefishInstance(instance_object, delete_report)
|
|
|
|
self.assertEqual(delete_report.status, "FAIL")
|
|
mock_lock.SetInUse.assert_called_once_with(False)
|
|
mock_lock.Unlock.assert_called_once()
|
|
|
|
def testDeleteLocalGoldfishInstanceSuccess(self):
|
|
"""Test DeleteLocalGoldfishInstance."""
|
|
mock_adb_tools = mock.Mock()
|
|
mock_adb_tools.EmuCommand.return_value = 0
|
|
mock_instance = mock.Mock(adb=mock_adb_tools,
|
|
adb_port=5555,
|
|
device_serial="serial",
|
|
instance_dir="/unit/test")
|
|
# name is a positional argument of Mock().
|
|
mock_instance.name = "unittest"
|
|
mock_lock = mock.Mock()
|
|
mock_lock.Lock.return_value = True
|
|
mock_instance.GetLock.return_value = mock_lock
|
|
|
|
delete_report = report.Report(command="delete")
|
|
delete.DeleteLocalGoldfishInstance(mock_instance, delete_report)
|
|
|
|
mock_adb_tools.EmuCommand.assert_called_with("kill")
|
|
self.assertEqual(delete_report.data, {
|
|
"deleted": [
|
|
{
|
|
"type": "instance",
|
|
"name": "unittest",
|
|
},
|
|
],
|
|
})
|
|
self.assertEqual(delete_report.status, "SUCCESS")
|
|
mock_lock.SetInUse.assert_called_once_with(False)
|
|
mock_lock.Unlock.assert_called_once()
|
|
|
|
mock_lock.Lock.return_value = False
|
|
delete.DeleteLocalGoldfishInstance(mock_instance, delete_report)
|
|
self.assertEqual(delete_report.status, "FAIL")
|
|
|
|
def testDeleteLocalGoldfishInstanceFailure(self):
|
|
"""Test DeleteLocalGoldfishInstance with adb command failure."""
|
|
mock_adb_tools = mock.Mock()
|
|
mock_adb_tools.EmuCommand.return_value = 1
|
|
mock_instance = mock.Mock(adb=mock_adb_tools,
|
|
adb_port=5555,
|
|
device_serial="serial",
|
|
instance_dir="/unit/test")
|
|
# name is a positional argument of Mock().
|
|
mock_instance.name = "unittest"
|
|
mock_lock = mock.Mock()
|
|
mock_lock.Lock.return_value = True
|
|
mock_instance.GetLock.return_value = mock_lock
|
|
|
|
delete_report = report.Report(command="delete")
|
|
delete.DeleteLocalGoldfishInstance(mock_instance, delete_report)
|
|
|
|
mock_adb_tools.EmuCommand.assert_called_with("kill")
|
|
self.assertTrue(len(delete_report.errors) > 0)
|
|
self.assertEqual(delete_report.status, "FAIL")
|
|
mock_lock.SetInUse.assert_called_once_with(False)
|
|
mock_lock.Unlock.assert_called_once()
|
|
|
|
def testResetLocalInstanceLockByName(self):
|
|
"""test ResetLocalInstanceLockByName."""
|
|
mock_lock = mock.Mock()
|
|
mock_lock.Lock.return_value = True
|
|
self.Patch(list_instances, "GetLocalInstanceLockByName",
|
|
return_value=mock_lock)
|
|
delete_report = report.Report(command="delete")
|
|
delete.ResetLocalInstanceLockByName("unittest", delete_report)
|
|
|
|
self.assertEqual(delete_report.data, {
|
|
"deleted": [
|
|
{
|
|
"type": "instance",
|
|
"name": "unittest",
|
|
},
|
|
],
|
|
})
|
|
mock_lock.Lock.assert_called_once()
|
|
mock_lock.SetInUse.assert_called_once_with(False)
|
|
mock_lock.Unlock.assert_called_once()
|
|
|
|
mock_lock.Lock.return_value = False
|
|
delete.ResetLocalInstanceLockByName("unittest", delete_report)
|
|
self.assertEqual(delete_report.status, "FAIL")
|
|
|
|
def testResetLocalInstanceLockByNameFailure(self):
|
|
"""test ResetLocalInstanceLockByName with an invalid name."""
|
|
self.Patch(list_instances, "GetLocalInstanceLockByName",
|
|
return_value=None)
|
|
delete_report = report.Report(command="delete")
|
|
delete.ResetLocalInstanceLockByName("unittest", delete_report)
|
|
|
|
self.assertTrue(len(delete_report.errors) > 0)
|
|
self.assertEqual(delete_report.status, "FAIL")
|
|
|
|
@mock.patch("acloud.delete.delete.emulator_console.RemoteEmulatorConsole")
|
|
def testDeleteHostGoldfishInstance(self, mock_console):
|
|
"""test DeleteHostGoldfishInstance."""
|
|
mock_console_obj = mock.MagicMock()
|
|
mock_console.return_value = mock_console_obj
|
|
mock_console_obj.__enter__.return_value = mock_console_obj
|
|
|
|
cfg_attrs = {"ssh_private_key_path": "cfg_key_path",
|
|
"extra_args_ssh_tunnel": "extra args"}
|
|
mock_cfg = mock.Mock(spec_set=list(cfg_attrs.keys()), **cfg_attrs)
|
|
instance_name = "host-goldfish-192.0.2.1-5554-123456-sdk_x86_64-sdk"
|
|
delete_report = report.Report(command="delete")
|
|
|
|
delete.DeleteHostGoldfishInstance(mock_cfg, instance_name,
|
|
None, None, delete_report)
|
|
mock_console.assert_called_with("192.0.2.1", 5554, "vsoc-01",
|
|
"cfg_key_path", "extra args")
|
|
mock_console_obj.Kill.assert_called()
|
|
self.assertEqual(delete_report.status, "SUCCESS")
|
|
self.assertEqual(delete_report.data, {
|
|
"deleted": [
|
|
{
|
|
"type": "instance",
|
|
"name": instance_name,
|
|
},
|
|
],
|
|
})
|
|
|
|
mock_console_obj.reset_mock()
|
|
mock_console_obj.Kill.side_effect = errors.DeviceConnectionError
|
|
delete_report = report.Report(command="delete")
|
|
|
|
delete.DeleteHostGoldfishInstance(mock_cfg, instance_name,
|
|
"user", "key_path", delete_report)
|
|
mock_console.assert_called_with("192.0.2.1", 5554, "user",
|
|
"key_path", "extra args")
|
|
self.assertEqual(delete_report.status, "FAIL")
|
|
self.assertEqual(len(delete_report.errors), 1)
|
|
|
|
@mock.patch.object(delete, "ssh")
|
|
@mock.patch.object(delete, "cvd_utils")
|
|
def testCleanUpRemoteHost(self, mock_cvd_utils, mock_ssh):
|
|
"""Test CleanUpRemoteHost."""
|
|
mock_ssh_ip = mock.Mock()
|
|
mock_ssh.IP.return_value = mock_ssh_ip
|
|
mock_ssh_obj = mock.Mock()
|
|
mock_ssh.Ssh.return_value = mock_ssh_obj
|
|
cfg_attrs = {"ssh_private_key_path": "cfg_key_path"}
|
|
mock_cfg = mock.Mock(spec_set=list(cfg_attrs.keys()), **cfg_attrs)
|
|
delete_report = report.Report(command="delete")
|
|
delete.CleanUpRemoteHost(mock_cfg, "192.0.2.1", "vsoc-01", None, ".",
|
|
delete_report)
|
|
|
|
mock_ssh.IP.assert_called_with(ip="192.0.2.1")
|
|
mock_ssh.Ssh.assert_called_with(
|
|
ip=mock_ssh_ip,
|
|
user="vsoc-01",
|
|
ssh_private_key_path="cfg_key_path")
|
|
mock_cvd_utils.CleanUpRemoteCvd.assert_called_with(
|
|
mock_ssh_obj, ".", raise_error=True)
|
|
self.assertEqual(delete_report.status, "SUCCESS")
|
|
self.assertEqual(delete_report.data, {
|
|
"deleted": [
|
|
{
|
|
"type": "remote host",
|
|
"name": "192.0.2.1",
|
|
},
|
|
],
|
|
})
|
|
|
|
mock_ssh_ip.reset_mock()
|
|
mock_ssh_obj.reset_mock()
|
|
mock_cvd_utils.reset_mock()
|
|
mock_cvd_utils.CleanUpRemoteCvd.side_effect = (
|
|
subprocess.CalledProcessError(cmd="test", returncode=1))
|
|
delete_report = report.Report(command="delete")
|
|
|
|
delete.CleanUpRemoteHost(mock_cfg, "192.0.2.2", "user", "key_path",
|
|
"acloud_cf_1", delete_report)
|
|
mock_ssh.IP.assert_called_with(ip="192.0.2.2")
|
|
mock_ssh.Ssh.assert_called_with(
|
|
ip=mock_ssh_ip,
|
|
user="user",
|
|
ssh_private_key_path="key_path")
|
|
mock_cvd_utils.CleanUpRemoteCvd.assert_called_with(
|
|
mock_ssh_obj, "acloud_cf_1", raise_error=True)
|
|
self.assertEqual(delete_report.status, "FAIL")
|
|
self.assertEqual(len(delete_report.errors), 1)
|
|
|
|
@mock.patch.object(delete, "DeleteInstances", return_value="")
|
|
@mock.patch.object(delete, "ResetLocalInstanceLockByName")
|
|
@mock.patch.object(delete, "CleanUpRemoteHost")
|
|
@mock.patch.object(delete, "DeleteHostGoldfishInstance")
|
|
@mock.patch.object(delete, "DeleteRemoteInstances", return_value="")
|
|
def testDeleteInstanceByNames(self, mock_delete_remote_ins,
|
|
mock_delete_host_gf_ins,
|
|
mock_clean_up_remote_host, mock_reset_lock,
|
|
mock_delete_local_ins):
|
|
"""test DeleteInstanceByNames."""
|
|
cfg = mock.Mock()
|
|
# Test delete local instances.
|
|
instances = ["local-instance-1", "local-instance-2"]
|
|
mock_local_ins = mock.Mock()
|
|
mock_local_ins.name = "local-instance-1"
|
|
self.Patch(list_instances, "GetLocalInstancesByNames",
|
|
return_value=[mock_local_ins])
|
|
delete.DeleteInstanceByNames(cfg, instances, None, None)
|
|
mock_delete_local_ins.assert_called_with(cfg, [mock_local_ins])
|
|
mock_reset_lock.assert_called_with("local-instance-2", mock.ANY)
|
|
|
|
# Test delete remote host instances.
|
|
instances = ["host-goldfish-192.0.2.1-5554-123456-sdk_x86_64-sdk",
|
|
"host-192.0.2.2-3-123456-aosp_cf_x86_64_phone"]
|
|
delete.DeleteInstanceByNames(cfg, instances, "user", "key")
|
|
mock_delete_host_gf_ins.assert_called_with(
|
|
cfg, instances[0], "user", "key", mock.ANY)
|
|
mock_clean_up_remote_host.assert_called_with(
|
|
cfg, "192.0.2.2", "user", "key", "acloud_cf_3", mock.ANY)
|
|
|
|
# Test delete remote instances.
|
|
instances = ["ins-id1-cf-x86-phone-userdebug",
|
|
"ins-id2-cf-x86-phone-userdebug"]
|
|
delete.DeleteInstanceByNames(cfg, instances, None, None)
|
|
mock_delete_remote_ins.assert_called()
|
|
|
|
@mock.patch.object(oxygen_client.OxygenClient, "ReleaseDevice")
|
|
def testReleaseOxygenDevice(self, mock_release):
|
|
"""test ReleaseOxygenDevice"""
|
|
cfg = mock.Mock()
|
|
cfg.oxygen_client = "oxygen_client"
|
|
ip = "0.0.0.0"
|
|
# Raise exception for multiple instances
|
|
instances = ["local-instance-1", "local-instance-2"]
|
|
self.assertRaises(errors.CommandArgError, delete._ReleaseOxygenDevice, cfg, instances, ip)
|
|
|
|
# Test release device with oxygen client
|
|
instances = ["local-instance-1"]
|
|
delete._ReleaseOxygenDevice(cfg, instances, ip)
|
|
mock_release.assert_called_once()
|
|
|
|
mock_release.side_effect = subprocess.CalledProcessError(
|
|
0, "fake_cmd",
|
|
"Error received while trying to release device: error_msg")
|
|
delete_report = delete._ReleaseOxygenDevice(cfg, instances, ip)
|
|
self.assertEqual(delete_report.errors, ["error_msg"])
|
|
|
|
mock_release.side_effect = subprocess.CalledProcessError(
|
|
0, "fake_cmd",
|
|
"error")
|
|
delete_report = delete._ReleaseOxygenDevice(cfg, instances, ip)
|
|
self.assertEqual(delete_report.status, "FAIL")
|
|
|
|
def testDeleteInstances(self):
|
|
"""test DeleteInstances."""
|
|
fake_ins = mock.MagicMock()
|
|
fake_ins.islocal = False
|
|
fake_ins.avd_type = "cuttlefish"
|
|
fake_ins.vnc_port = None
|
|
|
|
fake_ins2 = mock.MagicMock()
|
|
fake_ins2.islocal = True
|
|
fake_ins2.avd_type = "cuttlefish"
|
|
fake_ins2.vnc_port = None
|
|
|
|
fake_ins3 = mock.MagicMock()
|
|
fake_ins3.islocal = True
|
|
fake_ins3.avd_type = "goldfish"
|
|
fake_ins3.vnc_port = None
|
|
|
|
fake_ins4 = mock.MagicMock()
|
|
fake_ins4.islocal = True
|
|
fake_ins4.avd_type = "unknown"
|
|
fake_ins4.vnc_port = 12345
|
|
|
|
self.Patch(delete, "DeleteLocalGoldfishInstance")
|
|
self.Patch(delete, "DeleteLocalCuttlefishInstance")
|
|
self.Patch(delete, "DeleteRemoteInstances")
|
|
self.Patch(utils, "CleanupSSVncviewer")
|
|
|
|
fake_instances_to_delete = []
|
|
delete.DeleteInstances(None, fake_instances_to_delete)
|
|
delete.DeleteRemoteInstances.assert_not_called()
|
|
|
|
fake_instances_to_delete = [
|
|
fake_ins, fake_ins2, fake_ins3, fake_ins4]
|
|
delete.DeleteInstances(None, fake_instances_to_delete)
|
|
delete.DeleteRemoteInstances.assert_called_once()
|
|
delete.DeleteLocalGoldfishInstance.assert_called_once()
|
|
delete.DeleteLocalCuttlefishInstance.assert_called_once()
|
|
utils.CleanupSSVncviewer.assert_called_once()
|
|
|
|
def testDeleteRemoteInstances(self):
|
|
"""test DeleteRemoteInstances."""
|
|
fake_cfg = mock.MagicMock()
|
|
fake_cfg.SupportRemoteInstance = mock.MagicMock()
|
|
fake_cfg.SupportRemoteInstance.return_value = True
|
|
fake_instances_to_delete = ["fake_ins"]
|
|
delete_report = report.Report(command="delete")
|
|
self.Patch(device_driver, "DeleteAndroidVirtualDevices",
|
|
return_value=delete_report)
|
|
delete.DeleteRemoteInstances(fake_cfg, fake_instances_to_delete)
|
|
device_driver.DeleteAndroidVirtualDevices.assert_called_once()
|
|
|
|
fake_cfg.SupportRemoteInstance.return_value = False
|
|
self.assertRaises(errors.ConfigError,
|
|
delete.DeleteRemoteInstances,
|
|
fake_cfg, fake_instances_to_delete)
|
|
|
|
def testRun(self):
|
|
"""test Run."""
|
|
args = mock.MagicMock()
|
|
args.oxygen = False
|
|
args.instance_names = None
|
|
args.remote_host = None
|
|
args.local_only = True
|
|
args.adb_port = None
|
|
args.fastboot_port = None
|
|
args.all = True
|
|
|
|
self.Patch(delete, "_ReleaseOxygenDevice")
|
|
self.Patch(delete, "DeleteInstanceByNames")
|
|
self.Patch(delete, "CleanUpRemoteHost")
|
|
fake_cfg = mock.MagicMock()
|
|
fake_cfg.SupportRemoteInstance = mock.MagicMock()
|
|
self.Patch(config, "GetAcloudConfig", return_value=fake_cfg)
|
|
self.Patch(list_instances, "GetLocalInstances",
|
|
return_value=[])
|
|
self.Patch(list_instances, "GetRemoteInstances",
|
|
return_value=["remote_instances"])
|
|
self.Patch(list_instances, "FilterInstancesByAdbPort",
|
|
return_value=["filter_by_port_instance"])
|
|
self.Patch(list_instances, "ChooseInstancesFromList",
|
|
return_value=["choice_instance"])
|
|
self.Patch(delete, "DeleteInstances")
|
|
|
|
delete.Run(args)
|
|
delete.DeleteInstances.assert_called_with(fake_cfg, [])
|
|
|
|
list_instances.GetLocalInstances.return_value = ["local_instances"]
|
|
delete.Run(args)
|
|
delete.DeleteInstances.assert_called_with(fake_cfg, ["local_instances"])
|
|
|
|
args.all = False
|
|
delete.Run(args)
|
|
delete.DeleteInstances.assert_called_with(fake_cfg, ["choice_instance"])
|
|
|
|
args.adb_port = "12345"
|
|
delete.Run(args)
|
|
delete.DeleteInstances.assert_called_with(fake_cfg, ["filter_by_port_instance"])
|
|
|
|
args.local_only = False
|
|
args.all = True
|
|
args.adb_port = None
|
|
delete.Run(args)
|
|
delete.DeleteInstances.assert_called_with(
|
|
fake_cfg, ["local_instances", "remote_instances"])
|
|
|
|
args.remote_host = True
|
|
delete.Run(args)
|
|
delete.CleanUpRemoteHost.assert_called_once()
|
|
|
|
args.instance_names = ["fake_ins_name"]
|
|
delete.Run(args)
|
|
delete.DeleteInstanceByNames.assert_called_once()
|
|
|
|
args.oxygen = True
|
|
delete.Run(args)
|
|
delete._ReleaseOxygenDevice.assert_called_once()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|