rakuten-api/tests/test_gateway_leasing.py

"""长轮询与并发抢任务测试

覆盖规格 §9 验收清单里需要时序观察的条目：
- 两个 worker 同时 lease，只有一个拿到任务（全局并发度 1）
- 有任务在 leased/running 时，lease 一律返回空
- 无任务时 lease 挂起到 wait 秒才返回，且返回 200 + data: null
- submit 期间正在 wait 的 lease 被唤醒
- worker 心跳：lease 请求刷新 last_seen_at，/health 据此判断失联
"""
from __future__ import annotations

import asyncio
import time
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from app.shared.config import get_settings

TOKEN = get_settings().bearer_token
AUTH = {"Authorization": f"Bearer {TOKEN}"}


@pytest.fixture
def gateway_client(tmp_path: Path, monkeypatch):
    db_path = tmp_path / "gw.db"
    monkeypatch.setenv("RAKUTEN_GATEWAY_DB_PATH", str(db_path))
    monkeypatch.setenv("RAKUTEN_ACCOUNT_DISCOVERY_ENABLED", "false")
    # worker_offline_alert_seconds 设小一点，方便测 /health 失联判定
    monkeypatch.setenv("RAKUTEN_WORKER_OFFLINE_ALERT_SECONDS", "1")
    get_settings.cache_clear()
    try:
        from app.gateway.main import create_app

        app = create_app()
        with TestClient(app) as client:
            yield client
        # 让后台 sweep 任务退出后再清理
    finally:
        get_settings.cache_clear()


# ---- 全局并发度 1 ----


def test_concurrent_leases_only_one_gets_task(gateway_client):
    """两个 worker 同时 lease，只有一个拿到任务"""
    gateway_client.post(
        "/api/orders",
        json={"task_id": "t1", "site": "rakuten", "intent": {}},
        headers=AUTH,
    )

    # 同步发两个 lease，wait=0 立即返回
    r1 = gateway_client.get("/api/orders/lease?worker_id=w1&wait=0", headers=AUTH)
    r2 = gateway_client.get("/api/orders/lease?worker_id=w2&wait=0", headers=AUTH)

    owned = [r for r in (r1, r2) if r.json()["data"] is not None]
    assert len(owned) == 1
    assert owned[0].json()["data"]["task_id"] == "t1"


def test_active_task_blocks_subsequent_leases(gateway_client):
    """leased 或 running 状态下，lease 立即返回空"""
    gateway_client.post(
        "/api/orders",
        json={"task_id": "active", "site": "rakuten", "intent": {}},
        headers=AUTH,
    )
    gateway_client.post(
        "/api/orders",
        json={"task_id": "queued", "site": "rakuten", "intent": {}},
        headers=AUTH,
    )

    # 领走 active
    first = gateway_client.get("/api/orders/lease?worker_id=w1&wait=0", headers=AUTH).json()
    assert first["data"]["task_id"] == "active"

    # 队列里还有 queued，但 active 还在执行 → 立即返回空
    blocked = gateway_client.get("/api/orders/lease?worker_id=w2&wait=0", headers=AUTH).json()
    assert blocked["data"] is None


# ---- 长轮询 ----


def test_lease_hangs_then_returns_null_when_no_task(gateway_client):
    """无任务时 lease 挂起到 wait 秒才返回"""
    started = time.monotonic()
    response = gateway_client.get("/api/orders/lease?worker_id=w1&wait=2", headers=AUTH)
    elapsed = time.monotonic() - started
    assert response.status_code == 200
    assert response.json()["data"] is None
    # 至少挂了 2 秒（允许少量提前）
    assert elapsed >= 1.8


def test_lease_is_woken_up_when_task_submitted(gateway_client):
    """长轮询期间 submit 任务，lease 应被唤醒并立刻拿到"""
    async def submit_after_delay():
        await asyncio.sleep(0.3)
        # TestClient 是同步的，在另一个线程里发请求
        import threading

        def _submit():
            gateway_client.post(
                "/api/orders",
                json={"task_id": "t1", "site": "rakuten", "intent": {}},
                headers=AUTH,
            )

        threading.Thread(target=_submit, daemon=True).start()

    asyncio.run(submit_after_delay())

    started = time.monotonic()
    response = gateway_client.get("/api/orders/lease?worker_id=w1&wait=10", headers=AUTH)
    elapsed = time.monotonic() - started
    assert response.status_code == 200
    data = response.json()["data"]
    assert data is not None and data["task_id"] == "t1"
    # 被唤醒应远早于 10 秒
    assert elapsed < 5


# ---- 心跳与健康检查 ----


def test_lease_updates_worker_last_seen(gateway_client):
    """lease 请求兼作心跳，刷新 workers 表的 last_seen_at"""
    gateway_client.get("/api/orders/lease?worker_id=w1&wait=0", headers=AUTH)
    health = gateway_client.get("/health").json()["data"]
    worker_ids = {w["worker_id"] for w in health["workers"]}
    assert "w1" in worker_ids


def test_health_reports_offline_worker(gateway_client):
    """worker 心跳超阈值时 /health 报异常"""
    gateway_client.get("/api/orders/lease?worker_id=w1&wait=0", headers=AUTH)
    # 阈值被 fixture 设为 1 秒；时间戳为秒精度，需要 sleep 到 2 秒才能稳定越过阈值
    import time as _time

    _time.sleep(2.0)

    health = gateway_client.get("/health").json()["data"]
    assert health["status"] == "degraded"
    offline_ids = {w["worker_id"] for w in health["offline_workers"]}
    assert "w1" in offline_ids


def test_health_reports_stale_queued_task(gateway_client):
    """任务 queued 超过 worker_offline_alert_seconds 时报异常"""
    gateway_client.post(
        "/api/orders",
        json={"task_id": "stuck", "site": "rakuten", "intent": {}},
        headers=AUTH,
    )
    import time as _time

    _time.sleep(2.0)

    health = gateway_client.get("/health").json()["data"]
    assert health["status"] == "degraded"
    stuck_ids = {t["task_id"] for t in health["stale_queued_tasks"]}
    assert "stuck" in stuck_ids


def test_health_lists_active_tasks(gateway_client):
    gateway_client.post(
        "/api/orders",
        json={"task_id": "t1", "site": "rakuten", "intent": {}},
        headers=AUTH,
    )
    gateway_client.get("/api/orders/lease?worker_id=w1&wait=0", headers=AUTH)

    health = gateway_client.get("/health").json()["data"]
    active_ids = {t["task_id"] for t in health["active_tasks"]}
    assert "t1" in active_ids