setup/gcp_setup_runner_test.py

#!/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 acloud.setup.gcp_setup_runner."""

import unittest
import os

from unittest import mock
import six

# pylint: disable=no-name-in-module,import-error,no-member
from acloud import errors
from acloud.internal.lib import utils
from acloud.internal.proto import user_config_pb2
from acloud.public import config
from acloud.setup import gcp_setup_runner

_GCP_USER_CONFIG = """
[compute]
region = new_region
zone = new_zone
[core]
account = new@google.com
disable_usage_reporting = False
project = new_project
"""

def _CreateCfgFile():
    """A helper method that creates a mock configuration object."""
    default_cfg = """
project: "fake_project"
zone: "fake_zone"
storage_bucket_name: "fake_bucket"
client_id: "fake_client_id"
client_secret: "fake_client_secret"
"""
    return default_cfg


# pylint: disable=protected-access
class AcloudGCPSetupTest(unittest.TestCase):
    """Test GCP Setup steps."""

    def setUp(self):
        """Create config and gcp_env_runner."""
        self.cfg_path = "acloud_unittest.config"
        file_write = open(self.cfg_path, 'w')
        file_write.write(_CreateCfgFile().strip())
        file_write.close()
        self.gcp_env_runner = gcp_setup_runner.GcpTaskRunner(self.cfg_path)
        self.gcloud_runner = gcp_setup_runner.GoogleSDKBins("")

    def tearDown(self):
        """Remove temp file."""
        if os.path.isfile(self.cfg_path):
            os.remove(self.cfg_path)

    def testUpdateConfigFile(self):
        """Test update config file."""
        # Test update project field.
        gcp_setup_runner.UpdateConfigFile(self.cfg_path, "project",
                                          "test_project")
        cfg = config.AcloudConfigManager.LoadConfigFromProtocolBuffer(
            open(self.cfg_path, "r"), user_config_pb2.UserConfig)
        self.assertEqual(cfg.project, "test_project")
        self.assertEqual(cfg.ssh_private_key_path, "")
        # Test add ssh key path in config.
        gcp_setup_runner.UpdateConfigFile(self.cfg_path,
                                          "ssh_private_key_path", "test_path")
        cfg = config.AcloudConfigManager.LoadConfigFromProtocolBuffer(
            open(self.cfg_path, "r"), user_config_pb2.UserConfig)
        self.assertEqual(cfg.project, "test_project")
        self.assertEqual(cfg.ssh_private_key_path, "test_path")
        # Test config is not a file
        with mock.patch("os.path.isfile") as chkfile:
            chkfile.return_value = False
            gcp_setup_runner.UpdateConfigFile(self.cfg_path, "project",
                                              "test_project")
            cfg = config.AcloudConfigManager.LoadConfigFromProtocolBuffer(
                open(self.cfg_path, "r"), user_config_pb2.UserConfig)
            self.assertEqual(cfg.project, "test_project")

    @mock.patch("os.path.dirname", return_value="")
    @mock.patch.object(utils, "CheckOutput")
    def testSeupProjectZone(self, mock_runner, mock_path):
        """Test setup project and zone."""
        gcloud_runner = gcp_setup_runner.GoogleSDKBins(mock_path)
        self.gcp_env_runner.project = "fake_project"
        self.gcp_env_runner.zone = "fake_zone"
        mock_runner.side_effect = [0, _GCP_USER_CONFIG]
        self.gcp_env_runner._UpdateProject(gcloud_runner)
        self.assertEqual(self.gcp_env_runner.project, "new_project")
        self.assertEqual(self.gcp_env_runner.zone, "new_zone")

    @mock.patch.object(six.moves, "input")
    def testSetupClientIDSecret(self, mock_id):
        """Test setup client ID and client secret."""
        self.gcp_env_runner.client_id = "fake_client_id"
        self.gcp_env_runner.client_secret = "fake_client_secret"
        mock_id.side_effect = ["new_id", "new_secret"]
        self.gcp_env_runner._SetupClientIDSecret()
        self.assertEqual(self.gcp_env_runner.client_id, "new_id")
        self.assertEqual(self.gcp_env_runner.client_secret, "new_secret")

    @mock.patch.object(gcp_setup_runner, "UpdateConfigFile")
    @mock.patch.object(utils, "CreateSshKeyPairIfNotExist")
    def testSetupSSHKeys(self, mock_check, mock_update):
        """Test setup the pair of the ssh key for acloud.config."""
        # Test ssh key has already setup
        gcp_setup_runner.SetupSSHKeys(self.cfg_path,
                                      "fake_private_key_path",
                                      "fake_public_key_path")
        self.assertEqual(mock_update.call_count, 0)
        # Test if private_key_path is empty string
        with mock.patch('os.path.expanduser') as ssh_path:
            ssh_path.return_value = ""
            gcp_setup_runner.SetupSSHKeys(self.cfg_path,
                                          "fake_private_key_path",
                                          "fake_public_key_path")
            mock_check.assert_called_once_with(
                gcp_setup_runner._DEFAULT_SSH_PRIVATE_KEY,
                gcp_setup_runner._DEFAULT_SSH_PUBLIC_KEY)

            mock_update.assert_has_calls([
                mock.call(self.cfg_path, "ssh_private_key_path",
                          gcp_setup_runner._DEFAULT_SSH_PRIVATE_KEY),
                mock.call(self.cfg_path, "ssh_public_key_path",
                          gcp_setup_runner._DEFAULT_SSH_PUBLIC_KEY)])

    @mock.patch.object(gcp_setup_runner.GcpTaskRunner, "_CreateStableHostImage")
    @mock.patch.object(gcp_setup_runner.GcpTaskRunner, "_EnableGcloudServices")
    @mock.patch.object(gcp_setup_runner.GcpTaskRunner, "_SetupProject")
    @mock.patch.object(gcp_setup_runner, "GoogleSDKBins")
    def testSetupGcloudInfo(self, mock_sdk, mock_set, mock_run, mock_create):
        """test setup gcloud info"""
        with mock.patch("acloud.setup.google_sdk.GoogleSDK"):
            self.gcp_env_runner._SetupGcloudInfo()
            mock_sdk.assert_called_once()
            mock_set.assert_called_once()
            mock_run.assert_called_once()
            mock_create.assert_called_once()

    @mock.patch.object(gcp_setup_runner, "UpdateConfigFile")
    def testCreateStableHostImage(self, mock_update):
        """test create stable hostimage."""
        # Test no need to create stable hose image name.
        self.gcp_env_runner.stable_host_image_name = "fake_host_image_name"
        self.gcp_env_runner._CreateStableHostImage()
        self.assertEqual(mock_update.call_count, 0)
        # Test need to reset stable hose image name.
        self.gcp_env_runner.stable_host_image_name = ""
        self.gcp_env_runner._CreateStableHostImage()
        self.assertEqual(mock_update.call_count, 1)

    @mock.patch.object(gcp_setup_runner.GcpTaskRunner, "_CheckBillingEnable")
    @mock.patch.object(gcp_setup_runner.GcpTaskRunner, "_NeedProjectSetup")
    @mock.patch.object(gcp_setup_runner.GcpTaskRunner, "_SetupClientIDSecret")
    @mock.patch.object(gcp_setup_runner.GcpTaskRunner, "_UpdateProject")
    def testSetupProjectNoChange(self, mock_setproj, mock_setid,
                                 mock_chkproj, mock_check_billing):
        """test setup project and project not be changed."""
        # Test project didn't change, and no need to setup client id/secret
        mock_chkproj.return_value = False
        self.gcp_env_runner.client_id = "test_client_id"
        self.gcp_env_runner._SetupProject(self.gcloud_runner)
        self.assertEqual(mock_setproj.call_count, 0)
        self.assertEqual(mock_setid.call_count, 0)
        mock_check_billing.assert_called_once()
        # Test project didn't change, but client_id is empty
        self.gcp_env_runner.client_id = ""
        self.gcp_env_runner._SetupProject(self.gcloud_runner)
        self.assertEqual(mock_setproj.call_count, 0)
        mock_setid.assert_called_once()
        self.assertEqual(mock_check_billing.call_count, 2)

    @mock.patch.object(gcp_setup_runner.GcpTaskRunner, "_CheckBillingEnable")
    @mock.patch.object(gcp_setup_runner.GcpTaskRunner, "_NeedProjectSetup")
    @mock.patch.object(gcp_setup_runner.GcpTaskRunner, "_SetupClientIDSecret")
    @mock.patch.object(gcp_setup_runner.GcpTaskRunner, "_UpdateProject")
    def testSetupProjectChanged(self, mock_setproj, mock_setid,
                                mock_chkproj, mock_check_billing):
        """test setup project when project changed."""
        mock_chkproj.return_value = True
        mock_setproj.return_value = True
        self.gcp_env_runner._SetupProject(self.gcloud_runner)
        mock_setproj.assert_called_once()
        mock_setid.assert_called_once()
        mock_check_billing.assert_called_once()

    @mock.patch.object(utils, "GetUserAnswerYes")
    def testNeedProjectSetup(self, mock_ans):
        """test need project setup."""
        # Test need project setup.
        self.gcp_env_runner.project = ""
        self.gcp_env_runner.zone = ""
        self.assertTrue(self.gcp_env_runner._NeedProjectSetup())
        # Test no need project setup and get user's answer.
        self.gcp_env_runner.project = "test_project"
        self.gcp_env_runner.zone = "test_zone"
        self.gcp_env_runner._NeedProjectSetup()
        mock_ans.assert_called_once()

    def testNeedClientIDSetup(self):
        """test need client_id setup."""
        # Test project changed.
        self.assertTrue(self.gcp_env_runner._NeedClientIDSetup(True))
        # Test project is not changed but client_id or client_secret is empty.
        self.gcp_env_runner.client_id = ""
        self.gcp_env_runner.client_secret = ""
        self.assertTrue(self.gcp_env_runner._NeedClientIDSetup(False))
        # Test no need client_id setup.
        self.gcp_env_runner.client_id = "test_client_id"
        self.gcp_env_runner.client_secret = "test_client_secret"
        self.assertFalse(self.gcp_env_runner._NeedClientIDSetup(False))

    @mock.patch.object(utils, "CheckOutput")
    def testEnableGcloudServices(self, mock_run):
        """test enable Gcloud services."""
        mock_run.return_value = ""
        self.gcp_env_runner._EnableGcloudServices(self.gcloud_runner)
        mock_run.assert_has_calls([
            mock.call(["gcloud", "services", "enable",
                       gcp_setup_runner._ANDROID_BUILD_SERVICE],
                      env=self.gcloud_runner._env, stderr=-2),
            mock.call(["gcloud", "services", "enable",
                       gcp_setup_runner._COMPUTE_ENGINE_SERVICE],
                      env=self.gcloud_runner._env, stderr=-2)])

    @mock.patch.object(utils, "CheckOutput")
    def testGoogleAPIService(self, mock_run):
        """Test GoogleAPIService"""
        api_service = gcp_setup_runner.GoogleAPIService("service_name",
                                                        "error_message")
        api_service.EnableService(self.gcloud_runner)
        mock_run.assert_has_calls([
            mock.call(["gcloud", "services", "enable", "service_name"],
                      env=self.gcloud_runner._env, stderr=-2)])

    @mock.patch.object(utils, "CheckOutput")
    def testCheckBillingEnable(self, mock_run):
        """Test CheckBillingEnable"""
        # Test billing account in gcp project already enabled.
        mock_run.return_value = "billingEnabled: true"
        self.gcp_env_runner._CheckBillingEnable(self.gcloud_runner)
        mock_run.assert_has_calls([
            mock.call(
                [
                    "gcloud", "alpha", "billing", "projects", "describe",
                    self.gcp_env_runner.project
                ],
                env=self.gcloud_runner._env)
        ])

        # Test billing account in gcp project was not enabled.
        mock_run.return_value = "billingEnabled: false"
        with self.assertRaises(errors.NoBillingError):
            self.gcp_env_runner._CheckBillingEnable(self.gcloud_runner)


if __name__ == "__main__":
    unittest.main()