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

"""第三阶段压力测试安全控制与 P3 执行器测试（任务书 §16/§20/§21 验收表）。"""
from __future__ import annotations

import threading
import time

import pytest

from catalog import sample_adapters as sa
from load import controls
from load.metrics import LoadMetrics
from load.runner_p3 import P3Runner, _RateLimiter


def _valid_plan(**over) -> dict:
    plan = {
        "channel": "API_INBOUND",
        "environment": "dev", "tenant": "T1", "owner": "qa", "timeWindow": "t",
        "maxDataRows": 1000, "maxConcurrency": 4, "maxQps": 10, "maxDurationSeconds": 5,
        "stopThreshold": "连续 5 次错误/超时/429 自动停止", "cleanupPlan": "clean",
        "confirmed": True,
    }
    plan.update(over)
    return plan


# ---------------------------------------------------------------------------
# validate_plan（§16 十项必填 / 上限 / 二次确认）
# ---------------------------------------------------------------------------

def test_validate_plan_accepts_complete_plan():
    assert controls.validate_plan(_valid_plan()) == []


def test_validate_plan_rejects_missing_required_fields():
    errors = controls.validate_plan({"channel": "API_INBOUND"})
    assert len(errors) >= 10
    assert any("environment" in e for e in errors)


def test_validate_plan_requires_owner_confirmation():
    errors = controls.validate_plan(_valid_plan(confirmed=False))
    assert any("负责人确认" in e for e in errors)


def test_validate_plan_blocks_over_limit_without_escalation():
    errors = controls.validate_plan(_valid_plan(maxQps=999))
    assert any("超过保守上限" in e for e in errors)


def test_validate_plan_allows_escalation_but_blocks_hard_limit():
    ok = controls.validate_plan(_valid_plan(maxQps=60, escalationConfirm=True))
    assert ok == []
    errors = controls.validate_plan(_valid_plan(maxQps=10000, escalationConfirm=True))
    assert any("硬上限" in e for e in errors)


def test_validate_plan_rejects_bad_channel_and_error_rate_range():
    errors = controls.validate_plan(_valid_plan(channel="NOPE"))
    assert any("channel" in e for e in errors)
    errors = controls.validate_plan(_valid_plan(errorRate=1.5))
    assert any("errorRate" in e for e in errors)


# ---------------------------------------------------------------------------
# 租户字段剔除（§20：请求体不得携带任意 tenantId）
# ---------------------------------------------------------------------------

def test_strip_tenant_fields_removes_tenant_keys():
    row = {"ItemNum": "A", "tenantId": "999", "tenant_id": "888",
           "Domain": "SIM", "bizKey": "K1"}
    stripped = controls.strip_tenant_fields(dict(row))
    assert "tenantId" not in stripped and "tenant_id" not in stripped
    assert stripped["Domain"] == "SIM" and stripped["bizKey"] == "K1"


# ---------------------------------------------------------------------------
# AutoStopTracker（§20 自动停止阈值）
# ---------------------------------------------------------------------------

def test_auto_stop_tracker_consecutive_errors():
    tracker = controls.AutoStopTracker(consecutive_errors=3)
    assert not tracker.record(202)
    assert not tracker.record(500)
    assert not tracker.record(500)
    assert tracker.record(500)
    assert tracker.tripped_by == "consecutiveErrors"


def test_auto_stop_tracker_timeouts_and_429():
    tracker = controls.AutoStopTracker(timeouts=2, rate_limited=2)
    tracker.record(0)
    assert tracker.record(0) and tracker.tripped_by == "timeouts"
    tracker2 = controls.AutoStopTracker(timeouts=2, rate_limited=2)
    tracker2.record(429)
    assert tracker2.record(429) and tracker2.tripped_by == "rateLimited"


def test_auto_stop_tracker_success_resets_consecutive():
    tracker = controls.AutoStopTracker(consecutive_errors=3)
    tracker.record(500)
    tracker.record(500)
    tracker.record(202)
    assert not tracker.record(500)
    assert not tracker.record(500)


# ---------------------------------------------------------------------------
# LoadGuard 生命周期（默认关闭 / 立即停止 / 无残留）
# ---------------------------------------------------------------------------

def test_guard_default_state_is_idle_and_blocks_start():
    guard = controls.LoadGuard()
    assert guard.state == "idle"
    with pytest.raises(controls.ValidationError):
        guard.start()
    assert guard.snapshot()["noResidual"] is True


def test_guard_lifecycle_ready_running_stopping_stopped():
    guard = controls.LoadGuard()
    assert guard.submit_plan(_valid_plan())["ok"]
    assert guard.state == "ready"
    run_id = guard.start()
    assert guard.state == "running" and run_id.startswith("LOAD-")
    assert guard.request_stop("manual")["state"] == "stopping"
    result = guard.finish()
    assert result["ok"] and result["residualThreads"] == []
    assert guard.state == "stopped"


def test_guard_finish_reports_residual_worker():
    guard = controls.LoadGuard()
    guard.submit_plan(_valid_plan())
    guard.start()
    worker = threading.Thread(target=lambda: time.sleep(0.2), daemon=True)
    worker.start()
    guard.register_worker(worker)
    guard.request_stop()
    result = guard.finish()
    assert not result["ok"] and "residualThreads" in result
    worker.join(timeout=2)


def test_guard_finish_excludes_current_thread_from_residual():
    # P1/P3 执行器线程注册自身为 worker 并在自身线程内 finish()，
    # 不得把当前线程误报为残留（回归：曾导致每次收尾都报 residualThreads）
    guard = controls.LoadGuard()
    guard.submit_plan(_valid_plan())
    guard.start()
    result = {}

    def _finish_in_worker():
        guard.register_worker(threading.current_thread())
        guard.request_stop()
        result.update(guard.finish())

    worker = threading.Thread(target=_finish_in_worker, daemon=True)
    worker.start()
    worker.join(timeout=5)
    assert result["ok"] is True and result["residualThreads"] == []
    assert guard.state == "stopped"


# ---------------------------------------------------------------------------
# P3 执行器（固定 QPS 短跑 / 立即停止无残留 / 自动停止 / 错误比例注入）
# ---------------------------------------------------------------------------

class _FakeInboundClient:
    """进程内假客户端：捕获调用并返回固定状态，不触网。"""

    def __init__(self, status: int = 202):
        self.status = status
        self.calls: list[tuple] = []

    def op_push(self, entity, envelope, **kwargs):
        self.calls.append(("push", entity, envelope, kwargs))
        return self.status, {"ok": True}, {}

    def op_bulk(self, entity, lines, **kwargs):
        self.calls.append(("bulk", entity, lines, kwargs))
        return self.status, {"ok": True}, {}


def _p3_env(plan_over: dict | None = None, client_status: int = 202):
    guard = controls.LoadGuard()
    metrics = LoadMetrics()
    client = _FakeInboundClient(status=client_status)
    obj = next(o for o in sa.load_catalog()["objects"]
               if (o.get("apiInbound") or {}).get("supported"))
    entity = obj["apiInbound"]["entityCode"]
    rows = sa.adapt_inbound_rows(obj, "T-LOAD")["rows"]
    plan = _valid_plan(**{"entityCode": entity, "runId": "T-LOAD"})
    plan.update(plan_over or {})
    return guard, metrics, client, entity, rows, plan


def test_p3_runner_short_run_records_metrics_and_finishes_clean():
    guard, metrics, client, entity, rows, plan = _p3_env(
        {"maxDurationSeconds": 1, "maxQps": 30, "maxConcurrency": 4,
         "maxConsecutiveErrors": 5, "maxTimeouts": 5, "maxRateLimited": 5})
    guard.submit_plan(plan)
    guard.start()
    runner = P3Runner(client, entity, rows, plan, guard, metrics)
    runner.start()
    runner.join(timeout=10)
    assert not runner.is_alive()
    assert guard.state == "stopped"
    assert guard.snapshot()["noResidual"] is True
    snap = metrics.snapshot()
    assert snap["totalRequests"] > 0
    assert snap["success"] == snap["totalRequests"]
    assert snap["successRate"] == 1.0
    assert snap["p50Ms"] >= 0 and snap["startedAt"] and snap["finishedAt"]


def test_p3_runner_immediate_stop_leaves_no_residual():
    guard, metrics, client, entity, rows, plan = _p3_env(
        {"maxDurationSeconds": 10, "maxQps": 5, "maxConcurrency": 2})
    guard.submit_plan(plan)
    guard.start()
    runner = P3Runner(client, entity, rows, plan, guard, metrics)
    runner.start()
    time.sleep(0.2)
    guard.request_stop("manual stop")
    runner.join(timeout=10)
    assert guard.state == "stopped"
    assert guard.snapshot()["noResidual"] is True
    assert metrics.snapshot()["totalRequests"] < 100


def test_p3_runner_stop_cancels_queued_futures():
    # 回归：stop 后只等在途请求，排队未启动的任务必须取消（否则慢请求下收尾拖沓）
    guard, metrics, client, entity, rows, plan = _p3_env(
        {"maxDurationSeconds": 30, "maxQps": 50, "maxConcurrency": 2})
    guard.submit_plan(plan)
    guard.start()

    def slow_push(entity, envelope, **kwargs):
        time.sleep(0.5)
        client.calls.append(("push", entity, envelope, kwargs))
        return 202, {"ok": True}, {}

    client.op_push = slow_push
    runner = P3Runner(client, entity, rows, plan, guard, metrics)
    runner.start()
    time.sleep(1.0)
    guard.request_stop("manual stop")
    started = time.monotonic()
    runner.join(timeout=10)
    elapsed = time.monotonic() - started
    assert not runner.is_alive()
    assert guard.state == "stopped"
    assert elapsed < 5, f"停止收尾过慢：{elapsed:.1f}s（排队任务未被取消？）"
    assert guard.snapshot()["noResidual"] is True


def test_p3_runner_auto_stops_on_consecutive_errors():
    guard, metrics, client, entity, rows, plan = _p3_env(
        {"maxDurationSeconds": 30, "maxQps": 50, "maxConcurrency": 2,
         "maxConsecutiveErrors": 2},
        client_status=500)
    guard.submit_plan(plan)
    guard.start()
    runner = P3Runner(client, entity, rows, plan, guard, metrics)
    runner.start()
    runner.join(timeout=10)
    assert "自动停止" in (guard.stop_reason or "")
    assert guard.state == "stopped"
    assert metrics.snapshot()["failed"] >= 2


def test_p3_runner_error_rate_injects_missing_required_field():
    # errorRate 硬上限 0.9；提交计划后改回 1.0 保证每行都缺必填字段（确定性断言注入逻辑）
    guard, metrics, client, entity, rows, plan = _p3_env(
        {"maxDurationSeconds": 1, "maxQps": 10, "maxConcurrency": 2,
         "errorRate": 0.9, "escalationConfirm": True})
    guard.submit_plan(plan)
    plan["errorRate"] = 1.0
    guard.start()
    runner = P3Runner(client, entity, rows, plan, guard, metrics)
    runner.start()
    runner.join(timeout=10)
    assert client.calls, "压测应发出请求"
    first_required = runner._first_required
    for call in client.calls[:3]:
        envelope = call[2]
        assert first_required not in envelope["data"]["list"][0]


def test_p3_runner_never_injects_tenant_fields():
    guard, metrics, client, entity, rows, plan = _p3_env(
        {"maxDurationSeconds": 1, "maxQps": 10, "maxConcurrency": 2})
    guard.submit_plan(plan)
    guard.start()
    runner = P3Runner(client, entity, rows, plan, guard, metrics)
    runner.start()
    runner.join(timeout=10)
    for call in client.calls:
        row = call[2]["data"]["list"][0]
        assert "tenantId" not in row and "tenant_id" not in row


def test_rate_limiter_caps_qps_approximately():
    limiter = _RateLimiter(qps=100)
    started = time.monotonic()
    for _ in range(5):
        limiter.wait()
    elapsed = time.monotonic() - started
    # 5 个槽位应至少占 4 个间隔（4*10ms），宽松断言避免 CI 抖动误报
    assert elapsed >= 0.03