|
| 1 | +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. |
| 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 | + |
| 15 | +"""Unit tests for ve_identity auth_config module.""" |
| 16 | + |
| 17 | +import pytest |
| 18 | +from veadk.integrations.ve_identity import ( |
| 19 | + api_key_auth, |
| 20 | + oauth2_auth, |
| 21 | + workload_auth, |
| 22 | + ApiKeyAuthConfig, |
| 23 | + OAuth2AuthConfig, |
| 24 | + WorkloadAuthConfig, |
| 25 | +) |
| 26 | + |
| 27 | + |
| 28 | +class TestApiKeyAuth: |
| 29 | + """Tests for api_key_auth factory function.""" |
| 30 | + |
| 31 | + def test_api_key_auth_basic(self): |
| 32 | + """Test creating basic API key auth config.""" |
| 33 | + config = api_key_auth("test-provider") |
| 34 | + |
| 35 | + assert isinstance(config, ApiKeyAuthConfig) |
| 36 | + assert config.provider_name == "test-provider" |
| 37 | + assert config.auth_type == "api_key" |
| 38 | + assert config.region == "cn-beijing" |
| 39 | + assert config.identity_client is None |
| 40 | + |
| 41 | + def test_api_key_auth_with_region(self): |
| 42 | + """Test creating API key auth config with custom region.""" |
| 43 | + config = api_key_auth("test-provider", region="us-east-1") |
| 44 | + |
| 45 | + assert config.provider_name == "test-provider" |
| 46 | + assert config.region == "us-east-1" |
| 47 | + assert config.auth_type == "api_key" |
| 48 | + |
| 49 | + def test_api_key_auth_empty_provider_name(self): |
| 50 | + """Test that empty provider_name raises ValueError.""" |
| 51 | + with pytest.raises(ValueError, match="provider_name cannot be empty"): |
| 52 | + api_key_auth("") |
| 53 | + |
| 54 | + def test_api_key_auth_whitespace_provider_name(self): |
| 55 | + """Test that whitespace-only provider_name raises ValueError.""" |
| 56 | + with pytest.raises(ValueError, match="provider_name cannot be empty"): |
| 57 | + api_key_auth(" ") |
| 58 | + |
| 59 | + |
| 60 | +class TestOAuth2Auth: |
| 61 | + """Tests for oauth2_auth factory function.""" |
| 62 | + |
| 63 | + def test_oauth2_auth_basic(self): |
| 64 | + """Test creating basic OAuth2 auth config.""" |
| 65 | + config = oauth2_auth( |
| 66 | + provider_name="github", |
| 67 | + scopes=["repo", "user"], |
| 68 | + auth_flow="M2M" |
| 69 | + ) |
| 70 | + |
| 71 | + assert isinstance(config, OAuth2AuthConfig) |
| 72 | + assert config.provider_name == "github" |
| 73 | + assert config.scopes == ["repo", "user"] |
| 74 | + assert config.auth_flow == "M2M" |
| 75 | + assert config.auth_type == "oauth2" |
| 76 | + assert config.force_authentication is False |
| 77 | + assert config.callback_url is None |
| 78 | + |
| 79 | + def test_oauth2_auth_with_all_params(self): |
| 80 | + """Test creating OAuth2 auth config with all parameters.""" |
| 81 | + def on_auth_url_callback(url: str): |
| 82 | + pass |
| 83 | + |
| 84 | + config = oauth2_auth( |
| 85 | + provider_name="github", |
| 86 | + scopes=["repo", "user"], |
| 87 | + auth_flow="USER_FEDERATION", |
| 88 | + callback_url="https://example.com/callback", |
| 89 | + force_authentication=True, |
| 90 | + response_for_auth_required="Please authorize", |
| 91 | + on_auth_url=on_auth_url_callback, |
| 92 | + region="us-west-2" |
| 93 | + ) |
| 94 | + |
| 95 | + assert config.provider_name == "github" |
| 96 | + assert config.scopes == ["repo", "user"] |
| 97 | + assert config.auth_flow == "USER_FEDERATION" |
| 98 | + assert config.callback_url == "https://example.com/callback" |
| 99 | + assert config.force_authentication is True |
| 100 | + assert config.response_for_auth_required == "Please authorize" |
| 101 | + assert config.on_auth_url == on_auth_url_callback |
| 102 | + assert config.region == "us-west-2" |
| 103 | + |
| 104 | + def test_oauth2_auth_empty_scopes(self): |
| 105 | + """Test that empty scopes raises ValueError.""" |
| 106 | + with pytest.raises(ValueError, match="scopes cannot be empty"): |
| 107 | + oauth2_auth( |
| 108 | + provider_name="github", |
| 109 | + scopes=[], |
| 110 | + auth_flow="M2M" |
| 111 | + ) |
| 112 | + |
| 113 | + def test_oauth2_auth_empty_scope_value(self): |
| 114 | + """Test that empty scope value raises ValueError.""" |
| 115 | + with pytest.raises(ValueError, match="scope values cannot be empty"): |
| 116 | + oauth2_auth( |
| 117 | + provider_name="github", |
| 118 | + scopes=["repo", ""], |
| 119 | + auth_flow="M2M" |
| 120 | + ) |
| 121 | + |
| 122 | + def test_oauth2_auth_duplicate_scopes_removed(self): |
| 123 | + """Test that duplicate scopes are removed.""" |
| 124 | + config = oauth2_auth( |
| 125 | + provider_name="github", |
| 126 | + scopes=["repo", "user", "repo", "user"], |
| 127 | + auth_flow="M2M" |
| 128 | + ) |
| 129 | + |
| 130 | + assert config.scopes == ["repo", "user"] |
| 131 | + |
| 132 | + def test_oauth2_auth_invalid_callback_url(self): |
| 133 | + """Test that invalid callback URL raises ValueError.""" |
| 134 | + with pytest.raises(ValueError, match="callback_url must be a valid HTTP/HTTPS URL"): |
| 135 | + oauth2_auth( |
| 136 | + provider_name="github", |
| 137 | + scopes=["repo"], |
| 138 | + auth_flow="M2M", |
| 139 | + callback_url="invalid-url" |
| 140 | + ) |
| 141 | + |
| 142 | + def test_oauth2_auth_valid_https_callback_url(self): |
| 143 | + """Test that valid HTTPS callback URL is accepted.""" |
| 144 | + config = oauth2_auth( |
| 145 | + provider_name="github", |
| 146 | + scopes=["repo"], |
| 147 | + auth_flow="M2M", |
| 148 | + callback_url="https://example.com/callback" |
| 149 | + ) |
| 150 | + |
| 151 | + assert config.callback_url == "https://example.com/callback" |
| 152 | + |
| 153 | + def test_oauth2_auth_valid_http_callback_url(self): |
| 154 | + """Test that valid HTTP callback URL is accepted.""" |
| 155 | + config = oauth2_auth( |
| 156 | + provider_name="github", |
| 157 | + scopes=["repo"], |
| 158 | + auth_flow="M2M", |
| 159 | + callback_url="http://localhost:8080/callback" |
| 160 | + ) |
| 161 | + |
| 162 | + assert config.callback_url == "http://localhost:8080/callback" |
| 163 | + |
| 164 | + |
| 165 | +class TestWorkloadAuth: |
| 166 | + """Tests for workload_auth factory function.""" |
| 167 | + |
| 168 | + def test_workload_auth_basic(self): |
| 169 | + """Test creating basic workload auth config.""" |
| 170 | + config = workload_auth("test-provider") |
| 171 | + |
| 172 | + assert isinstance(config, WorkloadAuthConfig) |
| 173 | + assert config.provider_name == "test-provider" |
| 174 | + assert config.auth_type == "workload" |
| 175 | + assert config.region == "cn-beijing" |
| 176 | + assert config.identity_client is None |
| 177 | + |
| 178 | + def test_workload_auth_with_region(self): |
| 179 | + """Test creating workload auth config with custom region.""" |
| 180 | + config = workload_auth("test-provider", region="eu-west-1") |
| 181 | + |
| 182 | + assert config.provider_name == "test-provider" |
| 183 | + assert config.region == "eu-west-1" |
| 184 | + assert config.auth_type == "workload" |
| 185 | + |
| 186 | + def test_workload_auth_empty_provider_name(self): |
| 187 | + """Test that empty provider_name raises ValueError.""" |
| 188 | + with pytest.raises(ValueError, match="provider_name cannot be empty"): |
| 189 | + workload_auth("") |
| 190 | + |
0 commit comments