import asyncio import json import os from pathlib import Path from core.settings import AppSettings from dashboard.api import _account_survival_payload from main import _apply_runtime_mode, _build_background_tasks, _recent_pool_files, _rotate_log_tail, _runtime_payload, _summary_payload, create_app from ops.scan import ScanResult def _request_via_asgi( app, method: str, path: str, headers: dict[str, str] | None = None, body: bytes = b"", ) -> tuple[int, dict[str, str], bytes]: request_headers = [ (key.lower().encode("latin-1"), value.encode("latin-1")) for key, value in (headers or {}).items() ] scope = { "type": "http", "asgi": {"version": "3.0"}, "http_version": "1.1", "method": method, "scheme": "http", "path": path, "raw_path": path.encode("ascii"), "query_string": b"", "headers": request_headers, "client": ("127.0.0.1", 12345), "server": ("testserver", 80), "root_path": "", "app": app, } response: dict[str, object] = {"status": 500, "headers": {}, "body": b""} request_sent = False async def receive() -> dict[str, object]: nonlocal request_sent if request_sent: return {"type": "http.disconnect"} request_sent = True return {"type": "http.request", "body": body, "more_body": False} async def send(message: dict[str, object]) -> None: if message["type"] == "http.response.start": response["status"] = int(message["status"]) response["headers"] = { key.decode("latin-1"): value.decode("latin-1") for key, value in message.get("headers", []) } return if message["type"] == "http.response.body": response["body"] = bytes(response["body"]) + bytes(message.get("body", b"")) asyncio.run(app(scope, receive, send)) return int(response["status"]), dict(response["headers"]), bytes(response["body"]) def test_summary_payload_exposes_register_and_ops_commands(tmp_path: Path) -> None: app = create_app(enable_background_tasks=False) app.state.settings = AppSettings( pool_dir=tmp_path, runtime_state_file=tmp_path / "runtime_state.json", cleanup_enabled=False, validate_enabled=False, ) app.state.background_tasks = [] payload = _summary_payload(app) commands = payload["commands"] routes = payload["routes"] assert commands["chatgpt_preflight"].startswith("uv run python scripts/chatgpt_preflight.py") assert commands["chatgpt_register_once"].startswith("uv run python scripts/chatgpt_register_once.py") assert commands["chatgpt_callback_exchange"].startswith("uv run python scripts/chatgpt_exchange_callback.py") assert commands["update_priority_dry_run"].startswith("uv run python -m ops.update_priority") assert commands["validate_used_dry_run"].startswith("uv run python -m ops.validate --scope used") assert routes["chatgpt_callback_exchange"] == "/api/register/chatgpt/callback-exchange" assert "dashboard" not in routes def test_dashboard_route_removed(tmp_path: Path) -> None: app = create_app(enable_background_tasks=False) app.state.settings = AppSettings( pool_dir=tmp_path, runtime_state_file=tmp_path / "runtime_state.json", cleanup_enabled=False, validate_enabled=False, ) app.state.background_tasks = [] status, _headers, _body = _request_via_asgi(app, "GET", "/dashboard") assert status == 404 def test_summary_payload_exposes_register_log_tail(tmp_path: Path) -> None: log_path = tmp_path / "register.log" log_path.write_text("line-1\nline-2\nline-3\n", encoding="utf-8") app = create_app(enable_background_tasks=False) app.state.settings = AppSettings( pool_dir=tmp_path, register_log_file=str(log_path), cleanup_enabled=False, validate_enabled=False, ) app.state.background_tasks = [] payload = _summary_payload(app) log_tail = payload["register_log_tail"] assert log_tail["available"] is True assert log_tail["path"] == str(log_path) assert log_tail["error"] is None assert log_tail["lines"] == ["line-1", "line-2", "line-3"] def test_summary_payload_exposes_rotate_log_tail_and_runtime_state_meta(monkeypatch, tmp_path: Path) -> None: runtime_state_file = tmp_path / "runtime_state.json" runtime_state_file.write_text('{"updated_at":"2026-03-23T22:25:02"}', encoding="utf-8") app = create_app(enable_background_tasks=False) app.state.settings = AppSettings( pool_dir=tmp_path, runtime_state_file=runtime_state_file, cleanup_enabled=False, validate_enabled=False, ) app.state.background_tasks = [] monkeypatch.setattr( "main._rotate_log_tail", lambda **_kwargs: { "available": True, "path": "/home/sophomores/zhuce6/logs/dashboard.log", "updated_at": 1774276100.0, "updated_at_iso": "2026-03-23T22:28:20", "error": None, "lines": [ "[22:18:20] [rotate] summary | 主池: 800 → 790 | 401删除: 10 | quota探测: 797 | probe401: 14 | probe429: 0 | probe跳过: 0 | 429删除: 0", ], "recent_events": [ "[22:17:35] [rotate] 🔎 ocd0553a1fb1@mail.example.test.json quota probe → 401 invalidated", "[22:17:35] [rotate] ❌ ocd0553a1fb1@mail.example.test.json 401删除", ], "latest_summary": { "time": "22:18:20", "main_before": 800, "main_after": 790, "deleted_401": 10, "quota_probed": 797, "quota_probe_401": 14, "quota_probe_429": 0, "quota_probe_skipped": 0, "deleted_429": 0, }, "current_summary": { "time": "22:19:10", "main_before": None, "main_after": None, "deleted_401": 1, "quota_probed": 1, "quota_probe_401": 1, "quota_probe_429": 0, "quota_probe_skipped": 0, "deleted_429": 0, "partial": True, "event_count": 2, }, }, ) payload = _summary_payload(app) assert payload["rotate_latest_summary"]["deleted_401"] == 10 assert payload["rotate_latest_summary"]["quota_probe_401"] == 14 assert payload["rotate_current_summary"]["deleted_401"] == 1 assert payload["rotate_log_tail"]["available"] is True assert len(payload["rotate_log_tail"]["recent_events"]) == 2 assert payload["runtime_state_file"]["exists"] is True assert payload["runtime_state_file"]["path"] == str(runtime_state_file) def test_summary_payload_exposes_account_survival_payload(monkeypatch, tmp_path: Path) -> None: state_file = tmp_path / "account_survival.json" state_file.write_text( ( "{\n" ' "updated_at": "2026-03-26T13:10:00+08:00",\n' ' "summary": {"tracked": 4, "alive": 3, "invalid": 1}\n' "}\n" ), encoding="utf-8", ) app = create_app(enable_background_tasks=False) app.state.settings = AppSettings( pool_dir=tmp_path, account_survival_enabled=True, account_survival_state_file=state_file, responses_survival_state_file=tmp_path / "responses_survival_missing.json", cleanup_enabled=False, validate_enabled=False, ) app.state.background_tasks = [] payload = _summary_payload(app) assert payload["account_survival"]["available"] is True assert payload["account_survival"]["summary"]["tracked"] == 4 assert payload["routes"]["account_survival"] == "/api/account-survival" def test_summary_payload_prefers_responses_survival_payload_when_available(tmp_path: Path) -> None: account_state_file = tmp_path / "account_survival.json" account_state_file.write_text( ( "{\n" ' "probe_mode": "usage",\n' ' "summary": {"tracked": 4, "alive": 4, "invalid": 0}\n' "}\n" ), encoding="utf-8", ) responses_state_file = tmp_path / "responses_survival.json" responses_state_file.write_text( ( "{\n" ' "probe_mode": "responses",\n' ' "updated_at": "2026-03-30T20:30:00+08:00",\n' ' "summary": {"tracked": 8, "alive": 7, "invalid": 1, "first_invalid_count": 1}\n' "}\n" ), encoding="utf-8", ) app = create_app(enable_background_tasks=False) app.state.settings = AppSettings( pool_dir=tmp_path, account_survival_enabled=True, account_survival_state_file=account_state_file, responses_survival_state_file=responses_state_file, cleanup_enabled=False, validate_enabled=False, ) app.state.background_tasks = [] payload = _summary_payload(app) assert payload["account_survival"]["available"] is True assert payload["account_survival"]["probe_mode"] == "responses" assert payload["account_survival"]["summary"]["tracked"] == 8 def test_account_survival_payload_derives_promotion_stats_from_member_files(tmp_path: Path) -> None: warmup_file = tmp_path / "warmup@example.com.json" warmup_file.write_text( json.dumps( { "email": "warmup@example.com", "access_token": "tok", "account_id": "acct", "created_at": "2026-03-31T17:00:00+08:00", "warmup_required": True, "cpa_sync_status": "synced", }, ensure_ascii=False, ), encoding="utf-8", ) responses_state_file = tmp_path / "responses_survival.json" responses_state_file.write_text( json.dumps( { "probe_mode": "responses", "updated_at": "2026-03-31T17:30:00+08:00", "summary": {"tracked": 1, "alive": 1, "invalid": 0}, "members": [ { "email": "warmup@example.com", "path": str(warmup_file), "warmup_state": "passed", } ], "promotion_stats": { "promoted_success_total": 0, "promoted_failure_total": 0, }, }, ensure_ascii=False, ), encoding="utf-8", ) settings = AppSettings( pool_dir=tmp_path, responses_survival_state_file=responses_state_file, account_survival_enabled=True, cleanup_enabled=False, validate_enabled=False, ) payload = _account_survival_payload(settings) assert payload["promotion_stats"]["promoted_success_total"] == 1 assert payload["promotion_stats"]["promoted_failure_total"] == 0 def test_summary_payload_formats_survival_durations_as_hms(tmp_path: Path) -> None: responses_state_file = tmp_path / "responses_survival.json" responses_state_file.write_text( ( "{\n" ' "probe_mode": "responses",\n' ' "updated_at": "2026-03-31T13:30:00+08:00",\n' ' "summary": {"tracked": 2, "alive": 1, "invalid": 1, "first_invalid_count": 1},\n' ' "changes": [{"email": "a@example.com", "survival_seconds": 3661}],\n' ' "members": [\n' ' {"email": "a@example.com", "survival_seconds": 3661},\n' ' {"email": "b@example.com", "survival_seconds": 59}\n' " ]\n" "}\n" ), encoding="utf-8", ) app = create_app(enable_background_tasks=False) app.state.settings = AppSettings( pool_dir=tmp_path, account_survival_enabled=True, account_survival_state_file=tmp_path / "account_survival_missing.json", responses_survival_state_file=responses_state_file, cleanup_enabled=False, validate_enabled=False, ) app.state.background_tasks = [] payload = _summary_payload(app) assert payload["account_survival"]["changes"][0]["survival_text"] == "1h 1m 1s" assert payload["account_survival"]["members"][0]["survival_text"] == "1h 1m 1s" assert payload["account_survival"]["members"][1]["survival_text"] == "59s" def test_account_survival_payload_exposes_fresh_unauthorized_experiment(tmp_path: Path) -> None: responses_state_file = tmp_path / "responses_survival.json" responses_state_file.write_text( json.dumps( { "probe_mode": "responses", "updated_at": "2026-04-01T09:59:00+08:00", "summary": {"tracked": 1, "alive": 1, "invalid": 0}, "members": [], }, ensure_ascii=False, ), encoding="utf-8", ) experiment_state = tmp_path / "track_new8_unauthorized_20260401.json" experiment_state.write_text( json.dumps( { "started_at": "2026-04-01T09:57:26+08:00", "cohort_size": 8, "members": [ { "email": "fresh@example.com", "created_at": "2026-04-01T09:58:00+08:00", "registration_post_create_gate": "add_phone", "registration_proxy_key": "台湾-三网备用", "probe_count": 3, "last_status_code": 401, "last_category": "invalid", "last_detail": '{"detail":"Unauthorized"}', "first_401_at": "2026-04-01T10:01:05+08:00", "first_401_seconds": 185, "first_401_detail": '{"detail":"Unauthorized"}', } ], "history": [], }, ensure_ascii=False, ), encoding="utf-8", ) settings = AppSettings( pool_dir=tmp_path, state_dir=tmp_path, responses_survival_state_file=responses_state_file, account_survival_enabled=True, cleanup_enabled=False, validate_enabled=False, ) payload = _account_survival_payload(settings) experiment = payload["fresh_unauthorized_experiment"] assert experiment["available"] is True assert experiment["summary"]["tracked"] == 1 assert experiment["summary"]["first_401_count"] == 1 assert experiment["members"][0]["first_401_text"] == "3m 5s" assert experiment["members"][0]["first_401_detail"] == '{"detail":"Unauthorized"}' def test_account_survival_reseed_api_rebuilds_latest_ten_cohort(monkeypatch, tmp_path: Path) -> None: for idx in range(12): (tmp_path / f"user{idx:02d}@example.com.json").write_text( ( "{\n" f' "email": "user{idx:02d}@example.com",\n' ' "access_token": "tok",\n' ' "account_id": "acct",\n' f' "created_at": "2026-03-26T12:{idx:02d}:00+08:00"\n' "}\n" ), encoding="utf-8", ) state_file = tmp_path / "responses_survival.json" state_file.write_text( ( "{\n" ' "probe_mode": "responses",\n' ' "members": []\n' "}\n" ), encoding="utf-8", ) app = create_app(enable_background_tasks=False) app.state.settings = AppSettings( pool_dir=tmp_path, account_survival_enabled=True, account_survival_cohort_size=10, account_survival_state_file=tmp_path / "account_survival_unused.json", responses_survival_state_file=state_file, responses_survival_require_provenance=False, responses_survival_recent_window_seconds=0, cleanup_enabled=False, validate_enabled=False, ) app.state.background_tasks = [] monkeypatch.setattr( "ops.responses_survival.probe_responses_token_file", lambda path, proxy, timeout: ScanResult(file=path.name, category="normal", status_code=200, detail="ok"), ) status, _headers, body = _request_via_asgi(app, "POST", "/api/account-survival/reseed") payload = json.loads(body.decode("utf-8")) assert status == 200 assert payload["available"] is True assert payload["summary"]["tracked"] == 10 assert payload["seed_source"] == "latest_generated_pool_files" assert payload["members"][0]["email"] == "user11@example.com" def test_summary_payload_exposes_register_burst_plan(tmp_path: Path) -> None: app = create_app(enable_background_tasks=False) app.state.settings = AppSettings( pool_dir=tmp_path, register_batch_threads=1, register_batch_target_count=20, register_batch_interval_seconds=10800, cleanup_enabled=False, validate_enabled=False, ) app.state.background_tasks = [] payload = _summary_payload(app) burst_plan = payload["register_burst_plan"] assert burst_plan["mode"] == "burst" assert burst_plan["threads"] == 1 assert burst_plan["target_count"] == 20 assert burst_plan["interval_seconds"] == 10800 assert burst_plan["accounts_per_day"] == 160 assert burst_plan["accounts_needed_for_one_day_target"] == 20 assert burst_plan["accounts_needed_for_sustained_daily_target"] == 140 def test_rotate_log_tail_builds_current_summary_for_in_progress_rotate(monkeypatch, tmp_path: Path) -> None: log_path = tmp_path / "dashboard.log" log_path.write_text( "\n".join( [ "[10:00:00] [rotate] summary | 主池: 800 → 789 | 401删除: 10 | quota探测: 20 | probe401: 10 | probe429: 1 | probe跳过: 2 | 429删除: 1", "[10:05:00] [rotate] 🔎 a@example.com.json quota probe → 401 invalidated", "[10:05:01] [rotate] ❌ a@example.com.json 401删除", "[10:05:02] [rotate] 🔎 b@example.com.json quota probe → 429", "[10:05:03] [rotate] ❌ b@example.com.json 429删除", "", ] ) + "\n", encoding="utf-8", ) monkeypatch.setattr("main.DEFAULT_DASHBOARD_LOG_FILE", log_path) payload = _rotate_log_tail() assert payload["latest_summary"]["deleted_401"] == 10 assert payload["current_summary"]["quota_probed"] == 2 assert payload["current_summary"]["quota_probe_401"] == 1 assert payload["current_summary"]["quota_probe_429"] == 1 assert payload["current_summary"]["deleted_401"] == 1 assert payload["current_summary"]["deleted_429"] == 1 def test_rotate_log_tail_prefers_live_stdout_log_over_stale_dashboard_log(monkeypatch, tmp_path: Path) -> None: stale_log = tmp_path / "dashboard.log" stale_log.write_text( "[10:00:00] [rotate] summary | 主池: 800 → 790 | 401删除: 10 | quota探测: 20 | probe401: 10 | probe429: 0 | probe跳过: 0\n", encoding="utf-8", ) live_log = tmp_path / "main_full_8threads.log" live_log.write_text( "[10:05:00] [rotate] summary | 主池: 790 → 788 | 401删除: 2 | quota探测: 5 | probe401: 2 | probe429: 0 | probe跳过: 0\n", encoding="utf-8", ) monkeypatch.setattr("main.DEFAULT_DASHBOARD_LOG_FILE", stale_log) original_readlink = os.readlink def fake_readlink(path: str) -> str: if path == "/proc/self/fd/1": return str(live_log) return original_readlink(path) monkeypatch.setattr(os, "readlink", fake_readlink) payload = _rotate_log_tail() assert payload["path"] == str(live_log) assert payload["latest_summary"]["deleted_401"] == 2 def test_recent_pool_files_returns_latest_entries_without_glob_expansion_issue(tmp_path: Path) -> None: older = tmp_path / "older@example.com.json" newer = tmp_path / "newer@example.com.json" older.write_text('{"email":"older@example.com"}', encoding="utf-8") newer.write_text('{"email":"newer@example.com"}', encoding="utf-8") older.touch() newer.touch() items = _recent_pool_files(tmp_path, limit=2) assert len(items) == 2 assert {item["name"] for item in items} == {"older@example.com.json", "newer@example.com.json"} assert all(item["size_bytes"] > 0 for item in items) def test_summary_payload_exposes_dashboard_overview_fields(monkeypatch, tmp_path: Path) -> None: app = create_app(enable_background_tasks=False) app.state.settings = AppSettings( pool_dir=tmp_path, cleanup_enabled=False, validate_enabled=False, ) app.state.background_tasks = [] class FakeRegistrationLoop: def snapshot(self) -> dict[str, object]: return { "name": "register", "status": "running", "threads_alive": 2, "threads_total": 4, "total_attempts": 10, "total_success": 8, "total_success_registered": 8, "total_cpa_sync_success": 5, "total_cpa_sync_failure": 2, "total_failure": 2, "success_rate": 80.0, "registered_success_rate": 80.0, "cpa_sync_success_rate": 50.0, "target_count": None, "target_reached": False, "last_error": None, "proxy": None, "proxy_pool_enabled": False, "mail_provider": "mailtm", "interval_seconds": 5, "run_count": 10, "success_count": 8, "failure_count": 2, "is_running": True, "last_started_at": None, "last_finished_at": None, "last_duration_seconds": None, "next_run_at": None, "failure_by_stage": {}, "failure_signals": {}, "recent_failure_hotspots": [], "recent_attempts": [], "cfmail_domain_pool": { "target_count": 3, "active_count": 2, "active_domains": [ { "name": "cfmail-tw", "domain": "tw.example.test", "inflight": 1, "recent_attempts": 6, "recent_success": 5, "recent_failure": 1, } ], "replenishing": False, "replenish_reason": "", }, "cfmail_add_phone_stoploss": { "active_domain": "demo.example.test", "in_cooldown": True, "cooldown_remaining_seconds": 120, "last_triggered_at": "2026-03-23T00:00:00", "last_reason": "add_phone threshold reached", "last_add_phone_failures": 8, "last_successes": 0, "last_window_size": 12, "window_size": 12, "threshold": 8, "max_successes_in_window": 2, }, } app.state.registration_loop = FakeRegistrationLoop() monkeypatch.setattr( "main._fetch_management_auth_files", lambda settings: ( True, [ {"name": "a@example.com.json", "unavailable": False}, {"name": "b@example.com.json", "unavailable": True, "status_message": "usage_limit_reached"}, {"name": "c@example.com.json", "status_message": "token invalidated by upstream"}, ], ), ) monkeypatch.setattr("main._count_today_new", lambda pool_dir: 3) payload = _summary_payload(app) assert payload["cpa_count"] == 3 assert payload["regular_accounts"]["source_available"] is True assert payload["regular_accounts"]["available"] == 1 assert payload["regular_accounts"]["waiting_reset"] == 1 assert payload["regular_accounts"]["invalid"] == 1 assert payload["tokens"]["estimation_mode"] == "count_based" assert payload["tokens"]["baseline_source"] == "configured" assert payload["tokens"]["available_now"] == 5000000 assert payload["tokens"]["available_with_reset"] == 10000000 assert payload["today_new"] == 3 assert payload["success_rate"] == 80.0 assert payload["registered_success_total"] == 8 assert payload["cpa_sync_success_total"] == 5 assert payload["cpa_sync_failure_total"] == 2 assert payload["registered_success_rate"] == 80.0 assert payload["cpa_sync_success_rate"] == 50.0 assert payload["observed_loss"] == 2 assert payload["register_failure_by_stage"] == {} assert payload["register_failure_signals"] == {} assert payload["register_recent_failure_hotspots"] == [] assert payload["register_cfmail_domain_pool"]["active_count"] == 2 assert payload["register_cfmail_domain_pool"]["active_domains"][0]["domain"] == "tw.example.test" assert payload["register_cfmail_add_phone_stoploss"]["in_cooldown"] is True def test_summary_payload_falls_back_when_management_inventory_unavailable(monkeypatch, tmp_path: Path) -> None: app = create_app(enable_background_tasks=False) app.state.settings = AppSettings( pool_dir=tmp_path, runtime_state_file=tmp_path / "runtime_state.json", cleanup_enabled=False, validate_enabled=False, ) app.state.background_tasks = [] app.state.registration_loop = None monkeypatch.setattr("main._fetch_management_auth_files", lambda settings: (False, [])) monkeypatch.setattr("main._count_cpa_files", lambda settings: 9) payload = _summary_payload(app) assert payload["cpa_count"] == 9 assert payload["cpa_inventory"]["management_available"] is False assert payload["regular_accounts"]["source_available"] is False assert payload["regular_accounts"]["source_error"] == "management_data_unavailable" assert payload["tokens"]["estimation_mode"] == "count_based" assert payload["tokens"]["fallback_reason"] == "missing_management_inventory" assert payload["success_rate"] is None def test_runtime_payload_exposes_proxy_pool_snapshot(tmp_path: Path) -> None: app = create_app(enable_background_tasks=False) app.state.settings = AppSettings( pool_dir=tmp_path, cleanup_enabled=False, validate_enabled=False, ) app.state.background_tasks = [] class FakePool: def snapshot(self) -> list[dict[str, object]]: return [ { "name": "sg-node-1", "region": "sg", "proxy_url": "socks5://127.0.0.1:17891", "local_port": 17891, "in_use": True, "disabled": False, "successes": 7, "failures": 1, "last_error": "", } ] class FakeRegistrationLoop: def __init__(self) -> None: self._proxy_pool = FakePool() def snapshot(self) -> dict[str, object]: return { "name": "register", "status": "running", "threads_alive": 1, "threads_total": 1, "total_attempts": 8, "total_success": 7, "total_failure": 1, "success_rate": 87.5, "target_count": None, "target_reached": False, "last_error": None, "proxy": None, "proxy_pool_enabled": True, "mail_provider": "mailtm", "interval_seconds": 5, "run_count": 8, "success_count": 7, "failure_count": 1, "is_running": True, "last_started_at": None, "last_finished_at": None, "last_duration_seconds": None, "next_run_at": None, } app.state.registration_loop = FakeRegistrationLoop() payload = _runtime_payload(app) proxy_pool = payload["proxy_pool"] assert proxy_pool["enabled"] is True assert proxy_pool["node_count"] == 1 assert proxy_pool["in_use_count"] == 1 assert proxy_pool["disabled_count"] == 0 assert proxy_pool["nodes"][0]["name"] == "sg-node-1" def test_runtime_payload_uses_external_runtime_state_when_loop_runs_out_of_process(tmp_path: Path) -> None: runtime_state_file = tmp_path / "runtime_state.json" runtime_state_file.write_text( """ { "updated_at": "2026-03-22T22:40:00", "register_snapshot": { "name": "register", "status": "running", "threads_alive": 3, "threads_total": 3, "total_attempts": 12, "total_success": 9, "total_failure": 3, "success_rate": 75.0, "target_count": null, "target_reached": false, "last_error": "token acquisition failed", "proxy": "http://127.0.0.1:7899", "proxy_pool_enabled": true, "mail_provider": "cfmail", "interval_seconds": 5, "run_count": 12, "success_count": 9, "failure_count": 3, "is_running": true, "last_started_at": "2026-03-22T22:39:00", "last_finished_at": null, "last_duration_seconds": null, "next_run_at": null, "failure_by_stage": {"token_acquisition": 3}, "failure_signals": {"add_phone_gate": 2}, "recent_failure_hotspots": [{"key": "add_phone_gate", "stage": "add_phone_gate", "count": 2}], "recent_attempts": [{"timestamp": "2026-03-22T22:39:30", "success": false, "stage": "add_phone_gate", "signal": "add_phone_gate"}], "cfmail_add_phone_stoploss": {"active_domain": "demo.example.test", "in_cooldown": true} }, "proxy_pool": { "configured": true, "enabled": true, "snapshot_error": null, "node_count": 2, "in_use_count": 1, "disabled_count": 0, "nodes": [ {"name": "sg-1", "in_use": true, "disabled": false}, {"name": "tw-1", "in_use": false, "disabled": false} ] } } """.strip(), encoding="utf-8", ) app = create_app(enable_background_tasks=False) app.state.settings = AppSettings( pool_dir=tmp_path, runtime_state_file=runtime_state_file, cleanup_enabled=False, validate_enabled=False, ) app.state.background_tasks = [] app.state.registration_loop = None payload = _runtime_payload(app) assert payload["architecture"] == "split-runtime-fastapi+loop" register_state = next(task for task in payload["task_states"] if task["name"] == "register") assert register_state["threads_alive"] == 3 assert register_state["failure_by_stage"]["token_acquisition"] == 3 assert register_state["cfmail_add_phone_stoploss"]["in_cooldown"] is True assert payload["proxy_pool"]["enabled"] is True assert payload["proxy_pool"]["node_count"] == 2 def test_runtime_payload_marks_proxy_pool_configured_for_direct_urls(tmp_path: Path) -> None: app = create_app(enable_background_tasks=False) app.state.settings = AppSettings( pool_dir=tmp_path, runtime_state_file=tmp_path / "runtime_state.json", proxy_pool_direct_urls="http://5.6.7.8:8080", cleanup_enabled=False, validate_enabled=False, ) app.state.background_tasks = [] app.state.registration_loop = None payload = _runtime_payload(app) assert payload["proxy_pool"]["configured"] is True assert payload["proxy_pool"]["enabled"] is False def test_dashboard_cors_preflight_allows_configured_origin(monkeypatch) -> None: monkeypatch.setenv("ZHUCE6_DASHBOARD_ALLOWED_ORIGINS", "http://127.0.0.1:8317") app = create_app(enable_background_tasks=False) app.state.settings = AppSettings(cleanup_enabled=False, validate_enabled=False) app.state.background_tasks = [] app.state.registration_loop = None status_code, headers, _ = _request_via_asgi( app, "OPTIONS", "/api/summary", { "Origin": "http://127.0.0.1:8317", "Access-Control-Request-Method": "GET", }, ) assert status_code == 204 assert headers["access-control-allow-origin"] == "http://127.0.0.1:8317" def test_settings_api_returns_current_runtime_mode(tmp_path: Path) -> None: app = create_app(enable_background_tasks=False, mode="lite") app.state.settings = AppSettings( runtime_mode="lite", register_enabled=True, register_threads=2, register_batch_target_count=30, register_batch_interval_seconds=3600, register_mail_provider="cfmail", register_proxy="http://127.0.0.1:7899", enable_proxy_pool=True, proxy_pool_size=10, proxy_pool_direct_urls="http://1.2.3.4:8080", proxy_pool_regions=("jp", "tw"), rotate_enabled=False, rotate_interval=120, pool_dir=tmp_path, ) app.state.background_tasks = [] app.state.registration_loop = None status, _headers, body = _request_via_asgi(app, "GET", "/api/settings") payload = json.loads(body.decode("utf-8")) assert status == 200 assert payload["mode"] == "lite" assert payload["register"]["threads"] == 2 assert payload["proxy_pool"]["size"] == 10 assert payload["cpa"]["rotate_interval"] == 120 def test_settings_api_persists_whitelisted_updates(monkeypatch, tmp_path: Path) -> None: env_file = tmp_path / ".env" env_file.write_text("", encoding="utf-8") monkeypatch.setenv("ZHUCE6_ENV_FILE", str(env_file)) app = create_app(enable_background_tasks=False, mode="full") app.state.settings = AppSettings( runtime_mode="full", register_threads=1, register_batch_target_count=20, register_batch_interval_seconds=10800, register_mail_provider="cfmail", register_proxy="http://127.0.0.1:7899", enable_proxy_pool=True, proxy_pool_size=20, proxy_pool_direct_urls="", proxy_pool_regions=("jp", "tw", "hk", "sg"), rotate_interval=120, pool_dir=tmp_path, ) app.state.background_tasks = [] app.state.registration_loop = None status, _headers, body = _request_via_asgi( app, "PUT", "/api/settings", headers={"content-type": "application/json"}, body=json.dumps( { "register.threads": 3, "register.batch_target_count": 25, "proxy_pool.size": 12, "cpa.rotate_interval": 300, } ).encode("utf-8"), ) payload = json.loads(body.decode("utf-8")) persisted = env_file.read_text(encoding="utf-8") assert status == 200 assert payload["register"]["threads"] == 3 assert payload["proxy_pool"]["size"] == 12 assert payload["cpa"]["rotate_interval"] == 300 assert payload["restart_required"] is True assert "ZHUCE6_REGISTER_THREADS=3" in persisted assert "ZHUCE6_REGISTER_BATCH_TARGET_COUNT=25" in persisted assert "ZHUCE6_PROXY_POOL_SIZE=12" in persisted assert "ZHUCE6_ROTATE_INTERVAL=300" in persisted def test_register_control_api_starts_and_stops_loop(monkeypatch, tmp_path: Path) -> None: events: list[str] = [] class FakeLoop: def __init__(self, settings): # type: ignore[no-untyped-def] self.settings = settings def start(self) -> None: events.append("start") def stop(self) -> None: events.append("stop") monkeypatch.setattr("main.RegistrationLoop", FakeLoop) app = create_app(enable_background_tasks=False, mode="dashboard") app.state.settings = AppSettings(runtime_mode="dashboard", pool_dir=tmp_path) app.state.background_tasks = [] app.state.registration_loop = None status, _headers, body = _request_via_asgi( app, "POST", "/api/control/register", headers={"content-type": "application/json"}, body=b'{"action":"start"}', ) payload = json.loads(body.decode("utf-8")) assert status == 200 assert payload["status"] == "started" assert events == ["start"] status, _headers, body = _request_via_asgi( app, "POST", "/api/control/register", headers={"content-type": "application/json"}, body=b'{"action":"stop"}', ) payload = json.loads(body.decode("utf-8")) assert status == 200 assert payload["status"] == "stopped" assert events == ["start", "stop"] def test_health_dependencies_api_skips_cpa_checks_in_lite_mode(monkeypatch, tmp_path: Path) -> None: def fail_fetch(*_args, **_kwargs): # type: ignore[no-untyped-def] raise AssertionError("lite mode should not query CPA") monkeypatch.setattr("main._fetch_management_auth_files", fail_fetch) app = create_app(enable_background_tasks=False, mode="lite") app.state.settings = AppSettings( runtime_mode="lite", register_mail_provider="cfmail", enable_proxy_pool=False, pool_dir=tmp_path, ) app.state.background_tasks = [] app.state.registration_loop = None status, _headers, body = _request_via_asgi(app, "GET", "/api/health/dependencies") payload = json.loads(body.decode("utf-8")) assert status == 200 assert payload["cpa"]["status"] == "unconfigured" assert "docker" not in payload def test_lite_mode_summary_skips_management_inventory(monkeypatch, tmp_path: Path) -> None: def fail_fetch(*_args, **_kwargs): # type: ignore[no-untyped-def] raise AssertionError("lite mode should not fetch CPA inventory") monkeypatch.setattr("main._fetch_management_auth_files", fail_fetch) app = create_app(enable_background_tasks=False, mode="lite") app.state.settings = AppSettings(runtime_mode="lite", pool_dir=tmp_path) app.state.background_tasks = [] app.state.registration_loop = None payload = _summary_payload(app) assert payload["runtime"]["runtime_mode"] == "lite" assert payload["cpa_count"] is None assert payload["regular_accounts"] is None assert payload["tokens"] is None def test_dashboard_html_contains_settings_tab_and_control_api_hooks() -> None: html = Path("/home/sophomores/zhuce6/dashboard/zhuce6.html").read_text(encoding="utf-8") assert "Settings" in html assert "/api/settings" in html assert "/api/control/register" in html assert "/api/health/dependencies" in html assert "http://localhost:8317/management.html" not in html assert "settings.cpa.management_url" in html def test_create_app_lite_mode_registers_only_register_task(monkeypatch) -> None: class FakeLoop: def __init__(self, settings): # type: ignore[no-untyped-def] self.settings = settings def start(self) -> None: return None def stop(self) -> None: return None def snapshot(self) -> dict[str, object]: return { "name": "register", "status": "running", "run_count": 0, "success_count": 0, "failure_count": 0, "threads_alive": 1, "threads_total": 1, } monkeypatch.setattr("main.RegistrationLoop", FakeLoop) async def run_lifespan() -> None: app = create_app(enable_background_tasks=True, mode="lite") async with app.router.lifespan_context(app): payload = _runtime_payload(app) assert payload["runtime_mode"] == "lite" assert payload["registered_tasks"] == ["register"] asyncio.run(run_lifespan()) def test_dashboard_cors_get_adds_origin_header_for_allowed_origin(monkeypatch, tmp_path: Path) -> None: monkeypatch.setenv("ZHUCE6_DASHBOARD_ALLOWED_ORIGINS", "http://localhost:8317") app = create_app(enable_background_tasks=False) app.state.settings = AppSettings(pool_dir=tmp_path, cleanup_enabled=False, validate_enabled=False) app.state.background_tasks = [] app.state.registration_loop = None status_code, headers, _ = _request_via_asgi( app, "GET", "/api/summary", {"Origin": "http://localhost:8317"}, ) assert status_code == 200 assert headers["access-control-allow-origin"] == "http://localhost:8317" def test_dashboard_cors_headers_are_not_added_for_other_origins(monkeypatch) -> None: monkeypatch.setenv("ZHUCE6_DASHBOARD_ALLOWED_ORIGINS", "http://localhost:8317") app = create_app(enable_background_tasks=False) app.state.settings = AppSettings(cleanup_enabled=False, validate_enabled=False) app.state.background_tasks = [] app.state.registration_loop = None status_code, headers, _ = _request_via_asgi( app, "GET", "/api/runtime", {"Origin": "http://127.0.0.1:9999"}, ) assert status_code == 200 assert "access-control-allow-origin" not in headers def test_build_background_tasks_registers_validate_when_enabled() -> None: tasks = _build_background_tasks( AppSettings( cleanup_enabled=False, d1_cleanup_enabled=False, validate_enabled=True, validate_interval=90, validate_scope="used", rotate_enabled=False, account_survival_enabled=False, ) ) assert [task.name for task in tasks] == ["validate"] assert tasks[0].interval_seconds == 90 def test_build_background_tasks_registers_d1_cleanup_when_enabled() -> None: tasks = _build_background_tasks( AppSettings( cleanup_enabled=False, validate_enabled=False, rotate_enabled=False, d1_cleanup_enabled=True, d1_cleanup_interval=1800, account_survival_enabled=False, ) ) assert [task.name for task in tasks] == ["d1_cleanup"] assert tasks[0].interval_seconds == 1800 def test_build_background_tasks_uses_responses_survival_when_enabled(monkeypatch, tmp_path: Path) -> None: tasks = _build_background_tasks( AppSettings( cleanup_enabled=False, validate_enabled=False, rotate_enabled=False, d1_cleanup_enabled=False, account_survival_enabled=True, account_survival_interval=123, pool_dir=tmp_path / "pool", responses_survival_state_file=tmp_path / "responses_survival.json", responses_survival_recent_window_seconds=900, responses_survival_require_provenance=True, warmup_min_age_seconds=600, warmup_min_successful_probes=2, ) ) assert [task.name for task in tasks] == ["account_survival"] assert tasks[0].interval_seconds == 123 assert "responses_survival_once" in tasks[0].fn.__code__.co_names assert "print_responses_survival_summary" in tasks[0].fn.__code__.co_names def test_apply_runtime_mode_forces_account_survival_for_full_cpa_register() -> None: settings = AppSettings( runtime_mode="full", backend="cpa", register_enabled=True, account_survival_enabled=False, ) updated = _apply_runtime_mode(settings, "full") assert updated.account_survival_enabled is True def test_runtime_payload_counts_warmup_promotions_as_success(tmp_path: Path) -> None: responses_state_file = tmp_path / "responses_survival.json" responses_state_file.write_text("{}", encoding="utf-8") for idx in range(3): (tmp_path / f"warmup-{idx}.json").write_text( json.dumps( { "email": f"warmup-{idx}@example.com", "access_token": "tok", "account_id": f"acct-{idx}", "created_at": "2026-03-31T17:30:10+08:00", "warmup_required": True, "cpa_sync_status": "synced", }, ensure_ascii=False, ), encoding="utf-8", ) class _Task: def __init__(self, snapshot: dict[str, object]) -> None: self._snapshot = snapshot def snapshot(self) -> dict[str, object]: return dict(self._snapshot) register_snapshot = { "name": "register", "status": "running", "threads_alive": 4, "threads_total": 4, "total_attempts": 10, "total_success": 1, "total_success_registered": 1, "total_success_direct": 1, "total_warmup_pending": 2, "total_cpa_sync_success": 1, "total_cpa_sync_failure": 0, "total_failure": 4, "success_rate": 10.0, "registered_success_rate": 10.0, "cpa_sync_success_rate": 10.0, "last_started_at": "2026-03-31T17:30:00+08:00", } app = create_app(enable_background_tasks=False) app.state.settings = AppSettings( pool_dir=tmp_path, responses_survival_state_file=responses_state_file, cleanup_enabled=False, validate_enabled=False, rotate_enabled=False, ) app.state.background_tasks = [_Task(register_snapshot)] app.state.registration_loop = None payload = _runtime_payload(app) register_task = next(task for task in payload["task_states"] if task.get("name") == "register") assert register_task["total_success"] == 4 assert register_task["total_success_registered"] == 4 assert register_task["total_success_promoted"] == 3 assert register_task["total_cpa_sync_success"] == 4 assert register_task["success_rate"] == 40.0 assert register_task["registered_success_rate"] == 40.0 assert register_task["cpa_sync_success_rate"] == 40.0 def test_runtime_payload_exposes_worker_retry_split_and_current_warmup_backlog(tmp_path: Path) -> None: responses_state_file = tmp_path / "responses_survival.json" responses_state_file.write_text("{}", encoding="utf-8") runtime_start = "2026-04-01T10:00:00+08:00" (tmp_path / "pending.json").write_text( json.dumps( { "email": "pending@example.com", "access_token": "tok", "account_id": "acct-pending", "created_at": "2026-04-01T10:00:30+08:00", "warmup_required": True, "cpa_sync_status": "warmup_pending", }, ensure_ascii=False, ), encoding="utf-8", ) (tmp_path / "synced.json").write_text( json.dumps( { "email": "synced@example.com", "access_token": "tok", "account_id": "acct-synced", "created_at": "2026-04-01T10:00:40+08:00", "warmup_required": True, "cpa_sync_status": "synced", }, ensure_ascii=False, ), encoding="utf-8", ) class _Task: def __init__(self, snapshot: dict[str, object]) -> None: self._snapshot = snapshot def snapshot(self) -> dict[str, object]: return dict(self._snapshot) register_snapshot = { "name": "register", "status": "running", "threads_alive": 9, "threads_total": 9, "last_started_at": runtime_start, "pending_token_queue": {"queue_size": 0}, "total_attempts": 2, "total_success": 0, "total_success_direct": 0, "total_failure": 0, "total_warmup_pending": 2, } app = create_app(enable_background_tasks=False) app.state.settings = AppSettings( pool_dir=tmp_path, responses_survival_state_file=responses_state_file, cleanup_enabled=False, validate_enabled=False, rotate_enabled=False, ) app.state.background_tasks = [_Task(register_snapshot)] app.state.registration_loop = None payload = _runtime_payload(app) register_task = next(task for task in payload["task_states"] if task.get("name") == "register") assert register_task["register_worker_threads"] == 8 assert register_task["retry_sidecar_threads"] == 1 assert register_task["current_warmup_backlog"] == 1 def test_summary_payload_uses_effective_success_totals_from_warmup_promotions(tmp_path: Path) -> None: responses_state_file = tmp_path / "responses_survival.json" responses_state_file.write_text("{}", encoding="utf-8") for idx in range(3): (tmp_path / f"warmup-{idx}.json").write_text( json.dumps( { "email": f"warmup-{idx}@example.com", "access_token": "tok", "account_id": f"acct-{idx}", "created_at": "2026-03-31T17:30:10+08:00", "warmup_required": True, "cpa_sync_status": "synced", }, ensure_ascii=False, ), encoding="utf-8", ) class _Task: def __init__(self, snapshot: dict[str, object]) -> None: self._snapshot = snapshot def snapshot(self) -> dict[str, object]: return dict(self._snapshot) app = create_app(enable_background_tasks=False) app.state.settings = AppSettings( pool_dir=tmp_path, responses_survival_state_file=responses_state_file, cleanup_enabled=False, validate_enabled=False, rotate_enabled=False, ) app.state.background_tasks = [ _Task( { "name": "register", "status": "running", "threads_alive": 4, "threads_total": 4, "total_attempts": 10, "total_success": 1, "total_success_registered": 1, "total_success_direct": 1, "total_cpa_sync_success": 1, "total_cpa_sync_failure": 0, "total_failure": 4, "success_rate": 10.0, "registered_success_rate": 10.0, "cpa_sync_success_rate": 10.0, "last_started_at": "2026-03-31T17:30:00+08:00", } ) ] app.state.registration_loop = None payload = _summary_payload(app) assert payload["registered_success_total"] == 4 assert payload["cpa_sync_success_total"] == 4 assert payload["registered_success_rate"] == 40.0 assert payload["cpa_sync_success_rate"] == 40.0