diff --git a/app/api/endpoints/user.py b/app/api/endpoints/user.py index 05e560756..de4a9ab38 100644 --- a/app/api/endpoints/user.py +++ b/app/api/endpoints/user.py @@ -96,12 +96,12 @@ async def create_user( 新增用户 """ user_info = user_in.model_dump() - if user_info.get("password"): + password = user_info.pop("password", None) + if password: try: - user_info["hashed_password"] = get_password_hash(user_info["password"]) + user_info["hashed_password"] = get_password_hash(password) except PasswordTooLongError as error: return _SchemaResponse(success=False, message=str(error)) - user_info.pop("password") try: user = await service.create(user_info) except UserNameConflictError: @@ -192,7 +192,7 @@ async def upload_avatar( raise HTTPException(status_code=403, detail="用户权限不足") # 将文件转换为Base64 - file_base64 = str(base64.b64encode(file.file.read())) + file_base64 = base64.b64encode(await file.read()).decode("ascii") # 更新到用户表 user = await service.get_by_id(user_id) if not user: diff --git a/app/application/github_auth.py b/app/application/github_auth.py index 887512ef8..d6f87e0fd 100644 --- a/app/application/github_auth.py +++ b/app/application/github_auth.py @@ -141,6 +141,8 @@ class GithubAuthService: self._client_id = client_id self._clock = clock self._lock = asyncio.Lock() + self._credential_lock = asyncio.Lock() + self._refresh_lock = asyncio.Lock() self._pending_sessions: dict[str, _PendingDeviceSession] = {} async def start_device_auth(self) -> GithubDeviceAuthStartResult: @@ -281,6 +283,8 @@ class GithubAuthService: token, metadata, ) + if not token: + return await self.status() granted_scopes = set(str(metadata.get("scope") or "").replace(",", " ").split()) needs_reauthorization = needs_reauthorization or not set( GITHUB_DEVICE_AUTH_SCOPES @@ -328,26 +332,39 @@ class GithubAuthService: normalized = token.strip() if not normalized: raise GithubAuthError("GitHub Token 不能为空") - success, message = self._settings.update("GITHUB_TOKEN", normalized) - if success is False: - raise GithubAuthError(message or "GitHub Token 保存失败") - await self.forget_oauth_metadata() + await self._persist_credentials(normalized, None) return await self.status() async def clear_token(self) -> None: """ 清除 GitHub Token 和 OAuth 刷新元数据。 """ - success, message = self._settings.update("GITHUB_TOKEN", None) - if success is False: - raise GithubAuthError(message or "GitHub Token 清除失败") - await self.forget_oauth_metadata() + await self._persist_credentials(None, None) async def forget_oauth_metadata(self) -> None: """ 清除 OAuth 元数据但保留当前部署设置中的 Token 值。 """ - await self._system_config.async_delete(self._AUTH_CONFIG_KEY) + async with self._credential_lock: + await self._system_config.async_delete(self._AUTH_CONFIG_KEY) + + async def _persist_credentials( + self, token: str | None, metadata: dict[str, Any] | None, + ) -> None: + """ + 串行发布凭据和对应元数据,防止迟到刷新覆盖手动清除或更换。 + + :param token: 新访问令牌;None 表示清除 + :param metadata: OAuth 元数据;None 表示手动凭据或清除 + """ + async with self._credential_lock: + success, message = self._settings.update("GITHUB_TOKEN", token) + if success is False: + raise GithubAuthError(message or "GitHub Token 保存失败") + if metadata is None: + await self._system_config.async_delete(self._AUTH_CONFIG_KEY) + else: + await self._system_config.async_set(self._AUTH_CONFIG_KEY, metadata) async def _save_oauth_token( self, @@ -359,12 +376,6 @@ class GithubAuthService: """ if not exchange.access_token: raise GithubAuthError("GitHub 未返回访问 Token") - success, message = self._settings.update( - "GITHUB_TOKEN", - exchange.access_token, - ) - if success is False: - raise GithubAuthError(message or "GitHub Token 保存失败") metadata: dict[str, Any] = { "login": github_user.login, "scope": exchange.scope or "", @@ -375,7 +386,7 @@ class GithubAuthService: metadata["expires_at"] = int(self._clock() + max(exchange.expires_in, 0)) if exchange.refresh_token_expires_in is not None: metadata["refresh_token_expires_at"] = int(self._clock() + max(exchange.refresh_token_expires_in, 0)) - await self._system_config.async_set(self._AUTH_CONFIG_KEY, metadata) + await self._persist_credentials(exchange.access_token, metadata) async def _refresh_if_needed( self, @@ -383,7 +394,27 @@ class GithubAuthService: metadata: dict[str, Any], ) -> tuple[str, dict[str, Any], bool]: """ - 在过期窗口内刷新 OAuth Token,失败时保留原值供诊断和重试。 + 串行刷新 OAuth 快照,已被清除或替换时返回当前凭据。 + + :param token: 调用方读取的访问令牌快照 + :param metadata: 与令牌对应的 OAuth 元数据快照 + :return: 当前访问令牌、对应元数据及是否需要重新授权 + """ + async with self._refresh_lock: + current_token, current_metadata = self._read_access_token(), self._read_auth_metadata() + if (current_token, current_metadata) != (token, metadata): + return current_token, current_metadata, False + return await self._refresh_current_token(token, metadata) + + async def _refresh_current_token( + self, token: str, metadata: dict[str, Any], + ) -> tuple[str, dict[str, Any], bool]: + """ + 刷新单份仍有效的快照,并在写回时核对它未被其他凭据操作替换。 + + :param token: 刷新锁内已确认仍为当前值的访问令牌 + :param metadata: 已确认对应当前令牌的 OAuth 元数据 + :return: 当前访问令牌、对应元数据及是否需要重新授权 """ expires_at = self._as_int(metadata.get("expires_at")) if expires_at is None or expires_at > int(self._clock()) + 60: @@ -397,9 +428,9 @@ class GithubAuthService: refresh_token, ) except (GithubAuthError, GithubAuthTransportError): - return token, metadata, True + return await self._failed_refresh_snapshot(token, metadata) if not exchange.access_token: - return token, metadata, True + return await self._failed_refresh_snapshot(token, metadata) refreshed_metadata = copy.deepcopy(metadata) if exchange.refresh_token: refreshed_metadata["refresh_token"] = exchange.refresh_token @@ -409,15 +440,30 @@ class GithubAuthService: refreshed_metadata["refresh_token_expires_at"] = int( self._clock() + max(exchange.refresh_token_expires_in, 0) ) - success, _ = self._settings.update("GITHUB_TOKEN", exchange.access_token) - if success is False: - return token, metadata, True - await self._system_config.async_set( - self._AUTH_CONFIG_KEY, - refreshed_metadata, - ) + async with self._credential_lock: + current_token, current_metadata = self._read_access_token(), self._read_auth_metadata() + if (current_token, current_metadata) != (token, metadata): + return current_token, current_metadata, False + success, _ = self._settings.update("GITHUB_TOKEN", exchange.access_token) + if success is False: + return token, metadata, True + await self._system_config.async_set(self._AUTH_CONFIG_KEY, refreshed_metadata) return exchange.access_token, refreshed_metadata, False + async def _failed_refresh_snapshot( + self, token: str, metadata: dict[str, Any], + ) -> tuple[str, dict[str, Any], bool]: + """ + 刷新失败后返回当前凭据,清除或更换已生效时不再向调用者交付旧令牌。 + + :param token: 本次远端刷新使用的访问令牌快照 + :param metadata: 本次远端刷新使用的 OAuth 元数据快照 + :return: 当前凭据;仅原快照仍有效时标记需要重新授权 + """ + async with self._credential_lock: + current_token, current_metadata = self._read_access_token(), self._read_auth_metadata() + return current_token, current_metadata, (current_token, current_metadata) == (token, metadata) + async def _session_interval(self, session_id: str) -> int: """ 读取会话当前轮询间隔,并在会话刚过期时返回最小等待值。 diff --git a/app/application/security/throttle.py b/app/application/security/throttle.py index 6b2adfca8..5bf8c8a4f 100644 --- a/app/application/security/throttle.py +++ b/app/application/security/throttle.py @@ -90,6 +90,7 @@ class LoginThrottle: """ now = self._clock() with self._lock: + self._cleanup(now) entry = self._entries.get(key) if entry is None: return 0 @@ -102,6 +103,7 @@ class LoginThrottle: """ now = self._clock() with self._lock: + self._cleanup(now) entry = self._entries.setdefault(key, _Entry()) entry.failures += 1 entry.updated_at = now diff --git a/app/application/settings/contract.py b/app/application/settings/contract.py index 286693753..4ab24d9bd 100644 --- a/app/application/settings/contract.py +++ b/app/application/settings/contract.py @@ -721,11 +721,13 @@ def _runtime_settings_schema() -> dict[str, Any]: return Settings.model_json_schema() -def validate_setting_value(spec: SettingSpec, value: Any) -> None: +def validate_setting_value(spec: SettingSpec, value: Any) -> Any: """ - 按合同类型验证数据库设置的完整候选值。 + 按合同类型验证数据库设置的完整候选值,并返回规范化后的值。 + + :return: 类型转换后的候选值,调用方应将其序列化后持久化 """ - TypeAdapter(spec.value_annotation).validate_python(value) + return TypeAdapter(spec.value_annotation).validate_python(value) CORE_SETTING_SPECS, SYSTEMCONFIG_SETTING_SPECS = build_setting_specs() diff --git a/app/application/settings/service.py b/app/application/settings/service.py index a03956369..36a2f032c 100644 --- a/app/application/settings/service.py +++ b/app/application/settings/service.py @@ -9,6 +9,8 @@ import threading from collections.abc import Awaitable, Callable from typing import Any, ContextManager, Optional, cast +from pydantic import TypeAdapter + from app.application.configuration import ( RuntimeSettingsService, SystemConfigService, @@ -675,8 +677,8 @@ class SystemSettingsService: match_value, ) normalized_value = self._normalize_systemconfig_value(next_value) - validate_setting_value(spec, normalized_value) - return normalized_value + validated_value = validate_setting_value(spec, normalized_value) + return TypeAdapter(spec.value_annotation).dump_python(validated_value, mode="json", exclude_unset=True) async def update( self, diff --git a/app/db/adapters/user.py b/app/db/adapters/user.py index be9cfd771..1c5ba4971 100644 --- a/app/db/adapters/user.py +++ b/app/db/adapters/user.py @@ -8,7 +8,7 @@ from collections.abc import Callable, Mapping from contextlib import AbstractAsyncContextManager from typing import Any, Optional, cast -from sqlalchemy import func, select +from sqlalchemy import func, select, update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Session @@ -233,12 +233,21 @@ class SqlAlchemyUserRepository(UserRepository): ) -> Optional[User]: """ 先锁管理员集合再锁目标用户,保护并发下最后一个启用管理员。 + + SQLite 忽略 FOR UPDATE;先获取数据库写锁,并一直持有到调用方提交或回滚, + 防止多个事务基于旧管理员计数分别移除最后的两个账号。 """ + if session.get_bind().dialect.name == "sqlite": + await session.execute( + update(User).where(User.is_superuser.is_(True)).values(id=User.id) + .execution_options(synchronize_session=False) + ) result = await session.execute( select(User) .where(User.is_active.is_(True), User.is_superuser.is_(True)) .order_by(User.id) .with_for_update() + .execution_options(populate_existing=True) ) administrators = list(result.scalars().all()) model = next( @@ -246,7 +255,10 @@ class SqlAlchemyUserRepository(UserRepository): None, ) if model is None: - locked = await session.execute(select(User).where(User.id == user_id).with_for_update()) + locked = await session.execute( + select(User).where(User.id == user_id).with_for_update() + .execution_options(populate_existing=True) + ) model = cast(Optional[User], locked.scalars().first()) if model is None: return None diff --git a/app/db/oper/userconfig.py b/app/db/oper/userconfig.py index a3f14584e..1d7dc2136 100644 --- a/app/db/oper/userconfig.py +++ b/app/db/oper/userconfig.py @@ -8,6 +8,7 @@ from collections.abc import Iterator from contextlib import contextmanager from typing import Optional, Union +from sqlalchemy import update from sqlalchemy.orm import Session from app.db.base import DbOper @@ -75,9 +76,15 @@ class UserConfigOper(DbOper, metaclass=Singleton): ) -> bool: """ 在调用方 Session 中暂存写入,返回提交后是否应移除缓存项。 + + SQLite 写锁在读取配置前获取,避免并发删除用户后对已消失配置执行过期更新。 + 事务提交和回滚仍由调用方负责,用户不存在时由既有外键约束拒绝新增。 """ if isinstance(key, UserConfigKey): key = key.value + if db.get_bind().dialect.name == "sqlite": + db.execute(update(UserConfig).where(UserConfig.username == username, UserConfig.key == key) + .values(id=UserConfig.id).execution_options(synchronize_session=False)) conf = UserConfig.get_by_key(db=db, username=username, key=key) if conf: if value: diff --git a/app/schemas/initialization.py b/app/schemas/initialization.py index 063968986..9d43cb3b7 100644 --- a/app/schemas/initialization.py +++ b/app/schemas/initialization.py @@ -28,13 +28,13 @@ class InitializationRequest(BaseModel): confirm_password: str = Field(min_length=6, max_length=50, description="再次输入的管理员密码,必须与 password 完全一致") api_key: str = Field(min_length=16, max_length=256, description="接口访问密钥,长度为 16 至 256 个字符") - @field_validator("username", "api_key") + @field_validator("username", "api_key", mode="before") @classmethod def strip_text(cls, value: str) -> str: """ 去除用户名和 API Key 的首尾空白,避免产生不可见凭据差异。 """ - return value.strip() + return value.strip() if isinstance(value, str) else value @field_validator("password", "confirm_password") @classmethod diff --git a/tests/test_api_authorization.py b/tests/test_api_authorization.py index 42271fb37..7d82ca1d9 100644 --- a/tests/test_api_authorization.py +++ b/tests/test_api_authorization.py @@ -8,7 +8,7 @@ import io from types import SimpleNamespace import pytest -from fastapi import HTTPException +from fastapi import HTTPException, UploadFile from starlette.requests import Request from starlette.responses import Response @@ -506,7 +506,7 @@ def test_upload_avatar_rejects_other_user_for_non_superuser(): 普通用户不能通过 user_id 参数修改其他用户头像。 """ current_user = SimpleNamespace(id=1, is_superuser=False) - upload_file = SimpleNamespace(file=io.BytesIO(b"avatar"), filename="avatar.png") + upload_file = UploadFile(file=io.BytesIO(b"avatar"), filename="avatar.png") with pytest.raises(HTTPException) as exc_info: asyncio.run( @@ -529,7 +529,7 @@ def test_upload_avatar_returns_filename_in_data(monkeypatch): fake_user = SimpleNamespace() current_user = SimpleNamespace(id=1, is_superuser=False) - upload_file = SimpleNamespace(file=io.BytesIO(b"avatar"), filename="avatar.png") + upload_file = UploadFile(file=io.BytesIO(b"avatar"), filename="avatar.png") class FakeService: """ @@ -565,4 +565,4 @@ def test_upload_avatar_returns_filename_in_data(monkeypatch): assert response.data == {"filename": "avatar.png"} assert response.message == "" assert not hasattr(response, "message_i18n") - assert fake_user.values == {"avatar": "data:image/ico;base64,b'YXZhdGFy'"} + assert fake_user.values == {"avatar": "data:image/ico;base64,YXZhdGFy"} diff --git a/tests/test_identity_settings_review_regressions.py b/tests/test_identity_settings_review_regressions.py new file mode 100644 index 000000000..d2596fcea --- /dev/null +++ b/tests/test_identity_settings_review_regressions.py @@ -0,0 +1,264 @@ +""" +用户凭据、并发管理员保护与设置类型的审查回归。 +""" + +import asyncio +import threading +from io import BytesIO +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest +from fastapi import UploadFile +from pydantic import ValidationError +from sqlalchemy import event, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine + +from app.api.endpoints.user import create_user, upload_avatar +from app.application.github_auth import GithubAuthService +from app.application.security.throttle import LoginThrottle +from app.application.security.user import LastActiveSuperuserError +from app.application.settings.service import SystemSettingsService +from app.db.adapters.configuration import TransactionalUserConfigurationRepository +from app.db.adapters.user import SqlAlchemyUserRepository +from app.db.models.user import User +from app.db.models.userconfig import UserConfig +from app.db.oper.userconfig import UserConfigOper +from app.domain.github import GithubAuthTransportError, GithubTokenExchange, GithubUser +from app.schemas.initialization import InitializationRequest +from app.schemas.types import SystemConfigKey +from app.schemas.user import UserCreate +from tests.test_user_repository import _user_write_context + + +@pytest.mark.asyncio +async def test_avatar_contains_decodable_base64() -> None: + """ + 头像 data URL 只携带 Base64 文本,不包含 bytes 的 Python 表示。 + """ + import base64 + + service = SimpleNamespace(get_by_id=AsyncMock(return_value=object()), update=AsyncMock()) + file = UploadFile(filename="avatar.ico", file=BytesIO(b"review-image")) + response = await upload_avatar(7, service, file, SimpleNamespace(id=7, is_superuser=False)) + assert response.success + encoded = service.update.call_args.args[1]["avatar"].split(",", 1)[1] + assert base64.b64decode(encoded, validate=True) == b"review-image" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("password", [None, ""]) +async def test_create_user_removes_empty_optional_password(password) -> None: + """ + 可选密码为空时移除明文字段,创建无本地密码用户仍符合合同。 + """ + service = SimpleNamespace(create=AsyncMock(return_value=object())) + response = await create_user(service=service, user_in=UserCreate(name="review", password=password)) + assert response.success + assert "password" not in service.create.call_args.args[0] + assert "hashed_password" not in service.create.call_args.args[0] + + +@pytest.mark.parametrize("field,value", [("username", " "), ("api_key", " " * 16)]) +def test_initialization_rejects_blank_after_normalization(field, value) -> None: + """ + 首尾空白在长度约束之前去除,空账号和空令牌不能完成初始化。 + """ + payload = {"username": "review", "password": "Test123!", "confirm_password": "Test123!", + "api_key": "review-api-token-12345678", field: value} + with pytest.raises(ValidationError): + InitializationRequest.model_validate(payload) + + +def test_login_failure_ttl_resets_same_key() -> None: + """ + 同一用户隔超时再次失败从首次计数开始,未过期锁仍按预算保留。 + """ + now = [0.0] + throttle = LoginThrottle(max_failures=2, ttl_seconds=10, clock=lambda: now[0]) + key = ("127.0.0.1", "review") + assert throttle.record_failure(key) == 0 + now[0] = 11 + assert throttle.retry_after(key) == 0 + assert throttle.record_failure(key) == 0 + assert throttle.record_failure(key) == 30 + now[0] = 22 + assert throttle.retry_after(key) == 19 + + +@pytest.mark.asyncio +async def test_settings_persists_type_adapter_result() -> None: + """ + 合同接受的站点数字文本持久化为整数,消费者按站点主键匹配。 + """ + saved = {} + async def write(key, value, **_kwargs): + """ + 在内存端口记录已规范化候选,保留真实服务的类型处理。 + """ + saved[key] = value + return SimpleNamespace(normalized_value=value, changed=True) + config = SimpleNamespace(get=lambda key: saved.get(key), async_set_with_normalized_value=write) + service = SystemSettingsService(None, config, AsyncMock()) + await service.update(setting_key="IndexerSites", value=["1", "2"]) + assert saved[SystemConfigKey.IndexerSites] == [1, 2] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mutation", ["demote", "disable", "delete"]) +async def test_concurrent_sqlite_admin_changes_keep_one(tmp_path, mutation) -> None: + """ + 两个独占 SQLite 请求同时移除不同管理员时,后一个事务必须看到最新人数。 + """ + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'users.db'}") + try: + async with engine.begin() as connection: + await connection.run_sync(User.__table__.create) + async with AsyncSession(engine) as session: + session.add_all([User(name="a", is_active=True, is_superuser=True), + User(name="b", is_active=True, is_superuser=True)]) + await session.commit() + first_flushed, release_first = asyncio.Event(), asyncio.Event() + async def change(user_id, first=False): + """ + 首事务在提交前暂停,确保次事务遇到仍未落账的管理员修改。 + """ + async with AsyncSession(engine, expire_on_commit=False) as session: + repository = SqlAlchemyUserRepository(session) + try: + if mutation == "delete": + await repository.async_delete(user_id) + else: + await repository.async_update(user_id, { + "is_superuser" if mutation == "demote" else "is_active": False}) + if first: + first_flushed.set() + await release_first.wait() + await session.commit() + return "changed" + except LastActiveSuperuserError: + await session.rollback() + return "protected" + first = asyncio.create_task(change(1, first=True)) + await asyncio.wait_for(first_flushed.wait(), 5) + second = asyncio.create_task(change(2)) + try: + # 次事务必须等首事务提交,或在首事务完成后以新快照拒绝。 + await asyncio.sleep(0.05) + finally: + release_first.set() + assert sorted(await asyncio.wait_for(asyncio.gather(first, second), 5)) == ["changed", "protected"] + async with AsyncSession(engine) as session: + remaining = (await session.execute(select(User).where( + User.is_active.is_(True), User.is_superuser.is_(True)))).scalars().all() + assert len(remaining) == 1 + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_sqlite_userconfig_waits_for_user_delete_before_reading(monkeypatch, tmp_path) -> None: + """ + 删除已暂存但未提交时,配置写入先等数据库写锁,不能先读旧行再过期更新。 + """ + async with _user_write_context(tmp_path / "delete-config.db") as (session, sync_factory): + user = User(name="old", is_active=True, is_superuser=False) + session.add_all([user, UserConfig(username="old", key="theme", value="initial")]) + await session.commit() + snapshot = object.__new__(UserConfigOper) + snapshot.__init__() + configuration = TransactionalUserConfigurationRepository(sync_factory, snapshot) + configuration.load_snapshot() + await SqlAlchemyUserRepository(session).async_delete(user.id) + attempted, read_entered = threading.Event(), threading.Event() + original_read = UserConfig.get_by_key + + def record_read(**kwargs): + """ + 观察真实查询入口,保留查询和持久化算法。 + """ + read_entered.set() + return original_read(**kwargs) + + def before_sql(_connection, _cursor, _statement, _parameters, _context, _executemany): + """ + 在独立同步连接第一次尝试执行 SQL 时唤醒测试。 + """ + attempted.set() + + monkeypatch.setattr(UserConfig, "get_by_key", record_read) + bind = sync_factory.kw["bind"] + event.listen(bind, "before_cursor_execute", before_sql) + writing = asyncio.create_task(asyncio.to_thread(configuration.set, "old", "theme", "concurrent")) + try: + assert await asyncio.to_thread(attempted.wait, 5) + assert not read_entered.is_set() + finally: + await session.commit() + outcomes = await asyncio.wait_for(asyncio.gather(writing, return_exceptions=True), 5) + event.remove(bind, "before_cursor_execute", before_sql) + assert len(outcomes) == 1 and isinstance(outcomes[0], IntegrityError), outcomes + with sync_factory() as verify: + assert verify.execute(select(UserConfig)).scalars().all() == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("action", ["clear", "manual"]) +@pytest.mark.parametrize("refresh_fails", [False, True]) +async def test_github_refresh_does_not_restore_replaced_credentials(action, refresh_fails) -> None: + """ + 远端刷新等待期间清除或更换凭据,迟到结果不能恢复旧账号授权。 + """ + settings = Mock() + token = ["old-access"] + metadata = [{"login": "old-user", "expires_at": 1000, "refresh_token": "old-refresh"}] + entered, release = asyncio.Event(), asyncio.Event() + settings.get.side_effect = lambda _key: token[0] + def update(_key, value): + """ + 更新内存部署设置,避免写入真实配置。 + """ + token[0] = value + return True, "" + settings.update.side_effect = update + config = Mock() + config.get.side_effect = lambda _key: dict(metadata[0]) if metadata[0] else None + async def set_config(_key, value): + """ + 保存独立的 OAuth 元数据快照。 + """ + metadata[0] = dict(value) + async def delete_config(_key): + """ + 删除内存 OAuth 快照。 + """ + metadata[0] = None + config.async_set = set_config + config.async_delete = delete_config + async def refresh(*_args): + """ + 在远端响应边界制造可控并发窗口。 + """ + entered.set() + await release.wait() + if refresh_fails: + raise GithubAuthTransportError("隔离刷新失败") + return GithubTokenExchange(access_token="stale-refreshed", refresh_token="rotated", expires_in=3600) + transport = SimpleNamespace(refresh_access_token=refresh, + get_user=AsyncMock(return_value=GithubUser(login="manual-user"))) + service = GithubAuthService(settings=settings, system_config=config, transport=transport, + client_id="review", clock=lambda: 1000) + refreshing = asyncio.create_task(service.access_token()) + await asyncio.wait_for(entered.wait(), 5) + try: + if action == "clear": + await service.clear_token() + else: + await service.set_manual_token("new-manual") + finally: + release.set() + result = await asyncio.wait_for(refreshing, 5) + assert token[0] == (None if action == "clear" else "new-manual") + assert result == ("" if action == "clear" else "new-manual") + assert metadata[0] is None