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