Spaces:
Paused
Paused
| import json | |
| import logging | |
| import threading | |
| import time | |
| from autoteam import accounts, api, config | |
| def test_get_status_normalizes_main_account_status_from_saved_auth(tmp_path, monkeypatch): | |
| main_email = "owner@example.com" | |
| auth_file = tmp_path / "codex-main.json" | |
| auth_file.write_text(json.dumps({"access_token": "token-main"}), encoding="utf-8") | |
| monkeypatch.setattr( | |
| "autoteam.accounts.load_accounts", | |
| lambda: [ | |
| { | |
| "email": main_email, | |
| "status": "exhausted", | |
| "auth_file": "/app/auths/codex-main.json", | |
| "last_quota": { | |
| "primary_pct": 8, | |
| "primary_resets_at": 1710000000, | |
| "weekly_pct": 1, | |
| "weekly_resets_at": 1710600000, | |
| }, | |
| } | |
| ], | |
| ) | |
| monkeypatch.setattr(api, "_is_main_account_email", lambda email: email == main_email) | |
| monkeypatch.setattr("autoteam.codex_auth.get_saved_main_auth_file", lambda: str(auth_file)) | |
| monkeypatch.setattr( | |
| "autoteam.codex_auth.check_codex_quota", | |
| lambda access_token: ( | |
| "ok", | |
| { | |
| "primary_pct": 8, | |
| "primary_resets_at": 1710000000, | |
| "weekly_pct": 1, | |
| "weekly_resets_at": 1710600000, | |
| }, | |
| ), | |
| ) | |
| result = api.get_status() | |
| assert result["quota_cache"][main_email]["primary_pct"] == 8 | |
| assert result["accounts"][0]["is_main_account"] is True | |
| assert result["accounts"][0]["status"] == "active" | |
| assert result["summary"] == { | |
| "active": 1, | |
| "standby": 0, | |
| "exhausted": 0, | |
| "pending": 0, | |
| "personal": 0, | |
| "auth_invalid": 0, | |
| "orphan": 0, | |
| "disabled": 0, | |
| "total": 1, | |
| } | |
| def test_get_status_survives_runtime_resource_probe_failure(monkeypatch): | |
| monkeypatch.setattr("autoteam.accounts.load_accounts", lambda: []) | |
| monkeypatch.setattr( | |
| api, | |
| "collect_runtime_resource_snapshot", | |
| lambda: (_ for _ in ()).throw(RuntimeError("proc unavailable")), | |
| ) | |
| result = api.get_status() | |
| assert result["accounts"] == [] | |
| assert result["runtime_resources"] == {"error": "runtime_resource_snapshot_failed"} | |
| def test_get_status_includes_ipv6_pool_status(monkeypatch): | |
| monkeypatch.setattr("autoteam.accounts.load_accounts", lambda: []) | |
| monkeypatch.setattr( | |
| "autoteam.ipv6_pool.ipv6_pool.status", | |
| lambda: {"enabled": True, "required": False, "ok": True, "count": 2, "entries": []}, | |
| ) | |
| result = api.get_status() | |
| assert result["ipv6_pool"]["enabled"] is True | |
| assert result["ipv6_pool"]["count"] == 2 | |
| def test_get_status_exposes_clipproxy_and_rotation_validation(tmp_path, monkeypatch): | |
| auth_file = tmp_path / "codex-child.json" | |
| auth_file.write_text(json.dumps({"access_token": "token-child"}), encoding="utf-8") | |
| monkeypatch.setattr(api, "_is_main_account_email", lambda _email: False) | |
| monkeypatch.setattr( | |
| "autoteam.accounts.load_accounts", | |
| lambda: [{"email": "child@example.com", "status": accounts.STATUS_ACTIVE, "auth_file": str(auth_file)}], | |
| ) | |
| monkeypatch.setattr("autoteam.codex_auth.check_codex_quota", lambda *_args, **_kwargs: ("ok", {"primary_pct": 20})) | |
| monkeypatch.setattr( | |
| "autoteam.cliproxy_health.get_cliproxy_health", | |
| lambda: { | |
| "ok": False, | |
| "safe_read_only": True, | |
| "management_api": {"ok": True}, | |
| "provider_auth": { | |
| "ok": False, | |
| "provider": "codex", | |
| "model": "gpt-5.5", | |
| "reason": "no_provider_auth", | |
| "total": 0, | |
| "available": 0, | |
| "check_type": "management_metadata", | |
| "canary_required": True, | |
| }, | |
| }, | |
| ) | |
| monkeypatch.setattr( | |
| api, | |
| "_rotation_validation_cooldown", | |
| {"next_rotate_after": 0.0, "recorded_at": 123.0, "severity": "ok", "reason": ""}, | |
| ) | |
| result = api.get_status() | |
| assert result["summary"]["active"] == 1 | |
| assert result["cliproxy"]["safe_read_only"] is True | |
| assert result["cliproxy"]["provider_auth"]["reason"] == "no_provider_auth" | |
| assert result["rotation_validation"]["severity"] == "ok" | |
| assert result["rotation_validation"]["cooldown_remaining_seconds"] == 0 | |
| def test_get_status_fast_skips_live_quota_probe(tmp_path, monkeypatch): | |
| auth_file = tmp_path / "codex-child.json" | |
| auth_file.write_text(json.dumps({"access_token": "token-child"}), encoding="utf-8") | |
| monkeypatch.setattr(api, "_is_main_account_email", lambda _email: False) | |
| monkeypatch.setattr( | |
| "autoteam.accounts.load_accounts", | |
| lambda: [{"email": "child@example.com", "status": accounts.STATUS_ACTIVE, "auth_file": str(auth_file)}], | |
| ) | |
| monkeypatch.setattr( | |
| "autoteam.codex_auth.check_codex_quota", | |
| lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("fast status must not probe quota")), | |
| ) | |
| health_calls = [] | |
| monkeypatch.setattr( | |
| "autoteam.cliproxy_health.get_cliproxy_health", | |
| lambda **kwargs: health_calls.append(kwargs) | |
| or { | |
| "ok": True, | |
| "safe_read_only": True, | |
| "management_api": {"ok": True}, | |
| "provider_auth": { | |
| "ok": True, | |
| "provider": "codex", | |
| "model": "gpt-5.5", | |
| "reason": "provider_auth_has_candidates", | |
| "total": 1, | |
| "available": 1, | |
| "check_type": "management_metadata", | |
| "canary_required": True, | |
| }, | |
| }, | |
| ) | |
| result = api.get_status(fast=True) | |
| assert result["quota_cache"] == {} | |
| assert result["status_mode"] == {"fast": True, "quota_budget_seconds": 0, "live_quota": False} | |
| assert health_calls == [{"timeout": 0.5, "cache_ttl": 30.0}] | |
| def test_get_playwright_context_options_uses_fingerprint_constants(monkeypatch): | |
| monkeypatch.setattr(config, "PLAYWRIGHT_USER_AGENT", "AutoTeamTest/1.0") | |
| monkeypatch.setattr(config, "PLAYWRIGHT_LOCALE", "zh-CN") | |
| monkeypatch.setattr(config, "PLAYWRIGHT_TIMEZONE_ID", "Asia/Shanghai") | |
| monkeypatch.setattr(config, "PLAYWRIGHT_VIEWPORT_WIDTH", 1440) | |
| monkeypatch.setattr(config, "PLAYWRIGHT_VIEWPORT_HEIGHT", 900) | |
| monkeypatch.setattr(config, "PLAYWRIGHT_DEVICE_SCALE_FACTOR", 2) | |
| monkeypatch.setattr(config, "PLAYWRIGHT_COLOR_SCHEME", "light") | |
| options = config.get_playwright_context_options() | |
| assert options["viewport"] == {"width": 1440, "height": 900} | |
| assert options["user_agent"] == "AutoTeamTest/1.0" | |
| assert options["locale"] == "zh-CN" | |
| assert options["timezone_id"] == "Asia/Shanghai" | |
| assert options["device_scale_factor"] == 2 | |
| assert options["color_scheme"] == "light" | |
| assert options["extra_http_headers"] == {"Accept-Language": "zh-CN,zh;q=0.9"} | |
| def test_auto_check_cooldown_does_not_delay_real_team_shortage(tmp_path, monkeypatch): | |
| auth_files = [] | |
| for idx in range(1): | |
| auth_file = tmp_path / f"active-{idx}.json" | |
| auth_file.write_text(json.dumps({"access_token": f"token-{idx}"}), encoding="utf-8") | |
| auth_files.append(auth_file) | |
| started = [] | |
| def fake_start_task(command, func, params, *args, **kwargs): | |
| started.append((command, params, args, kwargs)) | |
| monkeypatch.setattr(api, "_auto_fill_last_trigger_ts", time.time()) | |
| monkeypatch.setattr(api, "_auto_check_config", {"interval": 0, "target_seats": 3, "threshold": 10, "min_low": 1}) | |
| monkeypatch.setattr(api, "log_runtime_resource_snapshot", lambda *args, **kwargs: {}) | |
| monkeypatch.setattr(api, "_is_main_account_email", lambda _email: False) | |
| monkeypatch.setattr("autoteam.sync_targets.is_sync_target_enabled", lambda _target: False) | |
| monkeypatch.setattr(api, "_auto_check_team_member_count", lambda *args, **kwargs: 2) | |
| monkeypatch.setattr(api, "_start_task", fake_start_task) | |
| monkeypatch.setattr( | |
| "autoteam.accounts.load_accounts", | |
| lambda: [ | |
| {"email": f"active-{idx}@example.com", "status": "active", "auth_file": str(auth_files[idx])} | |
| for idx in range(1) | |
| ], | |
| ) | |
| stop_event = threading.Event() | |
| restart_event = threading.Event() | |
| wait_calls = {"count": 0} | |
| def fake_wait(_seconds): | |
| wait_calls["count"] += 1 | |
| return wait_calls["count"] > 1 | |
| monkeypatch.setattr(stop_event, "wait", fake_wait) | |
| monkeypatch.setattr(api, "_auto_check_stop", stop_event) | |
| monkeypatch.setattr(api, "_auto_check_restart", restart_event) | |
| api._auto_check_loop() | |
| assert len(started) == 1 | |
| command, params, args, kwargs = started[0] | |
| assert command == "auto-fill" | |
| assert params == {"target_seats": 3} | |
| assert args == (3,) | |
| assert kwargs == {"background_post_sync": True} | |
| def test_auto_check_cooldown_keeps_full_team_from_refilling(tmp_path, monkeypatch): | |
| auth_files = [] | |
| for idx in range(1): | |
| auth_file = tmp_path / f"active-{idx}.json" | |
| auth_file.write_text(json.dumps({"access_token": f"token-{idx}"}), encoding="utf-8") | |
| auth_files.append(auth_file) | |
| started = [] | |
| probed = [] | |
| def fake_team_count(*args, **kwargs): | |
| probed.append((args, kwargs)) | |
| return 3 | |
| monkeypatch.setattr(api, "_auto_fill_last_trigger_ts", time.time()) | |
| monkeypatch.setattr(api, "_auto_check_config", {"interval": 0, "target_seats": 3, "threshold": 10, "min_low": 1}) | |
| monkeypatch.setattr(api, "log_runtime_resource_snapshot", lambda *args, **kwargs: {}) | |
| monkeypatch.setattr(api, "_is_main_account_email", lambda _email: False) | |
| monkeypatch.setattr("autoteam.sync_targets.is_sync_target_enabled", lambda _target: False) | |
| monkeypatch.setattr(api, "_auto_check_team_member_count", fake_team_count) | |
| monkeypatch.setattr(api, "_start_task", lambda *args, **kwargs: started.append((args, kwargs))) | |
| monkeypatch.setattr( | |
| "autoteam.accounts.load_accounts", | |
| lambda: [ | |
| {"email": f"active-{idx}@example.com", "status": "active", "auth_file": str(auth_files[idx])} | |
| for idx in range(1) | |
| ], | |
| ) | |
| monkeypatch.setattr( | |
| "autoteam.codex_auth.check_codex_quota", | |
| lambda _token: ("ok", {"primary_pct": 10, "primary_resets_at": 0, "weekly_pct": 1}), | |
| ) | |
| stop_event = threading.Event() | |
| restart_event = threading.Event() | |
| wait_calls = {"count": 0} | |
| def fake_wait(_seconds): | |
| wait_calls["count"] += 1 | |
| return wait_calls["count"] > 1 | |
| monkeypatch.setattr(stop_event, "wait", fake_wait) | |
| monkeypatch.setattr(api, "_auto_check_stop", stop_event) | |
| monkeypatch.setattr(api, "_auto_check_restart", restart_event) | |
| api._auto_check_loop() | |
| assert probed | |
| assert started == [] | |
| def test_auto_check_team_member_count_carries_invites_and_occupancy(monkeypatch): | |
| monkeypatch.setattr(api, "_run_playwright_probe", lambda *args, **kwargs: {"count": 3, "invites": 1, "occupancy": 4}) | |
| result = api._auto_check_team_member_count() | |
| assert result == 3 | |
| assert api._team_member_invite_count(result) == 1 | |
| assert api._team_member_occupancy(result) == 4 | |
| def test_auto_check_skips_expensive_probe_when_task_is_running(monkeypatch, caplog): | |
| monkeypatch.setattr(api, "_auto_check_config", {"interval": 0, "target_seats": 3, "threshold": 10, "min_low": 1}) | |
| monkeypatch.setattr(api, "_auto_check_stop", threading.Event()) | |
| monkeypatch.setattr(api, "_auto_check_restart", threading.Event()) | |
| monkeypatch.setattr( | |
| "autoteam.accounts.load_accounts", | |
| lambda: (_ for _ in ()).throw(AssertionError("busy auto-check must not load accounts")), | |
| ) | |
| monkeypatch.setattr( | |
| "autoteam.cliproxy_health.get_cliproxy_health", | |
| lambda **_kwargs: (_ for _ in ()).throw(AssertionError("busy auto-check must not query CPA gate")), | |
| ) | |
| stop_event = api._auto_check_stop | |
| wait_calls = {"count": 0} | |
| def fake_wait(_seconds): | |
| wait_calls["count"] += 1 | |
| return wait_calls["count"] > 1 | |
| monkeypatch.setattr(stop_event, "wait", fake_wait) | |
| api._playwright_lock.acquire() | |
| try: | |
| with caplog.at_level(logging.INFO): | |
| api._auto_check_loop() | |
| finally: | |
| if api._playwright_lock.locked(): | |
| api._playwright_lock.release() | |
| assert "跳过本轮自动巡检昂贵探测" in caplog.text | |
| def test_auto_check_cooldown_allows_full_team_blocker_replacement(monkeypatch): | |
| started = [] | |
| def fake_start_task(command, func, params, *args, **kwargs): | |
| started.append((command, params, args, kwargs)) | |
| monkeypatch.setattr(api, "_auto_fill_last_trigger_ts", time.time()) | |
| monkeypatch.setattr(api, "_auto_check_config", {"interval": 0, "target_seats": 3, "threshold": 10, "min_low": 1}) | |
| monkeypatch.setattr(api, "log_runtime_resource_snapshot", lambda *args, **kwargs: {}) | |
| monkeypatch.setattr(api, "_is_main_account_email", lambda _email: False) | |
| monkeypatch.setattr("autoteam.sync_targets.is_sync_target_enabled", lambda _target: False) | |
| monkeypatch.setattr(api, "_auto_check_team_member_count", lambda *args, **kwargs: 3) | |
| monkeypatch.setattr(api, "_start_task", fake_start_task) | |
| monkeypatch.setattr( | |
| "autoteam.accounts.load_accounts", | |
| lambda: [ | |
| {"email": "healthy@example.com", "status": "active", "auth_file": ""}, | |
| {"email": "blocked@example.com", "status": "auth_invalid", "auth_file": ""}, | |
| ], | |
| ) | |
| stop_event = threading.Event() | |
| restart_event = threading.Event() | |
| wait_calls = {"count": 0} | |
| def fake_wait(_seconds): | |
| wait_calls["count"] += 1 | |
| return wait_calls["count"] > 1 | |
| monkeypatch.setattr(stop_event, "wait", fake_wait) | |
| monkeypatch.setattr(api, "_auto_check_stop", stop_event) | |
| monkeypatch.setattr(api, "_auto_check_restart", restart_event) | |
| api._auto_check_loop() | |
| assert len(started) == 1 | |
| command, params, args, kwargs = started[0] | |
| assert command == "auto-fill" | |
| assert params == {"target_seats": 3} | |
| assert args == (3,) | |
| assert kwargs == {"background_post_sync": True} | |
| def test_auto_check_uses_read_only_cpa_gate_when_team_full_and_no_credentials(monkeypatch, caplog): | |
| started = [] | |
| def fake_start_task(command, func, params, *args, **kwargs): | |
| started.append((command, params, args, kwargs)) | |
| monkeypatch.setattr(api, "_auto_fill_last_trigger_ts", 0.0) | |
| monkeypatch.setattr(api, "_auto_check_config", {"interval": 0, "target_seats": 3, "threshold": 10, "min_low": 1}) | |
| monkeypatch.setattr(api, "log_runtime_resource_snapshot", lambda *args, **kwargs: {}) | |
| monkeypatch.setattr(api, "_is_main_account_email", lambda _email: False) | |
| monkeypatch.setattr("autoteam.accounts.load_accounts", lambda: []) | |
| monkeypatch.setattr(api, "_auto_check_team_member_count", lambda *args, **kwargs: 3) | |
| monkeypatch.setattr(api, "_start_task", fake_start_task) | |
| monkeypatch.setattr("autoteam.sync_targets.is_sync_target_enabled", lambda target: target == "cpa") | |
| monkeypatch.setattr( | |
| "autoteam.cliproxy_health.get_cliproxy_health", | |
| lambda **_kwargs: { | |
| "ok": False, | |
| "safe_read_only": True, | |
| "management_api": {"ok": True}, | |
| "provider_auth": { | |
| "ok": False, | |
| "provider": "codex", | |
| "model": "gpt-5.5", | |
| "reason": "no_provider_auth", | |
| "total": 0, | |
| "available": 0, | |
| "check_type": "management_metadata", | |
| "canary_required": True, | |
| }, | |
| }, | |
| ) | |
| monkeypatch.setattr( | |
| "autoteam.cpa_sync.sync_to_cpa", | |
| lambda: (_ for _ in ()).throw(AssertionError("CPA gate must not sync")), | |
| ) | |
| monkeypatch.setattr( | |
| "autoteam.cpa_sync.upload_to_cpa", | |
| lambda _path: (_ for _ in ()).throw(AssertionError("CPA gate must not upload")), | |
| ) | |
| monkeypatch.setattr( | |
| "autoteam.cpa_sync.delete_from_cpa", | |
| lambda _name: (_ for _ in ()).throw(AssertionError("CPA gate must not delete")), | |
| ) | |
| stop_event = threading.Event() | |
| restart_event = threading.Event() | |
| wait_calls = {"count": 0} | |
| def fake_wait(_seconds): | |
| wait_calls["count"] += 1 | |
| return wait_calls["count"] > 1 | |
| monkeypatch.setattr(stop_event, "wait", fake_wait) | |
| monkeypatch.setattr(api, "_auto_check_stop", stop_event) | |
| monkeypatch.setattr(api, "_auto_check_restart", restart_event) | |
| with caplog.at_level(logging.INFO): | |
| api._auto_check_loop() | |
| assert len(started) == 1 | |
| command, params, args, kwargs = started[0] | |
| assert command == "auto-fill" | |
| assert params == { | |
| "target_seats": 3, | |
| "trigger": "auto-check", | |
| "shortage": 2, | |
| "cpa_credential_gate": { | |
| "enabled": True, | |
| "safe_read_only": True, | |
| "management_ok": True, | |
| "available": 0, | |
| "total": 0, | |
| "reason": "no_provider_auth", | |
| "zero_available": True, | |
| }, | |
| "provider_auth_below_target": True, | |
| "provider_auth_available": 0, | |
| "provider_auth_target": 2, | |
| } | |
| assert args == (3,) | |
| assert kwargs == {"background_post_sync": True} | |
| assert "CPA 可用凭证 gate" in caplog.text | |
| assert "CPA 可用凭证为 0/0" in caplog.text | |
| def test_auto_check_triggers_cleanup_when_pending_invite_exceeds_occupancy(monkeypatch): | |
| started = [] | |
| def fake_start_task(command, func, params, *args, **kwargs): | |
| started.append((command, params, args, kwargs)) | |
| monkeypatch.setattr(api, "_auto_fill_last_trigger_ts", 0.0) | |
| monkeypatch.setattr(api, "_auto_check_config", {"interval": 0, "target_seats": 3, "threshold": 10, "min_low": 1}) | |
| monkeypatch.setattr(api, "log_runtime_resource_snapshot", lambda *args, **kwargs: {}) | |
| monkeypatch.setattr(api, "_is_main_account_email", lambda _email: False) | |
| monkeypatch.setattr("autoteam.accounts.load_accounts", lambda: []) | |
| monkeypatch.setattr( | |
| api, | |
| "_auto_check_team_member_count", | |
| lambda *args, **kwargs: api._TeamMemberCount(3, invites=1, occupancy=4), | |
| ) | |
| monkeypatch.setattr(api, "_start_task", fake_start_task) | |
| monkeypatch.setattr("autoteam.sync_targets.is_sync_target_enabled", lambda target: False) | |
| stop_event = threading.Event() | |
| restart_event = threading.Event() | |
| wait_calls = {"count": 0} | |
| def fake_wait(_seconds): | |
| wait_calls["count"] += 1 | |
| return wait_calls["count"] > 1 | |
| monkeypatch.setattr(stop_event, "wait", fake_wait) | |
| monkeypatch.setattr(api, "_auto_check_stop", stop_event) | |
| monkeypatch.setattr(api, "_auto_check_restart", restart_event) | |
| api._auto_check_loop() | |
| assert len(started) == 1 | |
| command, params, args, kwargs = started[0] | |
| assert command == "auto-cleanup" | |
| assert params["max_seats"] == 3 | |
| assert params["trigger"] == "auto-check" | |
| assert params["team_count"] == 3 | |
| assert params["invite_count"] == 1 | |
| assert params["team_occupancy"] == 4 | |
| assert args == (3,) | |
| assert kwargs == {} | |
| def test_auto_check_triggers_preventive_rotate_when_provider_auth_below_target(tmp_path, monkeypatch, caplog): | |
| auth_ok = tmp_path / "active-ok.json" | |
| auth_other = tmp_path / "active-other.json" | |
| auth_ok.write_text(json.dumps({"access_token": "token-ok"}), encoding="utf-8") | |
| auth_other.write_text(json.dumps({"access_token": "token-other"}), encoding="utf-8") | |
| started = [] | |
| def fake_start_task(command, func, params, *args, **kwargs): | |
| started.append((command, params, args, kwargs)) | |
| monkeypatch.setattr(api, "_auto_fill_last_trigger_ts", 0.0) | |
| monkeypatch.setattr(api, "_auto_check_config", {"interval": 0, "target_seats": 3, "threshold": 10, "min_low": 1}) | |
| monkeypatch.setattr(api, "log_runtime_resource_snapshot", lambda *args, **kwargs: {}) | |
| monkeypatch.setattr(api, "_is_main_account_email", lambda _email: False) | |
| monkeypatch.setattr( | |
| "autoteam.accounts.load_accounts", | |
| lambda: [ | |
| {"email": "one@example.com", "status": "active", "auth_file": str(auth_ok)}, | |
| {"email": "two@example.com", "status": "active", "auth_file": str(auth_other)}, | |
| ], | |
| ) | |
| monkeypatch.setattr( | |
| "autoteam.codex_auth.check_codex_quota", | |
| lambda _token: ("ok", {"primary_pct": 10, "primary_resets_at": 1234567890, "weekly_pct": 1}), | |
| ) | |
| monkeypatch.setattr( | |
| api, | |
| "_auto_check_team_member_count", | |
| lambda *args, **kwargs: api._TeamMemberCount(3, invites=0, occupancy=3), | |
| ) | |
| monkeypatch.setattr(api, "_start_task", fake_start_task) | |
| monkeypatch.setattr("autoteam.sync_targets.is_sync_target_enabled", lambda target: target == "cpa") | |
| monkeypatch.setattr( | |
| "autoteam.cliproxy_health.get_cliproxy_health", | |
| lambda **_kwargs: { | |
| "ok": True, | |
| "safe_read_only": True, | |
| "management_api": {"ok": True}, | |
| "provider_auth": { | |
| "ok": True, | |
| "provider": "codex", | |
| "model": "gpt-5.5", | |
| "reason": "provider_auth_has_candidates", | |
| "total": 4, | |
| "available": 1, | |
| "check_type": "management_metadata", | |
| "canary_required": True, | |
| }, | |
| }, | |
| ) | |
| stop_event = threading.Event() | |
| restart_event = threading.Event() | |
| wait_calls = {"count": 0} | |
| def fake_wait(_seconds): | |
| wait_calls["count"] += 1 | |
| return wait_calls["count"] > 1 | |
| monkeypatch.setattr(stop_event, "wait", fake_wait) | |
| monkeypatch.setattr(api, "_auto_check_stop", stop_event) | |
| monkeypatch.setattr(api, "_auto_check_restart", restart_event) | |
| with caplog.at_level(logging.INFO): | |
| api._auto_check_loop() | |
| assert len(started) == 1 | |
| command, params, args, kwargs = started[0] | |
| assert command == "auto-rotate" | |
| assert params["target"] == 3 | |
| assert params["trigger"] == "auto-check" | |
| assert params["shortage"] == 1 | |
| assert params["low_accounts"] == 0 | |
| assert params["provider_auth_below_target"] is True | |
| assert params["provider_auth_available"] == 1 | |
| assert params["provider_auth_target"] == 2 | |
| assert params["cpa_credential_gate"]["available"] == 1 | |
| assert args == (3,) | |
| assert kwargs == {"background_post_sync": True} | |
| assert "provider-auth 可用凭证低水位" in caplog.text | |
| def test_auto_check_cpa_gate_does_not_treat_management_failure_as_zero_credentials(monkeypatch, caplog): | |
| started = [] | |
| monkeypatch.setattr(api, "_auto_fill_last_trigger_ts", 0.0) | |
| monkeypatch.setattr(api, "_auto_check_config", {"interval": 0, "target_seats": 3, "threshold": 10, "min_low": 1}) | |
| monkeypatch.setattr(api, "log_runtime_resource_snapshot", lambda *args, **kwargs: {}) | |
| monkeypatch.setattr(api, "_is_main_account_email", lambda _email: False) | |
| monkeypatch.setattr("autoteam.accounts.load_accounts", lambda: []) | |
| monkeypatch.setattr(api, "_auto_check_team_member_count", lambda *args, **kwargs: 3) | |
| monkeypatch.setattr(api, "_start_task", lambda *args, **kwargs: started.append((args, kwargs))) | |
| monkeypatch.setattr("autoteam.sync_targets.is_sync_target_enabled", lambda target: target == "cpa") | |
| monkeypatch.setattr( | |
| "autoteam.cliproxy_health.get_cliproxy_health", | |
| lambda **_kwargs: { | |
| "ok": False, | |
| "safe_read_only": True, | |
| "management_api": {"ok": False, "reason": "request_failed"}, | |
| "provider_auth": { | |
| "ok": False, | |
| "provider": "codex", | |
| "model": "gpt-5.5", | |
| "reason": "management_api_unavailable", | |
| "total": 0, | |
| "available": 0, | |
| "check_type": "management_metadata", | |
| "canary_required": True, | |
| }, | |
| }, | |
| ) | |
| stop_event = threading.Event() | |
| restart_event = threading.Event() | |
| wait_calls = {"count": 0} | |
| def fake_wait(_seconds): | |
| wait_calls["count"] += 1 | |
| return wait_calls["count"] > 1 | |
| monkeypatch.setattr(stop_event, "wait", fake_wait) | |
| monkeypatch.setattr(api, "_auto_check_stop", stop_event) | |
| monkeypatch.setattr(api, "_auto_check_restart", restart_event) | |
| with caplog.at_level(logging.INFO): | |
| api._auto_check_loop() | |
| assert started == [] | |
| assert "CPA 可用凭证 gate" in caplog.text | |
| assert "management_ok=False" in caplog.text | |
| def test_collect_cpa_credential_gate_requires_safe_read_only_for_zero_available(monkeypatch): | |
| monkeypatch.setattr("autoteam.sync_targets.is_sync_target_enabled", lambda target: target == "cpa") | |
| monkeypatch.setattr( | |
| "autoteam.cliproxy_health.get_cliproxy_health", | |
| lambda **_kwargs: { | |
| "ok": True, | |
| "safe_read_only": False, | |
| "management_api": {"ok": True}, | |
| "provider_auth": { | |
| "ok": True, | |
| "provider": "codex", | |
| "model": "gpt-5.5", | |
| "reason": "provider_auth_has_candidates", | |
| "total": 2, | |
| "available": 0, | |
| "check_type": "management_metadata", | |
| "canary_required": True, | |
| }, | |
| }, | |
| ) | |
| gate = api._collect_cpa_credential_gate() | |
| assert gate["safe_read_only"] is False | |
| assert gate["zero_available"] is False | |
| assert api._cpa_provider_auth_below_pool_target(gate, 2) is False | |
| def test_runtime_validation_degrades_when_provider_auth_below_target(tmp_path, monkeypatch): | |
| auth_one = tmp_path / "one.json" | |
| auth_two = tmp_path / "two.json" | |
| auth_one.write_text(json.dumps({"access_token": "token-one"}), encoding="utf-8") | |
| auth_two.write_text(json.dumps({"access_token": "token-two"}), encoding="utf-8") | |
| monkeypatch.setattr(api, "_auto_check_config", {"interval": 0, "target_seats": 3, "threshold": 10, "min_low": 1}) | |
| monkeypatch.setattr(api, "_is_main_account_email", lambda _email: False) | |
| monkeypatch.setattr(api, "_auto_check_team_member_count", lambda *args, **kwargs: 3) | |
| monkeypatch.setattr( | |
| "autoteam.accounts.load_accounts", | |
| lambda: [ | |
| {"email": "one@example.com", "status": accounts.STATUS_ACTIVE, "auth_file": str(auth_one)}, | |
| {"email": "two@example.com", "status": accounts.STATUS_ACTIVE, "auth_file": str(auth_two)}, | |
| ], | |
| ) | |
| monkeypatch.setattr("autoteam.codex_auth.check_codex_quota", lambda _token: ("ok", {"primary_pct": 10})) | |
| monkeypatch.setattr( | |
| api, | |
| "_collect_cpa_credential_gate", | |
| lambda: { | |
| "enabled": True, | |
| "safe_read_only": True, | |
| "management_ok": True, | |
| "available": 1, | |
| "total": 4, | |
| "reason": "provider_auth_has_candidates", | |
| "zero_available": False, | |
| }, | |
| ) | |
| validation = api._log_task_runtime_validation("auto-rotate") | |
| assert validation["ok"] is False | |
| assert validation["severity"] == "degraded" | |
| assert validation["operation_success"] is True | |
| assert validation["pool_health_ok"] is False | |
| assert validation["followup_required"] is True | |
| assert validation["provider_auth_available"] == 1 | |
| assert validation["reason"] == "provider_auth_available=1/2" | |
| assert validation["cpa_credential_gate"]["available"] == 1 | |
| def test_sanitize_account_keeps_exportable_main_account_active_without_live_quota(tmp_path, monkeypatch): | |
| main_email = "owner@example.com" | |
| auth_file = tmp_path / "codex-main.json" | |
| auth_file.write_text("{}", encoding="utf-8") | |
| monkeypatch.setattr(api, "_is_main_account_email", lambda email: email == main_email) | |
| monkeypatch.setattr("autoteam.codex_auth.get_saved_main_auth_file", lambda: str(auth_file)) | |
| sanitized = api._sanitize_account( | |
| {"email": main_email, "status": "exhausted", "auth_file": "/app/auths/missing.json"} | |
| ) | |
| assert sanitized["is_main_account"] is True | |
| assert sanitized["status"] == "active" | |
| def test_sanitize_account_masks_disabled_non_main_status(monkeypatch): | |
| monkeypatch.setattr(api, "_is_main_account_email", lambda _email: False) | |
| sanitized = api._sanitize_account({"email": "user@example.com", "status": "active", "disabled": True}) | |
| assert sanitized["raw_status"] == "active" | |
| assert sanitized["status"] == "disabled" | |
| assert sanitized["disabled"] is True | |
| def test_post_rotate_runs_final_sync_in_background(monkeypatch): | |
| started = [] | |
| rotate_calls = [] | |
| def fake_start_task(command, func, params, *args, **kwargs): | |
| started.append((command, func, params, args, kwargs)) | |
| return {"task_id": "rotate-task", "command": command, "params": params} | |
| monkeypatch.setattr(api, "_start_task", fake_start_task) | |
| monkeypatch.setattr( | |
| "autoteam.manager.cmd_rotate", | |
| lambda *args, **kwargs: rotate_calls.append((args, kwargs)), | |
| ) | |
| result = api.post_rotate(api.TaskParams(target=3)) | |
| assert result["task_id"] == "rotate-task" | |
| assert len(started) == 1 | |
| command, func, params, args, kwargs = started[0] | |
| assert command == "rotate" | |
| assert params == {"target": 3} | |
| assert args == (3,) | |
| assert kwargs == {} | |
| func(3) | |
| assert rotate_calls == [((3,), {"force_auth_repair": True, "background_post_sync": True})] | |
| def test_post_fill_defers_sync_and_status_to_background(monkeypatch): | |
| started = [] | |
| fill_calls = [] | |
| sync_calls = [] | |
| def fake_start_task(command, func, params, *args, **kwargs): | |
| started.append((command, func, params, args, kwargs)) | |
| return {"task_id": "fill-task", "command": command, "params": params} | |
| monkeypatch.setattr(api, "_start_task", fake_start_task) | |
| monkeypatch.setattr(api._pw_executor, "run", lambda fn: (True, "ok", {})) | |
| monkeypatch.setattr( | |
| "autoteam.manager.cmd_fill", | |
| lambda *args, **kwargs: fill_calls.append((args, kwargs)) or {"filled": True}, | |
| ) | |
| monkeypatch.setattr( | |
| "autoteam.manager._schedule_post_task_sync", | |
| lambda stage_label: sync_calls.append(stage_label), | |
| ) | |
| result = api.post_fill(api.TaskParams(target=3)) | |
| assert result["task_id"] == "fill-task" | |
| assert len(started) == 1 | |
| command, func, params, args, kwargs = started[0] | |
| assert command == "fill" | |
| assert params == {"target": 3, "leave_workspace": False} | |
| assert args == (3,) | |
| assert kwargs == {"leave_workspace": False} | |
| assert func(3, leave_workspace=False) == {"filled": True} | |
| assert fill_calls == [ | |
| ((3,), {"leave_workspace": False, "post_sync": False, "print_status": False}) | |
| ] | |
| assert sync_calls == ["[填充]"] | |
| def test_disable_and_enable_account_toggle_local_flag(tmp_path, monkeypatch): | |
| accounts_file = tmp_path / "accounts.json" | |
| monkeypatch.setattr(accounts, "ACCOUNTS_FILE", accounts_file) | |
| monkeypatch.setattr(accounts, "get_admin_email", lambda: "owner@example.com") | |
| monkeypatch.setattr(api, "_is_main_account_email", lambda email: email == "owner@example.com") | |
| accounts.save_accounts( | |
| [ | |
| {"email": "member@example.com", "status": "standby", "disabled": False}, | |
| {"email": "owner@example.com", "status": "active", "disabled": False}, | |
| ] | |
| ) | |
| disabled_result = api.post_disable_account("member@example.com") | |
| enabled_result = api.post_enable_account("member@example.com") | |
| assert disabled_result["disabled"] is True | |
| assert disabled_result["account"]["status"] == "disabled" | |
| assert disabled_result["account"]["raw_status"] == "standby" | |
| assert enabled_result["disabled"] is False | |
| assert enabled_result["account"]["status"] == "standby" | |
| assert accounts.find_account(accounts.load_accounts(), "member@example.com")["disabled"] is False | |
| def test_bulk_disable_accounts_updates_multiple_rows_and_skips_non_targets(tmp_path, monkeypatch): | |
| accounts_file = tmp_path / "accounts.json" | |
| monkeypatch.setattr(accounts, "ACCOUNTS_FILE", accounts_file) | |
| monkeypatch.setattr(accounts, "get_admin_email", lambda: "owner@example.com") | |
| monkeypatch.setattr(api, "_is_main_account_email", lambda email: email == "owner@example.com") | |
| accounts.save_accounts( | |
| [ | |
| {"email": "first@example.com", "status": "standby", "disabled": False}, | |
| {"email": "second@example.com", "status": "active", "disabled": False}, | |
| {"email": "already@example.com", "status": "standby", "disabled": True}, | |
| {"email": "owner@example.com", "status": "active", "disabled": False}, | |
| ] | |
| ) | |
| result = api.post_bulk_disable_accounts( | |
| api.AccountDisableParams( | |
| emails=[ | |
| "first@example.com", | |
| "second@example.com", | |
| "already@example.com", | |
| "owner@example.com", | |
| "missing@example.com", | |
| "first@example.com", | |
| ] | |
| ) | |
| ) | |
| stored = {acc["email"]: acc for acc in accounts.load_accounts()} | |
| assert result["updated_count"] == 2 | |
| assert result["updated_emails"] == ["first@example.com", "second@example.com"] | |
| assert result["unchanged_emails"] == ["already@example.com"] | |
| assert result["skipped_main_accounts"] == ["owner@example.com"] | |
| assert result["missing_emails"] == ["missing@example.com"] | |
| assert stored["first@example.com"]["disabled"] is True | |
| assert stored["second@example.com"]["disabled"] is True | |
| assert stored["already@example.com"]["disabled"] is True | |
| assert stored["owner@example.com"]["disabled"] is False | |
| def test_bulk_enable_accounts_updates_disabled_rows_only(tmp_path, monkeypatch): | |
| accounts_file = tmp_path / "accounts.json" | |
| monkeypatch.setattr(accounts, "ACCOUNTS_FILE", accounts_file) | |
| monkeypatch.setattr(accounts, "get_admin_email", lambda: "owner@example.com") | |
| monkeypatch.setattr(api, "_is_main_account_email", lambda email: email == "owner@example.com") | |
| accounts.save_accounts( | |
| [ | |
| {"email": "first@example.com", "status": "standby", "disabled": True}, | |
| {"email": "second@example.com", "status": "active", "disabled": True}, | |
| {"email": "already@example.com", "status": "standby", "disabled": False}, | |
| {"email": "owner@example.com", "status": "active", "disabled": False}, | |
| ] | |
| ) | |
| result = api.post_bulk_enable_accounts( | |
| api.AccountDisableParams( | |
| emails=[ | |
| "first@example.com", | |
| "second@example.com", | |
| "already@example.com", | |
| "owner@example.com", | |
| "missing@example.com", | |
| ] | |
| ) | |
| ) | |
| stored = {acc["email"]: acc for acc in accounts.load_accounts()} | |
| assert result["updated_count"] == 2 | |
| assert result["updated_emails"] == ["first@example.com", "second@example.com"] | |
| assert result["unchanged_emails"] == ["already@example.com"] | |
| assert result["skipped_main_accounts"] == ["owner@example.com"] | |
| assert result["missing_emails"] == ["missing@example.com"] | |
| assert stored["first@example.com"]["disabled"] is False | |
| assert stored["second@example.com"]["disabled"] is False | |
| assert stored["already@example.com"]["disabled"] is False | |
| assert stored["owner@example.com"]["disabled"] is False | |
| def test_run_task_records_progress_history(monkeypatch): | |
| task_id = "progresshist" | |
| api._tasks[task_id] = { | |
| "task_id": task_id, | |
| "command": "auto-rotate", | |
| "params": {}, | |
| "status": "pending", | |
| "created_at": time.time(), | |
| "started_at": None, | |
| "finished_at": None, | |
| "result": None, | |
| "error": None, | |
| "last_progress_at": time.time(), | |
| "progress": "pending", | |
| "progress_history": [], | |
| } | |
| monkeypatch.setattr(api, "_log_task_runtime_validation", lambda _command: {"ok": True, "severity": "ok"}) | |
| def fake_work(): | |
| api.bump_task_progress("rotate:sync_team") | |
| api.bump_task_progress("rotate:check_quota") | |
| return {"ok": True} | |
| try: | |
| api._run_task(task_id, fake_work) | |
| task = api._tasks[task_id] | |
| stages = [item["stage"] for item in task["progress_history"]] | |
| assert stages == ["started", "rotate:sync_team", "rotate:check_quota", "runtime_validation", "finished"] | |
| assert task["progress"] == "rotate:check_quota" | |
| assert task["progress_history"][0]["duration_sec"] >= 0 | |
| assert task["progress_history"][-1]["stage"] == "finished" | |
| finally: | |
| api._tasks.pop(task_id, None) | |
| def test_get_status_counts_disabled_and_skips_disabled_quota_checks(tmp_path, monkeypatch): | |
| enabled_auth = tmp_path / "enabled.json" | |
| disabled_auth = tmp_path / "disabled.json" | |
| enabled_auth.write_text(json.dumps({"access_token": "token-enabled"}), encoding="utf-8") | |
| disabled_auth.write_text(json.dumps({"access_token": "token-disabled"}), encoding="utf-8") | |
| monkeypatch.setattr( | |
| "autoteam.accounts.load_accounts", | |
| lambda: [ | |
| {"email": "enabled@example.com", "status": "active", "auth_file": str(enabled_auth), "disabled": False}, | |
| {"email": "disabled@example.com", "status": "active", "auth_file": str(disabled_auth), "disabled": True}, | |
| ], | |
| ) | |
| monkeypatch.setattr(api, "_is_main_account_email", lambda _email: False) | |
| seen_tokens = [] | |
| def fake_check_quota(access_token): | |
| seen_tokens.append(access_token) | |
| return ( | |
| "ok", | |
| { | |
| "primary_pct": 15, | |
| "primary_resets_at": 1710000000, | |
| "weekly_pct": 5, | |
| "weekly_resets_at": 1710600000, | |
| }, | |
| ) | |
| monkeypatch.setattr("autoteam.codex_auth.check_codex_quota", fake_check_quota) | |
| result = api.get_status() | |
| assert seen_tokens == ["token-enabled"] | |
| assert {item["email"]: item["status"] for item in result["accounts"]} == { | |
| "enabled@example.com": "active", | |
| "disabled@example.com": "disabled", | |
| } | |
| assert result["summary"] == { | |
| "active": 1, | |
| "standby": 0, | |
| "exhausted": 0, | |
| "pending": 0, | |
| "personal": 0, | |
| "auth_invalid": 0, | |
| "orphan": 0, | |
| "disabled": 1, | |
| "total": 2, | |
| } | |
| def test_post_setup_save_keeps_cpa_url_required_and_generates_api_key(monkeypatch): | |
| written = {} | |
| def fake_write_env(key, value): | |
| written[key] = value | |
| monkeypatch.setattr("autoteam.setup_wizard._write_env", fake_write_env) | |
| monkeypatch.setattr("autoteam.setup_wizard._verify_cloudmail", lambda: True) | |
| monkeypatch.setattr("autoteam.setup_wizard._verify_cpa", lambda: True) | |
| monkeypatch.setattr("secrets.token_urlsafe", lambda _n: "generated-token") | |
| monkeypatch.setattr("importlib.reload", lambda module: module) | |
| monkeypatch.setattr(api, "API_KEY", "") | |
| monkeypatch.delenv("CPA_URL", raising=False) | |
| monkeypatch.delenv("API_KEY", raising=False) | |
| result = api.post_setup_save( | |
| api.SetupConfig( | |
| CLOUDMAIL_BASE_URL="http://mail.example.com", | |
| CLOUDMAIL_EMAIL="admin@example.com", | |
| CLOUDMAIL_PASSWORD="secret", | |
| CLOUDMAIL_DOMAIN="@example.com", | |
| CPA_URL="", | |
| CPA_KEY="key-1", | |
| PLAYWRIGHT_PROXY_URL="", | |
| PLAYWRIGHT_PROXY_BYPASS="", | |
| API_KEY="", | |
| ) | |
| ) | |
| assert written["CPA_URL"] == "http://127.0.0.1:8317" | |
| assert written["API_KEY"] == "generated-token" | |
| assert result["api_key"] == "generated-token" | |
| assert api.API_KEY == "generated-token" | |
| def test_post_setup_save_normalizes_rotation_account_strategy(monkeypatch): | |
| written = {} | |
| def fake_write_env(key, value): | |
| written[key] = value | |
| monkeypatch.setattr("autoteam.setup_wizard._write_env", fake_write_env) | |
| monkeypatch.setattr("autoteam.setup_wizard._verify_cloudmail", lambda: True) | |
| monkeypatch.setattr("autoteam.setup_wizard._verify_cpa", lambda: True) | |
| monkeypatch.setattr("importlib.reload", lambda module: module) | |
| monkeypatch.setattr(api, "API_KEY", "old-key") | |
| monkeypatch.setenv("API_KEY", "old-key") | |
| result = api.post_setup_save( | |
| api.SetupConfig( | |
| CLOUDMAIL_BASE_URL="http://mail.example.com", | |
| CLOUDMAIL_PASSWORD="secret", | |
| CLOUDMAIL_DOMAIN="@example.com", | |
| CPA_URL="http://127.0.0.1:8317", | |
| CPA_KEY="key-1", | |
| API_KEY="old-key", | |
| ROTATE_NEW_ACCOUNT_MODE="DOMAIN_AUTO_JOIN_FIRST", | |
| AUTOTEAM_AUTO_JOIN_DOMAINS="@Example.com, other.example.", | |
| ROTATE_DOMAIN_AUTO_JOIN_FALLBACK_INVITE=True, | |
| ) | |
| ) | |
| assert result["message"] == "配置保存成功" | |
| assert written["ROTATE_NEW_ACCOUNT_MODE"] == "domain_auto_join_first" | |
| assert written["AUTOTEAM_AUTO_JOIN_DOMAINS"] == "example.com,other.example" | |
| assert written["ROTATE_DOMAIN_AUTO_JOIN_FALLBACK_INVITE"] == "true" | |
| def test_post_setup_save_rejects_invalid_rotation_account_strategy(monkeypatch): | |
| monkeypatch.setattr("secrets.token_urlsafe", lambda _n: "generated-token") | |
| result = api.post_setup_save( | |
| api.SetupConfig( | |
| ROTATE_NEW_ACCOUNT_MODE="invite-maybe", | |
| ROTATE_DOMAIN_AUTO_JOIN_FALLBACK_INVITE="maybe", | |
| ) | |
| ) | |
| assert result.status_code == 400 | |
| body = json.loads(result.body.decode("utf-8")) | |
| assert body["message"] == "ROTATE_NEW_ACCOUNT_MODE 必须是 domain_auto_join_first、invite_first 或 direct_first" | |