#!/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()