• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
24
25# pylint: disable=import-error
26import dateutil.parser
27import dateutil.tz
28
29from acloud.internal import constants
30from acloud.internal.lib import driver_test_lib
31from acloud.internal.lib.adb_tools import AdbTools
32from acloud.list import instance
33
34
35class InstanceTest(driver_test_lib.BaseDriverTest):
36    """Test instance."""
37    PS_SSH_TUNNEL = ("/fake_ps_1 --fake arg \n"
38                     "/fake_ps_2 --fake arg \n"
39                     "/usr/bin/ssh -i ~/.ssh/acloud_rsa "
40                     "-o UserKnownHostsFile=/dev/null "
41                     "-o StrictHostKeyChecking=no -L 12345:127.0.0.1:6444 "
42                     "-L 54321:127.0.0.1:6520 -N -f -l user fake_ip")
43    PS_LAUNCH_CVD = ("Sat Nov 10 21:55:10 2018 /fake_path/bin/launch_cvd "
44                     "--daemon --cpus 2 --x_res 1080 --y_res 1920 --dpi 480"
45                     " --memory_mb 4096 --blank_data_image_mb 4096 --data_policy"
46                     " always_create --system_image_dir /fake "
47                     "--vnc_server_port 6444")
48
49    # pylint: disable=protected-access
50    def testCreateLocalInstance(self):
51        """"Test get local instance info from launch_cvd process."""
52        self.Patch(subprocess, "check_output", return_value=self.PS_LAUNCH_CVD)
53        self.Patch(instance, "_GetElapsedTime", return_value="fake_time")
54        local_instance = instance.LocalInstance()
55        self.assertEqual(constants.LOCAL_INS_NAME, local_instance.name)
56        self.assertEqual(True, local_instance.islocal)
57        self.assertEqual("1080x1920 (480)", local_instance.display)
58        self.assertEqual("Sat Nov 10 21:55:10 2018", local_instance.createtime)
59        expected_full_name = "device serial: 127.0.0.1:%s (%s) elapsed time: %s" % (
60            constants.CF_ADB_PORT, constants.LOCAL_INS_NAME, "fake_time")
61        self.assertEqual(expected_full_name, local_instance.fullname)
62
63        # test return None if no launch_cvd process found
64        self.Patch(subprocess, "check_output", return_value="no launch_cvd "
65                                                            "found")
66        self.assertEqual(None, instance.LocalInstance())
67
68    def testGetElapsedTime(self):
69        """Test _GetElapsedTime"""
70        # Instance time can't parse
71        start_time = "error time"
72        self.assertEqual(instance._MSG_UNABLE_TO_CALCULATE,
73                         instance._GetElapsedTime(start_time))
74
75        # Remote instance elapsed time
76        now = "2019-01-14T13:00:00.000-07:00"
77        start_time = "2019-01-14T03:00:00.000-07:00"
78        self.Patch(instance, "datetime")
79        instance.datetime.datetime.now.return_value = dateutil.parser.parse(now)
80        self.assertEqual(
81            datetime.timedelta(hours=10), instance._GetElapsedTime(start_time))
82
83        # Local instance elapsed time
84        now = "Mon Jan 14 10:10:10 2019"
85        start_time = "Mon Jan 14 08:10:10 2019"
86        instance.datetime.datetime.now.return_value = dateutil.parser.parse(
87            now).replace(tzinfo=dateutil.tz.tzlocal())
88        self.assertEqual(
89            datetime.timedelta(hours=2), instance._GetElapsedTime(start_time))
90
91    # pylint: disable=protected-access
92    def testGetAdbVncPortFromSSHTunnel(self):
93        """"Test Get forwarding adb and vnc port from ssh tunnel."""
94        self.Patch(subprocess, "check_output", return_value=self.PS_SSH_TUNNEL)
95        self.Patch(instance, "_GetElapsedTime", return_value="fake_time")
96        forwarded_ports = instance.RemoteInstance(
97            mock.MagicMock()).GetAdbVncPortFromSSHTunnel(
98                "fake_ip", constants.TYPE_CF)
99        self.assertEqual(54321, forwarded_ports.adb_port)
100        self.assertEqual(12345, forwarded_ports.vnc_port)
101
102    # pylint: disable=protected-access
103    def testProcessGceInstance(self):
104        """"Test process instance detail."""
105        gce_instance = {
106            constants.INS_KEY_NAME: "fake_ins_name",
107            constants.INS_KEY_CREATETIME: "fake_create_time",
108            constants.INS_KEY_STATUS: "fake_status",
109            "networkInterfaces": [{"accessConfigs": [{"natIP": "fake_ip"}]}],
110            "labels": {constants.INS_KEY_AVD_TYPE: "fake_type",
111                       constants.INS_KEY_AVD_FLAVOR: "fake_flavor"},
112            "metadata": {}
113        }
114
115        fake_adb = 123456
116        fake_vnc = 654321
117        forwarded_ports = collections.namedtuple("ForwardedPorts",
118                                                 [constants.VNC_PORT,
119                                                  constants.ADB_PORT])
120        self.Patch(
121            instance.RemoteInstance,
122            "GetAdbVncPortFromSSHTunnel",
123            return_value=forwarded_ports(vnc_port=fake_vnc, adb_port=fake_adb))
124        self.Patch(instance, "_GetElapsedTime", return_value="fake_time")
125        self.Patch(AdbTools, "IsAdbConnected", return_value=True)
126
127        # test ssh_tunnel_is_connected will be true if ssh tunnel connection is found
128        instance_info = instance.RemoteInstance(gce_instance)
129        self.assertTrue(instance_info.ssh_tunnel_is_connected)
130        self.assertEqual(instance_info.forwarding_adb_port, fake_adb)
131        self.assertEqual(instance_info.forwarding_vnc_port, fake_vnc)
132        expected_full_name = "device serial: 127.0.0.1:%s (%s) elapsed time: %s" % (
133            fake_adb, gce_instance[constants.INS_KEY_NAME], "fake_time")
134        self.assertEqual(expected_full_name, instance_info.fullname)
135
136        # test ssh tunnel is connected but adb is disconnected
137        self.Patch(AdbTools, "IsAdbConnected", return_value=False)
138        instance_info = instance.RemoteInstance(gce_instance)
139        self.assertTrue(instance_info.ssh_tunnel_is_connected)
140        expected_full_name = "device serial: not connected (%s) elapsed time: %s" % (
141            instance_info.name, "fake_time")
142        self.assertEqual(expected_full_name, instance_info.fullname)
143
144        # test ssh_tunnel_is_connected will be false if ssh tunnel connection is not found
145        self.Patch(
146            instance.RemoteInstance,
147            "GetAdbVncPortFromSSHTunnel",
148            return_value=forwarded_ports(vnc_port=None, adb_port=None))
149        instance_info = instance.RemoteInstance(gce_instance)
150        self.assertFalse(instance_info.ssh_tunnel_is_connected)
151        expected_full_name = "device serial: not connected (%s) elapsed time: %s" % (
152            gce_instance[constants.INS_KEY_NAME], "fake_time")
153        self.assertEqual(expected_full_name, instance_info.fullname)
154
155
156if __name__ == "__main__":
157    unittest.main()
158