AiDOPWarehouse/tools/integration-simulator/tests/test_load_metrics.py

"""第三阶段压力指标、P1 批量生成、P2 压力模式测试（任务书 §17/§18/§19）。

全部进程内验证（纯函数/TestClient），不依赖 Ai-DOP 在线与真实 MySQL。
"""
from __future__ import annotations

import threading
import time
from concurrent.futures import ThreadPoolExecutor

import pytest

fastapi_testclient = pytest.importorskip("fastapi.testclient")
TestClient = fastapi_testclient.TestClient

from load import metrics as lm  # noqa: E402
from providers import mock_api  # noqa: E402
from providers import mock_db  # noqa: E402


# ---------------------------------------------------------------------------
# percentile（§17 P50/P95/P99 计算）
# ---------------------------------------------------------------------------

def test_percentile_linear_interpolation():
    assert lm.percentile([1.0, 2.0, 3.0, 4.0], 50) == 2.5
    assert lm.percentile([1.0, 2.0, 3.0, 4.0], 25) == 1.75


def test_percentile_boundaries_and_empty():
    assert lm.percentile([1.0, 2.0, 3.0], 0) == 1.0
    assert lm.percentile([1.0, 2.0, 3.0], 100) == 3.0
    assert lm.percentile([], 50) == 0.0
    assert lm.percentile([5.0], 95) == 5.0


def test_percentile_p95_p99_of_100_values():
    vals = [float(i) for i in range(100)]
    assert lm.percentile(vals, 95) == 94.05
    assert lm.percentile(vals, 99) == 98.01


# ---------------------------------------------------------------------------
# LoadMetrics（§17 指标采集与快照）
# ---------------------------------------------------------------------------

def test_record_classifies_statuses_and_snapshot():
    m = lm.LoadMetrics()
    m.mark_start()
    m.record(200, 0.01, rows=1)
    m.record(201, 0.02)
    m.record(202, 0.03, batches=2)
    m.record(500, 0.04)
    m.record(0, 0.05)
    m.record(429, 0.06)
    m.mark_finish()
    s = m.snapshot()
    assert s["totalRequests"] == 6
    assert s["success"] == 3 and s["failed"] == 3
    assert s["successRate"] == 0.5
    assert s["timeouts"] == 1 and s["rateLimited"] == 1
    assert s["httpErrors"] == {500: 1}
    assert s["totalBatches"] == 7
    assert s["totalRows"] == 1
    assert s["avgMs"] == 35.0
    assert s["maxMs"] == 60.0
    assert s["p50Ms"] == 35.0
    assert s["firstFailureAt"] is not None
    assert s["startedAt"] is not None and s["finishedAt"] is not None


def test_snapshot_contains_all_metric_fields():
    m = lm.LoadMetrics()
    m.mark_start()
    m.record(200, 0.1, rows=2)
    m.mark_finish()
    s = m.snapshot()
    for field, _ in lm.METRIC_FIELDS:
        assert field in s, f"missing metric field: {field}"
    assert s["totalRequests"] == 1 and s["successRate"] == 1.0


def test_actual_qps_empty_and_completion_window_pruning():
    m = lm.LoadMetrics(window_seconds=10)
    assert m.actual_qps() == 0.0
    now = time.monotonic()
    with m._lock:
        m._completion_times[:] = [now - 100, now - 5, now - 0.5]
        pruned = m._pruned_completion_times(now)
    assert pruned == [now - 5, now - 0.5]


def test_row_throughput_uses_wall_clock_span():
    m = lm.LoadMetrics()
    m.mark_start()
    m.record(200, 0.001, rows=10)
    m.mark_finish()
    with m._lock:
        m.started_at = "2026-01-01T10:00:00"
        m.finished_at = "2026-01-01T10:00:10"
    assert m.row_throughput() == 1.0


def test_snapshot_freezes_qps_after_finish():
    # 停止后快照的 QPS 必须定格：滚动窗口分母不再随时间变大（用户报告“停止后 QPS 还在变”）
    m = lm.LoadMetrics(window_seconds=10)
    m.mark_start()
    m.record(200, 0.01)
    m.record(200, 0.01)
    m.mark_finish()
    s1 = m.snapshot()
    time.sleep(0.15)
    s2 = m.snapshot()
    assert s1["actualQps"] > 0
    assert s1["actualQps"] == s2["actualQps"]
    assert s1["rowThroughput"] == s2["rowThroughput"]


def test_mark_start_resets_previous_round_metrics():
    # 单例跨轮复用：新一轮 mark_start 必须清空上一轮累计（防止多轮指标叠加）
    m = lm.LoadMetrics()
    m.mark_start()
    m.record(200, 0.01, rows=3)
    m.mark_finish()
    m.mark_start()
    s = m.snapshot()
    assert s["totalRequests"] == 0
    assert s["totalRows"] == 0
    assert s["startedAt"] is not None
    assert s["finishedAt"] is None


def test_metrics_thread_safe_concurrent_records():
    m = lm.LoadMetrics()

    def hit(i: int) -> None:
        m.record(200 if i % 3 else 500, 0.001)

    with ThreadPoolExecutor(max_workers=8) as pool:
        list(pool.map(hit, range(300)))
    s = m.snapshot()
    assert s["totalRequests"] == 300
    assert s["success"] == 200 and s["failed"] == 100
    assert s["httpErrors"] == {500: 100}


# ---------------------------------------------------------------------------
# P1 批量生成 generate_row_blocks（§18：档位/增量比例/字段宽度/一键停止）
# ---------------------------------------------------------------------------

def test_p1_row_blocks_chunked_by_batch_size():
    rows = [{"bizKey": "B", "sourceUpdatedAt": "2026-01-01T00:00:00"}]
    blocks = list(mock_db.generate_row_blocks(rows, 2500, batch_size=1000))
    assert [len(b) for b in blocks] == [1000, 1000, 500]


def test_p1_row_blocks_bizkey_unique_with_full_increment():
    rows = [{"bizKey": "B"}]
    blocks = list(mock_db.generate_row_blocks(rows, 50, increment_ratio=1.0, batch_size=10))
    keys = [r["bizKey"] for b in blocks for r in b]
    assert len(set(keys)) == 50
    assert all(k.startswith("B#LOAD-") for k in keys)


def test_p1_row_blocks_increment_ratio_reuses_template_keys():
    rows = [{"bizKey": "B"}]
    blocks = list(mock_db.generate_row_blocks(rows, 10, increment_ratio=0.5, batch_size=100))
    keys = [r["bizKey"] for b in blocks for r in b]
    assert keys[:5] == [f"B#LOAD-{i}" for i in range(5)]
    assert keys[5:] == ["B"] * 5


def test_p1_row_blocks_pads_text_fields():
    rows = [{"bizKey": "B", "note": "hi"}]
    (block,) = list(mock_db.generate_row_blocks(rows, 1, field_width=8, batch_size=10))
    row = block[0]
    assert row["note"] == "hixxxxxx"
    assert len(row["bizKey"]) >= 8


def test_p1_row_blocks_increment_column_advances():
    rows = [{"bizKey": "B", "sourceUpdatedAt": "2026-01-01T00:00:00"}]
    blocks = list(mock_db.generate_row_blocks(rows, 100, batch_size=30))
    vals = [r["sourceUpdatedAt"] for b in blocks for r in b]
    assert len(set(vals)) == 100


def test_p1_row_blocks_stop_event_before_and_midway():
    rows = [{"bizKey": "B"}]
    stop = threading.Event()
    stop.set()
    assert list(mock_db.generate_row_blocks(rows, 100, stop_event=stop)) == []
    stop2 = threading.Event()
    gen = mock_db.generate_row_blocks(rows, 100, batch_size=10, stop_event=stop2)
    first = next(gen)
    stop2.set()
    assert len(first) == 10
    assert list(gen) == []


# ---------------------------------------------------------------------------
# P2 压力模式（§19：分页截断/错误比例/限流/统计/恢复）
# ---------------------------------------------------------------------------

@pytest.fixture()
def p2client():
    state = mock_api.MockApiState()
    state.settings.auth_mode = "NONE"
    return TestClient(mock_api.create_app(state)), state


def _override(c, path: str, n: int) -> None:
    c.post("/__control/overrides",
           json={"path": path, "rows": [{"bizKey": f"K{i}", "v": i} for i in range(n)]})


def test_p2_page_size_applies_only_in_load_mode(p2client):
    c, _ = p2client
    _override(c, "/api/load-a", 10)
    c.post("/__control/settings", json={"page_size": 3})
    assert len(c.get("/api/load-a").json()["data"]["list"]) == 10  # 未开压力模式不截断
    c.post("/__control/settings", json={"load_mode": True})
    assert len(c.get("/api/load-a").json()["data"]["list"]) == 3


def test_p2_total_rows_limit_truncates_in_load_mode(p2client):
    c, _ = p2client
    _override(c, "/api/load-b", 10)
    c.post("/__control/settings", json={"load_mode": True, "total_rows_limit": 5})
    assert len(c.get("/api/load-b").json()["data"]["list"]) == 5


def test_p2_error_rate_injects_500(p2client):
    c, _ = p2client
    _override(c, "/api/load-c", 3)
    c.post("/__control/settings", json={"load_mode": True, "error_rate": 1.0})
    assert c.get("/api/load-c").status_code == 500


def test_p2_rate_limit_returns_429_beyond_threshold(p2client):
    c, _ = p2client
    _override(c, "/api/load-d", 3)
    c.post("/__control/settings", json={"load_mode": True, "rate_limit": 1})
    assert c.get("/api/load-d").status_code == 200
    assert c.get("/api/load-d").status_code == 429


def test_p2_load_stats_tracks_requests_and_errors(p2client):
    c, _ = p2client
    _override(c, "/api/load-e", 3)
    c.post("/__control/settings", json={"load_mode": True, "error_rate": 1.0})
    c.get("/api/load-e")
    stats = c.get("/__control/load-stats").json()
    assert stats["loadMode"] is True
    assert stats["stats"]["requests"] == 1
    # JSON 序列化后状态码键为字符串
    assert stats["stats"]["statusCounts"] == {"500": 1}
    assert set(stats["stats"]) >= {
        "requests", "active", "peakConcurrency", "qpsWindow",
        "avgMs", "maxMs", "statusCounts", "rateLimited"}


def test_p2_reset_clears_pressure_settings(p2client):
    c, _ = p2client
    c.post("/__control/settings", json={
        "load_mode": True, "page_size": 2, "total_rows_limit": 4,
        "random_delay": 1.5, "error_rate": 0.5, "rate_limit": 10})
    c.post("/__control/settings", json={"reset": True})
    s = c.get("/__control/settings").json()
    assert s["loadMode"] is False
    assert s["pageSize"] == 0 and s["totalRowsLimit"] == 0
    assert s["randomDelay"] == 0.0 and s["errorRate"] == 0.0 and s["rateLimit"] == 0.0