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 instance class."""
17
18import collections
19import datetime
20import subprocess
21
22import unittest
23import mock
24from six import b
25
26# pylint: disable=import-error
27import dateutil.parser
28import dateutil.tz
29
30from acloud.internal import constants
31from acloud.internal.lib import cvd_runtime_config
32from acloud.internal.lib import driver_test_lib
33from acloud.internal.lib.adb_tools import AdbTools
34from acloud.list import instance
35
36
37class InstanceTest(driver_test_lib.BaseDriverTest):
38    """Test instance."""
39    PS_SSH_TUNNEL = b("/fake_ps_1 --fake arg \n"
40                      "/fake_ps_2 --fake arg \n"
41                      "/usr/bin/ssh -i ~/.ssh/acloud_rsa "
42                      "-o UserKnownHostsFile=/dev/null "
43                      "-o StrictHostKeyChecking=no -L 12345:127.0.0.1:6444 "
44                      "-L 54321:127.0.0.1:6520 -N -f -l user 1.1.1.1")
45    PS_LAUNCH_CVD = b("Sat Nov 10 21:55:10 2018 /fake_path/bin/run_cvd ")
46    PS_RUNTIME_CF_CONFIG = {"x_res": "1080", "y_res": "1920", "dpi": "480"}
47    GCE_INSTANCE = {
48        constants.INS_KEY_NAME: "fake_ins_name",
49        constants.INS_KEY_CREATETIME: "fake_create_time",
50        constants.INS_KEY_STATUS: "fake_status",
51        constants.INS_KEY_ZONE: "test/zones/fake_zone",
52        "networkInterfaces": [{"accessConfigs": [{"natIP": "1.1.1.1"}]}],
53        "labels": {constants.INS_KEY_AVD_TYPE: "fake_type",
54                   constants.INS_KEY_AVD_FLAVOR: "fake_flavor"},
55        "metadata": {
56            "items":[{"key":constants.INS_KEY_AVD_TYPE,
57                      "value":"fake_type"},
58                     {"key":constants.INS_KEY_AVD_FLAVOR,
59                      "value":"fake_flavor"}]}
60    }
61
62    # pylint: disable=protected-access
63    def testCreateLocalInstance(self):
64        """"Test get local instance info from launch_cvd process."""
65        self.Patch(subprocess, "check_output", return_value=self.PS_LAUNCH_CVD)
66        cf_config = mock.MagicMock(
67            instance_id=2,
68            x_res=1080,
69            y_res=1920,
70            dpi=480,
71            instance_dir="fake_instance_dir",
72            adb_port=6521,
73            vnc_port=6445,
74            adb_ip_port="127.0.0.1:6521",
75        )
76        self.Patch(cvd_runtime_config, "CvdRuntimeConfig",
77                   return_value=cf_config)
78        local_instance = instance.LocalInstance(cf_config)
79
80        self.assertEqual(constants.LOCAL_INS_NAME + "-2", local_instance.name)
81        self.assertEqual(True, local_instance.islocal)
82        self.assertEqual("1080x1920 (480)", local_instance.display)
83        expected_full_name = ("device serial: 127.0.0.1:%s (%s) elapsed time: %s"
84                              % ("6521",
85                                 constants.LOCAL_INS_NAME + "-2",
86                                 "None"))
87        self.assertEqual(expected_full_name, local_instance.fullname)
88        self.assertEqual(6521, local_instance.adb_port)
89        self.assertEqual(6445, local_instance.vnc_port)
90
91    @mock.patch("acloud.list.instance.tempfile")
92    @mock.patch("acloud.list.instance.AdbTools")
93    def testCreateLocalGoldfishInstance(self, mock_adb_tools, mock_tempfile):
94        """"Test the attributes of LocalGoldfishInstance."""
95        mock_tempfile.gettempdir.return_value = "/unit/test"
96        mock_adb_tools.return_value = mock.Mock(device_information={})
97
98        inst = instance.LocalGoldfishInstance(1)
99
100        self.assertEqual(inst.name, "local-goldfish-instance-1")
101        self.assertEqual(inst.avd_type, constants.TYPE_GF)
102        self.assertEqual(inst.adb_port, 5555)
103        self.assertTrue(inst.islocal)
104        self.assertEqual(inst.console_port, 5554)
105        self.assertEqual(inst.device_serial, "emulator-5554")
106        self.assertEqual(inst.instance_dir,
107                         "/unit/test/acloud_gf_temp/local-goldfish-instance-1")
108
109    @mock.patch("acloud.list.instance.open",
110                mock.mock_open(read_data="test createtime"))
111    @mock.patch("acloud.list.instance.os.path.isfile")
112    @mock.patch("acloud.list.instance.os.listdir")
113    @mock.patch("acloud.list.instance.os.path.isdir")
114    @mock.patch("acloud.list.instance.tempfile")
115    @mock.patch("acloud.list.instance.AdbTools")
116    @mock.patch("acloud.list.instance._GetElapsedTime")
117    def testGetLocalGoldfishInstances(self, mock_get_elapsed_time,
118                                      mock_adb_tools, mock_tempfile,
119                                      mock_isdir, mock_listdir, mock_isfile):
120        """Test LocalGoldfishInstance.GetExistingInstances."""
121        mock_get_elapsed_time.return_value = datetime.timedelta(hours=10)
122        mock_adb_tools.return_value = mock.Mock(device_information={})
123        mock_tempfile.gettempdir.return_value = "/unit/test"
124        acloud_gf_temp_path = "/unit/test/acloud_gf_temp"
125        subdir_names = (
126            "local-goldfish-instance-1",
127            "local-goldfish-instance-2",
128            "local-goldfish-instance-3")
129        timestamp_paths = (
130            "/unit/test/acloud_gf_temp/local-goldfish-instance-1/"
131            "creation_timestamp.txt",
132            "/unit/test/acloud_gf_temp/local-goldfish-instance-2/"
133            "creation_timestamp.txt",
134            "/unit/test/acloud_gf_temp/local-goldfish-instance-3/"
135            "creation_timestamp.txt")
136        mock_isdir.side_effect = lambda path: path == acloud_gf_temp_path
137        mock_listdir.side_effect = lambda path: (
138            subdir_names if path == acloud_gf_temp_path else [])
139        mock_isfile.side_effect = lambda path: (
140            path in (timestamp_paths[0], timestamp_paths[2]))
141
142        instances = instance.LocalGoldfishInstance.GetExistingInstances()
143
144        mock_isdir.assert_called_with(acloud_gf_temp_path)
145        mock_listdir.assert_called_with(acloud_gf_temp_path)
146        for timestamp_path in timestamp_paths:
147            mock_isfile.assert_any_call(timestamp_path)
148        self.assertEqual(len(instances), 2)
149        self.assertEqual(instances[0].console_port, 5554)
150        self.assertEqual(instances[0].createtime, "test createtime")
151        self.assertEqual(instances[0].fullname,
152                         "device serial: emulator-5554 "
153                         "(local-goldfish-instance-1) "
154                         "elapsed time: 10:00:00")
155        self.assertEqual(instances[1].console_port, 5558)
156        self.assertEqual(instances[1].createtime, "test createtime")
157        self.assertEqual(instances[1].fullname,
158                         "device serial: emulator-5558 "
159                         "(local-goldfish-instance-3) "
160                         "elapsed time: 10:00:00")
161
162    def testGetElapsedTime(self):
163        """Test _GetElapsedTime"""
164        # Instance time can't parse
165        start_time = "error time"
166        self.assertEqual(instance._MSG_UNABLE_TO_CALCULATE,
167                         instance._GetElapsedTime(start_time))
168
169        # Remote instance elapsed time
170        now = "2019-01-14T13:00:00.000-07:00"
171        start_time = "2019-01-14T03:00:00.000-07:00"
172        self.Patch(instance, "datetime")
173        instance.datetime.datetime.now.return_value = dateutil.parser.parse(now)
174        self.assertEqual(
175            datetime.timedelta(hours=10), instance._GetElapsedTime(start_time))
176
177        # Local instance elapsed time
178        now = "Mon Jan 14 10:10:10 2019"
179        start_time = "Mon Jan 14 08:10:10 2019"
180        instance.datetime.datetime.now.return_value = dateutil.parser.parse(
181            now).replace(tzinfo=dateutil.tz.tzlocal())
182        self.assertEqual(
183            datetime.timedelta(hours=2), instance._GetElapsedTime(start_time))
184
185    # pylint: disable=protected-access
186    def testGetAdbVncPortFromSSHTunnel(self):
187        """"Test Get forwarding adb and vnc port from ssh tunnel."""
188        self.Patch(subprocess, "check_output", return_value=self.PS_SSH_TUNNEL)
189        self.Patch(instance, "_GetElapsedTime", return_value="fake_time")
190        self.Patch(instance.RemoteInstance, "_GetZoneName", return_value="fake_zone")
191        forwarded_ports = instance.RemoteInstance(
192            mock.MagicMock()).GetAdbVncPortFromSSHTunnel(
193                "1.1.1.1", constants.TYPE_CF)
194        self.assertEqual(54321, forwarded_ports.adb_port)
195        self.assertEqual(12345, forwarded_ports.vnc_port)
196
197        # If avd_type is undefined in utils.AVD_PORT_DICT.
198        forwarded_ports = instance.RemoteInstance(
199            mock.MagicMock()).GetAdbVncPortFromSSHTunnel(
200                "1.1.1.1", "undefined_avd_type")
201        self.assertEqual(None, forwarded_ports.adb_port)
202        self.assertEqual(None, forwarded_ports.vnc_port)
203
204    # pylint: disable=protected-access
205    def testProcessGceInstance(self):
206        """"Test process instance detail."""
207        fake_adb = 123456
208        fake_vnc = 654321
209        forwarded_ports = collections.namedtuple("ForwardedPorts",
210                                                 [constants.VNC_PORT,
211                                                  constants.ADB_PORT])
212        self.Patch(
213            instance.RemoteInstance,
214            "GetAdbVncPortFromSSHTunnel",
215            return_value=forwarded_ports(vnc_port=fake_vnc, adb_port=fake_adb))
216        self.Patch(instance, "_GetElapsedTime", return_value="fake_time")
217        self.Patch(AdbTools, "IsAdbConnected", return_value=True)
218
219        # test ssh_tunnel_is_connected will be true if ssh tunnel connection is found
220        instance_info = instance.RemoteInstance(self.GCE_INSTANCE)
221        self.assertTrue(instance_info.ssh_tunnel_is_connected)
222        self.assertEqual(instance_info.adb_port, fake_adb)
223        self.assertEqual(instance_info.vnc_port, fake_vnc)
224        self.assertEqual("1.1.1.1", instance_info.ip)
225        self.assertEqual("fake_status", instance_info.status)
226        self.assertEqual("fake_type", instance_info.avd_type)
227        self.assertEqual("fake_flavor", instance_info.avd_flavor)
228        expected_full_name = "device serial: 127.0.0.1:%s (%s) elapsed time: %s" % (
229            fake_adb, self.GCE_INSTANCE[constants.INS_KEY_NAME], "fake_time")
230        self.assertEqual(expected_full_name, instance_info.fullname)
231
232        # test ssh tunnel is connected but adb is disconnected
233        self.Patch(AdbTools, "IsAdbConnected", return_value=False)
234        instance_info = instance.RemoteInstance(self.GCE_INSTANCE)
235        self.assertTrue(instance_info.ssh_tunnel_is_connected)
236        expected_full_name = "device serial: not connected (%s) elapsed time: %s" % (
237            instance_info.name, "fake_time")
238        self.assertEqual(expected_full_name, instance_info.fullname)
239
240        # test ssh_tunnel_is_connected will be false if ssh tunnel connection is not found
241        self.Patch(
242            instance.RemoteInstance,
243            "GetAdbVncPortFromSSHTunnel",
244            return_value=forwarded_ports(vnc_port=None, adb_port=None))
245        instance_info = instance.RemoteInstance(self.GCE_INSTANCE)
246        self.assertFalse(instance_info.ssh_tunnel_is_connected)
247        expected_full_name = "device serial: not connected (%s) elapsed time: %s" % (
248            self.GCE_INSTANCE[constants.INS_KEY_NAME], "fake_time")
249        self.assertEqual(expected_full_name, instance_info.fullname)
250
251    def testInstanceSummary(self):
252        """Test instance summary."""
253        fake_adb = 123456
254        fake_vnc = 654321
255        forwarded_ports = collections.namedtuple("ForwardedPorts",
256                                                 [constants.VNC_PORT,
257                                                  constants.ADB_PORT])
258        self.Patch(
259            instance.RemoteInstance,
260            "GetAdbVncPortFromSSHTunnel",
261            return_value=forwarded_ports(vnc_port=fake_vnc, adb_port=fake_adb))
262        self.Patch(instance, "_GetElapsedTime", return_value="fake_time")
263        self.Patch(AdbTools, "IsAdbConnected", return_value=True)
264        remote_instance = instance.RemoteInstance(self.GCE_INSTANCE)
265        result_summary = (" name: fake_ins_name\n "
266                          "   IP: 1.1.1.1\n "
267                          "   create time: fake_create_time\n "
268                          "   elapse time: fake_time\n "
269                          "   status: fake_status\n "
270                          "   avd type: fake_type\n "
271                          "   display: None\n "
272                          "   vnc: 127.0.0.1:654321\n "
273                          "   zone: fake_zone\n "
274                          "   adb serial: 127.0.0.1:123456\n "
275                          "   product: None\n "
276                          "   model: None\n "
277                          "   device: None\n "
278                          "   transport_id: None")
279        self.assertEqual(remote_instance.Summary(), result_summary)
280
281        self.Patch(
282            instance.RemoteInstance,
283            "GetAdbVncPortFromSSHTunnel",
284            return_value=forwarded_ports(vnc_port=None, adb_port=None))
285        self.Patch(instance, "_GetElapsedTime", return_value="fake_time")
286        self.Patch(AdbTools, "IsAdbConnected", return_value=False)
287        remote_instance = instance.RemoteInstance(self.GCE_INSTANCE)
288        result_summary = (" name: fake_ins_name\n "
289                          "   IP: 1.1.1.1\n "
290                          "   create time: fake_create_time\n "
291                          "   elapse time: fake_time\n "
292                          "   status: fake_status\n "
293                          "   avd type: fake_type\n "
294                          "   display: None\n "
295                          "   vnc: 127.0.0.1:None\n "
296                          "   zone: fake_zone\n "
297                          "   adb serial: disconnected")
298        self.assertEqual(remote_instance.Summary(), result_summary)
299
300    def testGetZoneName(self):
301        """Test GetZoneName."""
302        zone_info = "v1/projects/project/zones/us-central1-c"
303        expected_result = "us-central1-c"
304        self.assertEqual(instance.RemoteInstance._GetZoneName(zone_info),
305                         expected_result)
306        # Test can't get zone name from zone info.
307        zone_info = "v1/projects/project/us-central1-c"
308        self.assertEqual(instance.RemoteInstance._GetZoneName(zone_info), None)
309
310
311if __name__ == "__main__":
312    unittest.main()
313