Sam Chiu | 5029a25 | 2018-11-06 20:54:13 +0800 | [diff] [blame^] | 1 | #!/usr/bin/env python |
| 2 | # |
| 3 | # Copyright 2018 - The Android Open Source Project |
| 4 | # |
| 5 | # Licensed under the Apache License, Version 2.0 (the "License"); |
| 6 | # you may not use this file except in compliance with the License. |
| 7 | # You may obtain a copy of the License at |
| 8 | # |
| 9 | # http://www.apache.org/licenses/LICENSE-2.0 |
| 10 | # |
| 11 | # Unless required by applicable law or agreed to in writing, software |
| 12 | # distributed under the License is distributed on an "AS IS" BASIS, |
| 13 | # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 14 | # See the License for the specific language governing permissions and |
| 15 | # limitations under the License. |
| 16 | """Tests for host_setup_runner.""" |
| 17 | import collections |
| 18 | import subprocess |
| 19 | |
| 20 | import unittest |
| 21 | |
| 22 | from acloud.internal import constants |
| 23 | from acloud.internal.lib import driver_test_lib |
| 24 | from acloud.list import instance |
| 25 | |
| 26 | |
| 27 | class InstanceTest(driver_test_lib.BaseDriverTest): |
| 28 | """Test instance.""" |
| 29 | PS_SSH_TUNNEL = ("/fake_ps_1 --fake arg \n" |
| 30 | "/fake_ps_2 --fake arg \n" |
| 31 | "/usr/bin/ssh -i ~/.ssh/acloud_rsa " |
| 32 | "-o UserKnownHostsFile=/dev/null " |
| 33 | "-o StrictHostKeyChecking=no -L 12345:127.0.0.1:6444 " |
| 34 | "-L 54321:127.0.0.1:6520 -N -f -l user fake_ip") |
| 35 | PS_LAUNCH_CVD = ("Sat Nov 10 21:55:10 2018 /fake_path/bin/launch_cvd " |
| 36 | "--daemon --cpus 2 --x_res 1080 --y_res 1920 --dpi 480" |
| 37 | " --memory_mb 4096 --blank_data_image_mb 4096 --data_policy" |
| 38 | " always_create --system_image_dir /fake " |
| 39 | "--vnc_server_port 6444") |
| 40 | |
| 41 | # pylint: disable=protected-access |
| 42 | def testCreateLocalInstance(self): |
| 43 | """"Test get local instance info from launch_cvd process.""" |
| 44 | self.Patch(subprocess, "check_output", return_value=self.PS_LAUNCH_CVD) |
| 45 | local_instance = instance.LocalInstance() |
| 46 | self.assertEqual(constants.LOCAL_INS_NAME, local_instance.name) |
| 47 | self.assertEqual(True, local_instance.islocal) |
| 48 | self.assertEqual("1080x1920 (480)", local_instance.display) |
| 49 | self.assertEqual("Sat Nov 10 21:55:10 2018", local_instance.createtime) |
| 50 | expected_full_name = "device serial: 127.0.0.1:%s (%s)" % (constants.DEFAULT_ADB_PORT, |
| 51 | local_instance.name) |
| 52 | self.assertEqual(expected_full_name, local_instance.fullname) |
| 53 | |
| 54 | # test return None if no launch_cvd process found |
| 55 | self.Patch(subprocess, "check_output", return_value="no launch_cvd " |
| 56 | "found") |
| 57 | self.assertEqual(None, instance.LocalInstance()) |
| 58 | |
| 59 | # pylint: disable=protected-access |
| 60 | def testGetAdbVncPortFromSSHTunnel(self): |
| 61 | """"Test Get forwarding adb and vnc port from ssh tunnel.""" |
| 62 | self.Patch(subprocess, "check_output", return_value=self.PS_SSH_TUNNEL) |
| 63 | forwarded_ports = instance.RemoteInstance._GetAdbVncPortFromSSHTunnel("fake_ip") |
| 64 | self.assertEqual(54321, forwarded_ports.adb_port) |
| 65 | self.assertEqual(12345, forwarded_ports.vnc_port) |
| 66 | |
| 67 | # pylint: disable=protected-access |
| 68 | def testProcessGceInstance(self): |
| 69 | """"Test process instance detail.""" |
| 70 | gce_instance = { |
| 71 | constants.INS_KEY_NAME: "fake_ins_name", |
| 72 | constants.INS_KEY_CREATETIME: "fake_create_time", |
| 73 | constants.INS_KEY_STATUS: "fake_status", |
| 74 | "networkInterfaces": [{"accessConfigs": [{"natIP": "fake_ip"}]}], |
| 75 | "labels": {constants.INS_KEY_AVD_TYPE: "fake_type", |
| 76 | constants.INS_KEY_AVD_FLAVOR: "fake_flavor"}, |
| 77 | "metadata": {} |
| 78 | } |
| 79 | |
| 80 | fake_adb = 123456 |
| 81 | fake_vnc = 654321 |
| 82 | forwarded_ports = collections.namedtuple("ForwardedPorts", |
| 83 | [constants.VNC_PORT, |
| 84 | constants.ADB_PORT]) |
| 85 | self.Patch( |
| 86 | instance.RemoteInstance, |
| 87 | "_GetAdbVncPortFromSSHTunnel", |
| 88 | return_value=forwarded_ports(vnc_port=fake_vnc, adb_port=fake_adb)) |
| 89 | |
| 90 | # test is_connected will be true if ssh tunnel connection is found |
| 91 | instance_info = instance.RemoteInstance(gce_instance) |
| 92 | self.assertTrue(instance_info.is_connected) |
| 93 | self.assertEqual(instance_info.forwarding_adb_port, fake_adb) |
| 94 | self.assertEqual(instance_info.forwarding_vnc_port, fake_vnc) |
| 95 | expected_full_name = "device serial: 127.0.0.1:%s (%s)" % (fake_adb, |
| 96 | instance_info.name) |
| 97 | self.assertEqual(expected_full_name, instance_info.fullname) |
| 98 | |
| 99 | # test is_connected will be false if ssh tunnel connection is not found |
| 100 | self.Patch( |
| 101 | instance.RemoteInstance, |
| 102 | "_GetAdbVncPortFromSSHTunnel", |
| 103 | return_value=forwarded_ports(vnc_port=None, adb_port=None)) |
| 104 | instance_info = instance.RemoteInstance(gce_instance) |
| 105 | self.assertFalse(instance_info.is_connected) |
| 106 | expected_full_name = ("device serial: not connected (%s)" % |
| 107 | instance_info.name) |
| 108 | self.assertEqual(expected_full_name, instance_info.fullname) |
| 109 | |
| 110 | |
| 111 | if __name__ == "__main__": |
| 112 | unittest.main() |