#!/usr/bin/env 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 instance class."""
|
|
import collections
|
import datetime
|
import subprocess
|
|
import unittest
|
import mock
|
|
# pylint: disable=import-error
|
import dateutil.parser
|
import dateutil.tz
|
|
from acloud.internal import constants
|
from acloud.internal.lib import driver_test_lib
|
from acloud.internal.lib.adb_tools import AdbTools
|
from acloud.list import instance
|
|
|
class InstanceTest(driver_test_lib.BaseDriverTest):
|
"""Test instance."""
|
PS_SSH_TUNNEL = ("/fake_ps_1 --fake arg \n"
|
"/fake_ps_2 --fake arg \n"
|
"/usr/bin/ssh -i ~/.ssh/acloud_rsa "
|
"-o UserKnownHostsFile=/dev/null "
|
"-o StrictHostKeyChecking=no -L 12345:127.0.0.1:6444 "
|
"-L 54321:127.0.0.1:6520 -N -f -l user fake_ip")
|
PS_LAUNCH_CVD = ("Sat Nov 10 21:55:10 2018 /fake_path/bin/launch_cvd "
|
"--daemon --cpus 2 --x_res 1080 --y_res 1920 --dpi 480"
|
" --memory_mb 4096 --blank_data_image_mb 4096 --data_policy"
|
" always_create --system_image_dir /fake "
|
"--vnc_server_port 6444")
|
|
# pylint: disable=protected-access
|
def testCreateLocalInstance(self):
|
""""Test get local instance info from launch_cvd process."""
|
self.Patch(subprocess, "check_output", return_value=self.PS_LAUNCH_CVD)
|
self.Patch(instance, "_GetElapsedTime", return_value="fake_time")
|
local_instance = instance.LocalInstance()
|
self.assertEqual(constants.LOCAL_INS_NAME, local_instance.name)
|
self.assertEqual(True, local_instance.islocal)
|
self.assertEqual("1080x1920 (480)", local_instance.display)
|
self.assertEqual("Sat Nov 10 21:55:10 2018", local_instance.createtime)
|
expected_full_name = "device serial: 127.0.0.1:%s (%s) elapsed time: %s" % (
|
constants.CF_ADB_PORT, constants.LOCAL_INS_NAME, "fake_time")
|
self.assertEqual(expected_full_name, local_instance.fullname)
|
|
# test return None if no launch_cvd process found
|
self.Patch(subprocess, "check_output", return_value="no launch_cvd "
|
"found")
|
self.assertEqual(None, instance.LocalInstance())
|
|
def testGetElapsedTime(self):
|
"""Test _GetElapsedTime"""
|
# Instance time can't parse
|
start_time = "error time"
|
self.assertEqual(instance._MSG_UNABLE_TO_CALCULATE,
|
instance._GetElapsedTime(start_time))
|
|
# Remote instance elapsed time
|
now = "2019-01-14T13:00:00.000-07:00"
|
start_time = "2019-01-14T03:00:00.000-07:00"
|
self.Patch(instance, "datetime")
|
instance.datetime.datetime.now.return_value = dateutil.parser.parse(now)
|
self.assertEqual(
|
datetime.timedelta(hours=10), instance._GetElapsedTime(start_time))
|
|
# Local instance elapsed time
|
now = "Mon Jan 14 10:10:10 2019"
|
start_time = "Mon Jan 14 08:10:10 2019"
|
instance.datetime.datetime.now.return_value = dateutil.parser.parse(
|
now).replace(tzinfo=dateutil.tz.tzlocal())
|
self.assertEqual(
|
datetime.timedelta(hours=2), instance._GetElapsedTime(start_time))
|
|
# pylint: disable=protected-access
|
def testGetAdbVncPortFromSSHTunnel(self):
|
""""Test Get forwarding adb and vnc port from ssh tunnel."""
|
self.Patch(subprocess, "check_output", return_value=self.PS_SSH_TUNNEL)
|
self.Patch(instance, "_GetElapsedTime", return_value="fake_time")
|
forwarded_ports = instance.RemoteInstance(
|
mock.MagicMock()).GetAdbVncPortFromSSHTunnel(
|
"fake_ip", constants.TYPE_CF)
|
self.assertEqual(54321, forwarded_ports.adb_port)
|
self.assertEqual(12345, forwarded_ports.vnc_port)
|
|
# pylint: disable=protected-access
|
def testProcessGceInstance(self):
|
""""Test process instance detail."""
|
gce_instance = {
|
constants.INS_KEY_NAME: "fake_ins_name",
|
constants.INS_KEY_CREATETIME: "fake_create_time",
|
constants.INS_KEY_STATUS: "fake_status",
|
"networkInterfaces": [{"accessConfigs": [{"natIP": "fake_ip"}]}],
|
"labels": {constants.INS_KEY_AVD_TYPE: "fake_type",
|
constants.INS_KEY_AVD_FLAVOR: "fake_flavor"},
|
"metadata": {}
|
}
|
|
fake_adb = 123456
|
fake_vnc = 654321
|
forwarded_ports = collections.namedtuple("ForwardedPorts",
|
[constants.VNC_PORT,
|
constants.ADB_PORT])
|
self.Patch(
|
instance.RemoteInstance,
|
"GetAdbVncPortFromSSHTunnel",
|
return_value=forwarded_ports(vnc_port=fake_vnc, adb_port=fake_adb))
|
self.Patch(instance, "_GetElapsedTime", return_value="fake_time")
|
self.Patch(AdbTools, "IsAdbConnected", return_value=True)
|
|
# test ssh_tunnel_is_connected will be true if ssh tunnel connection is found
|
instance_info = instance.RemoteInstance(gce_instance)
|
self.assertTrue(instance_info.ssh_tunnel_is_connected)
|
self.assertEqual(instance_info.forwarding_adb_port, fake_adb)
|
self.assertEqual(instance_info.forwarding_vnc_port, fake_vnc)
|
expected_full_name = "device serial: 127.0.0.1:%s (%s) elapsed time: %s" % (
|
fake_adb, gce_instance[constants.INS_KEY_NAME], "fake_time")
|
self.assertEqual(expected_full_name, instance_info.fullname)
|
|
# test ssh tunnel is connected but adb is disconnected
|
self.Patch(AdbTools, "IsAdbConnected", return_value=False)
|
instance_info = instance.RemoteInstance(gce_instance)
|
self.assertTrue(instance_info.ssh_tunnel_is_connected)
|
expected_full_name = "device serial: not connected (%s) elapsed time: %s" % (
|
instance_info.name, "fake_time")
|
self.assertEqual(expected_full_name, instance_info.fullname)
|
|
# test ssh_tunnel_is_connected will be false if ssh tunnel connection is not found
|
self.Patch(
|
instance.RemoteInstance,
|
"GetAdbVncPortFromSSHTunnel",
|
return_value=forwarded_ports(vnc_port=None, adb_port=None))
|
instance_info = instance.RemoteInstance(gce_instance)
|
self.assertFalse(instance_info.ssh_tunnel_is_connected)
|
expected_full_name = "device serial: not connected (%s) elapsed time: %s" % (
|
gce_instance[constants.INS_KEY_NAME], "fake_time")
|
self.assertEqual(expected_full_name, instance_info.fullname)
|
|
|
if __name__ == "__main__":
|
unittest.main()
|
