1# Copyright 2016 Google LLC 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); 4# you may not use this file except in compliance with the License. 5# You may obtain a copy of the License at 6# 7# http://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12# See the License for the specific language governing permissions and 13# limitations under the License. 14 15import io 16import json 17import os 18import subprocess 19 20import mock 21import pytest 22 23from google.auth import _cloud_sdk 24from google.auth import environment_vars 25from google.auth import exceptions 26 27 28DATA_DIR = os.path.join(os.path.dirname(__file__), "data") 29AUTHORIZED_USER_FILE = os.path.join(DATA_DIR, "authorized_user.json") 30 31with io.open(AUTHORIZED_USER_FILE) as fh: 32 AUTHORIZED_USER_FILE_DATA = json.load(fh) 33 34SERVICE_ACCOUNT_FILE = os.path.join(DATA_DIR, "service_account.json") 35 36with io.open(SERVICE_ACCOUNT_FILE) as fh: 37 SERVICE_ACCOUNT_FILE_DATA = json.load(fh) 38 39with io.open(os.path.join(DATA_DIR, "cloud_sdk_config.json"), "rb") as fh: 40 CLOUD_SDK_CONFIG_FILE_DATA = fh.read() 41 42 43@pytest.mark.parametrize( 44 "data, expected_project_id", 45 [ 46 (CLOUD_SDK_CONFIG_FILE_DATA, "example-project"), 47 (b"I am some bad json", None), 48 (b"{}", None), 49 ], 50) 51def test_get_project_id(data, expected_project_id): 52 check_output_patch = mock.patch( 53 "subprocess.check_output", autospec=True, return_value=data 54 ) 55 56 with check_output_patch as check_output: 57 project_id = _cloud_sdk.get_project_id() 58 59 assert project_id == expected_project_id 60 assert check_output.called 61 62 63@mock.patch( 64 "subprocess.check_output", 65 autospec=True, 66 side_effect=subprocess.CalledProcessError(-1, None), 67) 68def test_get_project_id_call_error(check_output): 69 project_id = _cloud_sdk.get_project_id() 70 assert project_id is None 71 assert check_output.called 72 73 74def test__run_subprocess_ignore_stderr(): 75 command = [ 76 "python", 77 "-c", 78 "from __future__ import print_function;" 79 + "import sys;" 80 + "print('error', file=sys.stderr);" 81 + "print('output', file=sys.stdout)", 82 ] 83 84 # If we ignore stderr, then the output only has stdout 85 output = _cloud_sdk._run_subprocess_ignore_stderr(command) 86 assert output == b"output\n" 87 88 # If we pipe stderr to stdout, then the output is mixed with stdout and stderr. 89 output = subprocess.check_output(command, stderr=subprocess.STDOUT) 90 assert output == b"output\nerror\n" or output == b"error\noutput\n" 91 92 93@mock.patch("os.name", new="nt") 94def test_get_project_id_windows(): 95 check_output_patch = mock.patch( 96 "subprocess.check_output", 97 autospec=True, 98 return_value=CLOUD_SDK_CONFIG_FILE_DATA, 99 ) 100 101 with check_output_patch as check_output: 102 project_id = _cloud_sdk.get_project_id() 103 104 assert project_id == "example-project" 105 assert check_output.called 106 # Make sure the executable is `gcloud.cmd`. 107 args = check_output.call_args[0] 108 command = args[0] 109 executable = command[0] 110 assert executable == "gcloud.cmd" 111 112 113@mock.patch("google.auth._cloud_sdk.get_config_path", autospec=True) 114def test_get_application_default_credentials_path(get_config_dir): 115 config_path = "config_path" 116 get_config_dir.return_value = config_path 117 credentials_path = _cloud_sdk.get_application_default_credentials_path() 118 assert credentials_path == os.path.join( 119 config_path, _cloud_sdk._CREDENTIALS_FILENAME 120 ) 121 122 123def test_get_config_path_env_var(monkeypatch): 124 config_path_sentinel = "config_path" 125 monkeypatch.setenv(environment_vars.CLOUD_SDK_CONFIG_DIR, config_path_sentinel) 126 config_path = _cloud_sdk.get_config_path() 127 assert config_path == config_path_sentinel 128 129 130@mock.patch("os.path.expanduser") 131def test_get_config_path_unix(expanduser): 132 expanduser.side_effect = lambda path: path 133 134 config_path = _cloud_sdk.get_config_path() 135 136 assert os.path.split(config_path) == ("~/.config", _cloud_sdk._CONFIG_DIRECTORY) 137 138 139@mock.patch("os.name", new="nt") 140def test_get_config_path_windows(monkeypatch): 141 appdata = "appdata" 142 monkeypatch.setenv(_cloud_sdk._WINDOWS_CONFIG_ROOT_ENV_VAR, appdata) 143 144 config_path = _cloud_sdk.get_config_path() 145 146 assert os.path.split(config_path) == (appdata, _cloud_sdk._CONFIG_DIRECTORY) 147 148 149@mock.patch("os.name", new="nt") 150def test_get_config_path_no_appdata(monkeypatch): 151 monkeypatch.delenv(_cloud_sdk._WINDOWS_CONFIG_ROOT_ENV_VAR, raising=False) 152 monkeypatch.setenv("SystemDrive", "G:") 153 154 config_path = _cloud_sdk.get_config_path() 155 156 assert os.path.split(config_path) == ("G:/\\", _cloud_sdk._CONFIG_DIRECTORY) 157 158 159@mock.patch("os.name", new="nt") 160@mock.patch("subprocess.check_output", autospec=True) 161def test_get_auth_access_token_windows(check_output): 162 check_output.return_value = b"access_token\n" 163 164 token = _cloud_sdk.get_auth_access_token() 165 assert token == "access_token" 166 check_output.assert_called_with( 167 ("gcloud.cmd", "auth", "print-access-token"), stderr=subprocess.STDOUT 168 ) 169 170 171@mock.patch("subprocess.check_output", autospec=True) 172def test_get_auth_access_token_with_account(check_output): 173 check_output.return_value = b"access_token\n" 174 175 token = _cloud_sdk.get_auth_access_token(account="account") 176 assert token == "access_token" 177 check_output.assert_called_with( 178 ("gcloud", "auth", "print-access-token", "--account=account"), 179 stderr=subprocess.STDOUT, 180 ) 181 182 183@mock.patch("subprocess.check_output", autospec=True) 184def test_get_auth_access_token_with_exception(check_output): 185 check_output.side_effect = OSError() 186 187 with pytest.raises(exceptions.UserAccessTokenError): 188 _cloud_sdk.get_auth_access_token(account="account") 189