Spaces:
Paused
Paused
File size: 5,401 Bytes
a5784e9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 | import os
from unittest.mock import AsyncMock, patch
import pytest
from api_utils.auth_manager import AuthManager, auth_manager
# --- Fixtures ---
@pytest.fixture
def manager():
return AuthManager()
@pytest.fixture
def mock_saved_auth_dir(tmp_path):
"""Mock the SAVED_AUTH_DIR constant."""
with patch("api_utils.auth_manager.SAVED_AUTH_DIR", str(tmp_path)):
yield tmp_path
# --- Tests ---
def test_init_default():
"""Test initialization without environment variable."""
with patch.dict(os.environ, {}, clear=True):
am = AuthManager()
assert am.current_profile is None
assert am.failed_profiles == set()
def test_init_with_env_var():
"""Test initialization with ACTIVE_AUTH_JSON_PATH."""
with patch.dict(os.environ, {"ACTIVE_AUTH_JSON_PATH": "/path/to/auth.json"}):
am = AuthManager()
assert am.current_profile == "/path/to/auth.json"
@pytest.mark.asyncio
async def test_get_available_profiles_no_dir(manager):
"""Test get_available_profiles when directory doesn't exist."""
with patch("os.path.exists", return_value=False):
profiles = await manager.get_available_profiles()
assert profiles == []
@pytest.mark.asyncio
async def test_get_available_profiles_success(manager, mock_saved_auth_dir):
"""Test get_available_profiles with existing files."""
# Create dummy auth files
(mock_saved_auth_dir / "auth1.json").touch()
(mock_saved_auth_dir / "auth2.json").touch()
with patch("os.path.exists", return_value=True):
profiles = await manager.get_available_profiles()
assert len(profiles) == 2
assert any("auth1.json" in p for p in profiles)
assert any("auth2.json" in p for p in profiles)
# Check sorting
assert profiles == sorted(profiles)
@pytest.mark.asyncio
async def test_get_next_profile_success(manager):
"""Test getting next profile successfully."""
mock_profiles = ["/dir/auth1.json", "/dir/auth2.json"]
with patch.object(
manager, "get_available_profiles", new_callable=AsyncMock
) as mock_get:
mock_get.return_value = mock_profiles
# First call
next_p = await manager.get_next_profile()
assert next_p == "/dir/auth1.json"
assert manager.current_profile == "/dir/auth1.json"
@pytest.mark.asyncio
async def test_get_next_profile_skips_failed(manager):
"""Test that failed profiles are skipped."""
mock_profiles = ["/dir/auth1.json", "/dir/auth2.json", "/dir/auth3.json"]
manager.failed_profiles.add("/dir/auth1.json")
with patch.object(
manager, "get_available_profiles", new_callable=AsyncMock
) as mock_get:
mock_get.return_value = mock_profiles
next_p = await manager.get_next_profile()
assert next_p == "/dir/auth2.json"
# Mark auth2 as failed and try again
manager.mark_profile_failed() # marks current (auth2)
next_p_2 = await manager.get_next_profile()
assert next_p_2 == "/dir/auth3.json"
@pytest.mark.asyncio
async def test_get_next_profile_skips_current(manager):
"""Test that current profile is skipped even if not failed (to ensure rotation if needed, or just behavior check)."""
# Actually logic says: os.path.basename(p) != current_basename
# So if we call get_next_profile, it should give us a *different* one if available.
mock_profiles = ["/dir/auth1.json", "/dir/auth2.json"]
manager.current_profile = "/dir/auth1.json"
with patch.object(
manager, "get_available_profiles", new_callable=AsyncMock
) as mock_get:
mock_get.return_value = mock_profiles
next_p = await manager.get_next_profile()
assert next_p == "/dir/auth2.json"
@pytest.mark.asyncio
async def test_get_next_profile_exhausted(manager):
"""Test raising RuntimeError when no profiles available."""
with patch.object(
manager, "get_available_profiles", new_callable=AsyncMock
) as mock_get:
mock_get.return_value = []
with pytest.raises(RuntimeError, match="All authentication profiles exhausted"):
await manager.get_next_profile()
@pytest.mark.asyncio
async def test_get_next_profile_all_failed(manager):
"""Test raising RuntimeError when all profiles failed."""
mock_profiles = ["/dir/auth1.json"]
manager.failed_profiles.add("/dir/auth1.json")
with patch.object(
manager, "get_available_profiles", new_callable=AsyncMock
) as mock_get:
mock_get.return_value = mock_profiles
with pytest.raises(RuntimeError, match="All authentication profiles exhausted"):
await manager.get_next_profile()
def test_mark_profile_failed(manager):
"""Test marking profile as failed."""
# 1. With explicit path
manager.mark_profile_failed("/dir/auth1.json")
assert "/dir/auth1.json" in manager.failed_profiles
# 2. With current profile
manager.current_profile = "/dir/auth2.json"
manager.mark_profile_failed()
assert "/dir/auth2.json" in manager.failed_profiles
# 3. No profile active and no arg
manager.current_profile = None
manager.mark_profile_failed() # Should log warning but not crash
# (Assert log if needed, but no crash is enough for basic coverage)
def test_global_instance():
"""Ensure global instance exists."""
assert isinstance(auth_manager, AuthManager)
|