"""第三阶段压力测试安全控制与 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