|
15 | 15 | import datetime |
16 | 16 | from typing import Optional |
17 | 17 |
|
| 18 | +from aiohttp import ClientResponseError |
| 19 | +from aioresponses import aioresponses |
| 20 | +from google.auth.credentials import Credentials |
18 | 21 | from mocks import FakeCredentials |
19 | 22 | import pytest |
20 | 23 |
|
@@ -138,3 +141,130 @@ async def test_CloudSQLClient_user_agent( |
138 | 141 | assert client._user_agent == f"cloud-sql-python-connector/{version}+{driver}" |
139 | 142 | # close client |
140 | 143 | await client.close() |
| 144 | + |
| 145 | + |
| 146 | +async def test_cloud_sql_error_messages_get_metadata( |
| 147 | + fake_credentials: Credentials, |
| 148 | +) -> None: |
| 149 | + """ |
| 150 | + Test that Cloud SQL Admin API error messages are raised for _get_metadata. |
| 151 | + """ |
| 152 | + # mock Cloud SQL Admin API calls with exceptions |
| 153 | + client = CloudSQLClient( |
| 154 | + sqladmin_api_endpoint="https://sqladmin.googleapis.com", |
| 155 | + quota_project=None, |
| 156 | + credentials=fake_credentials, |
| 157 | + ) |
| 158 | + get_url = "https://sqladmin.googleapis.com/sql/v1beta4/projects/my-project/instances/my-instance/connectSettings" |
| 159 | + resp_body = { |
| 160 | + "error": { |
| 161 | + "code": 403, |
| 162 | + "message": "Cloud SQL Admin API has not been used in project 123456789 before or it is disabled", |
| 163 | + } |
| 164 | + } |
| 165 | + with aioresponses() as mocked: |
| 166 | + mocked.get( |
| 167 | + get_url, |
| 168 | + status=403, |
| 169 | + payload=resp_body, |
| 170 | + repeat=True, |
| 171 | + ) |
| 172 | + with pytest.raises(ClientResponseError) as exc_info: |
| 173 | + await client._get_metadata("my-project", "my-region", "my-instance") |
| 174 | + assert exc_info.value.status == 403 |
| 175 | + assert ( |
| 176 | + exc_info.value.message |
| 177 | + == "Cloud SQL Admin API has not been used in project 123456789 before or it is disabled" |
| 178 | + ) |
| 179 | + await client.close() |
| 180 | + |
| 181 | + |
| 182 | +async def test_get_metadata_error_parsing_json( |
| 183 | + fake_credentials: Credentials, |
| 184 | +) -> None: |
| 185 | + """ |
| 186 | + Test that aiohttp default error messages are raised when _get_metadata gets |
| 187 | + a bad JSON response. |
| 188 | + """ |
| 189 | + # mock Cloud SQL Admin API calls with exceptions |
| 190 | + client = CloudSQLClient( |
| 191 | + sqladmin_api_endpoint="https://sqladmin.googleapis.com", |
| 192 | + quota_project=None, |
| 193 | + credentials=fake_credentials, |
| 194 | + ) |
| 195 | + get_url = "https://sqladmin.googleapis.com/sql/v1beta4/projects/my-project/instances/my-instance/connectSettings" |
| 196 | + resp_body = ["error"] # invalid JSON |
| 197 | + with aioresponses() as mocked: |
| 198 | + mocked.get( |
| 199 | + get_url, |
| 200 | + status=403, |
| 201 | + payload=resp_body, |
| 202 | + repeat=True, |
| 203 | + ) |
| 204 | + with pytest.raises(ClientResponseError) as exc_info: |
| 205 | + await client._get_metadata("my-project", "my-region", "my-instance") |
| 206 | + assert exc_info.value.status == 403 |
| 207 | + assert exc_info.value.message == "Forbidden" |
| 208 | + await client.close() |
| 209 | + |
| 210 | + |
| 211 | +async def test_cloud_sql_error_messages_get_ephemeral( |
| 212 | + fake_credentials: Credentials, |
| 213 | +) -> None: |
| 214 | + """ |
| 215 | + Test that Cloud SQL Admin API error messages are raised for _get_ephemeral. |
| 216 | + """ |
| 217 | + # mock Cloud SQL Admin API calls with exceptions |
| 218 | + client = CloudSQLClient( |
| 219 | + sqladmin_api_endpoint="https://sqladmin.googleapis.com", |
| 220 | + quota_project=None, |
| 221 | + credentials=fake_credentials, |
| 222 | + ) |
| 223 | + post_url = "https://sqladmin.googleapis.com/sql/v1beta4/projects/my-project/instances/my-instance:generateEphemeralCert" |
| 224 | + resp_body = { |
| 225 | + "error": { |
| 226 | + "code": 404, |
| 227 | + "message": "The Cloud SQL instance does not exist.", |
| 228 | + } |
| 229 | + } |
| 230 | + with aioresponses() as mocked: |
| 231 | + mocked.post( |
| 232 | + post_url, |
| 233 | + status=404, |
| 234 | + payload=resp_body, |
| 235 | + repeat=True, |
| 236 | + ) |
| 237 | + with pytest.raises(ClientResponseError) as exc_info: |
| 238 | + await client._get_ephemeral("my-project", "my-instance", "my-key") |
| 239 | + assert exc_info.value.status == 404 |
| 240 | + assert exc_info.value.message == "The Cloud SQL instance does not exist." |
| 241 | + await client.close() |
| 242 | + |
| 243 | + |
| 244 | +async def test_get_ephemeral_error_parsing_json( |
| 245 | + fake_credentials: Credentials, |
| 246 | +) -> None: |
| 247 | + """ |
| 248 | + Test that aiohttp default error messages are raised when _get_ephemeral gets |
| 249 | + a bad JSON response. |
| 250 | + """ |
| 251 | + # mock Cloud SQL Admin API calls with exceptions |
| 252 | + client = CloudSQLClient( |
| 253 | + sqladmin_api_endpoint="https://sqladmin.googleapis.com", |
| 254 | + quota_project=None, |
| 255 | + credentials=fake_credentials, |
| 256 | + ) |
| 257 | + post_url = "https://sqladmin.googleapis.com/sql/v1beta4/projects/my-project/instances/my-instance:generateEphemeralCert" |
| 258 | + resp_body = ["error"] # invalid JSON |
| 259 | + with aioresponses() as mocked: |
| 260 | + mocked.post( |
| 261 | + post_url, |
| 262 | + status=404, |
| 263 | + payload=resp_body, |
| 264 | + repeat=True, |
| 265 | + ) |
| 266 | + with pytest.raises(ClientResponseError) as exc_info: |
| 267 | + await client._get_ephemeral("my-project", "my-instance", "my-key") |
| 268 | + assert exc_info.value.status == 404 |
| 269 | + assert exc_info.value.message == "Not Found" |
| 270 | + await client.close() |
0 commit comments