105 lines
3.8 KiB
Python
105 lines
3.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""可用性监控探针测试（issue #61）。
|
||
|
||
用本地 ThreadingHTTPServer 模拟健康端点：
|
||
1. 稳定 200 → 可用率 100% ≥ 99.8%；
|
||
2. 间歇 503 → 可用率按失败比例下降，探针统计正确；
|
||
3. 阈值判断：间歇失败超 0.2% 时 meets_target=False；
|
||
4. 失败明细记录（URL + 原因）。
|
||
"""
|
||
import http.server
|
||
import os
|
||
import sys
|
||
import threading
|
||
import unittest
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
from probe_availability import AvailabilityProbe # noqa: E402
|
||
|
||
|
||
class _HealthHandler(http.server.BaseHTTPRequestHandler):
|
||
"""可配置成功/失败模式的 /health 处理器。"""
|
||
|
||
fail_mode = [] # [True/False...] 按请求序号循环
|
||
|
||
def do_GET(self):
|
||
if self.path == "/health":
|
||
if self.fail_mode and self.fail_mode[0]:
|
||
self.fail_mode[0] = False # 一次性失败
|
||
self.send_response(503)
|
||
else:
|
||
self.send_response(200)
|
||
else:
|
||
self.send_response(404)
|
||
self.end_headers()
|
||
|
||
def log_message(self, *args): # 静默日志
|
||
pass
|
||
|
||
|
||
def start_server(fail_once=False):
|
||
handler = _HealthHandler
|
||
handler.fail_mode = [True] if fail_once else []
|
||
srv = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler)
|
||
thread = threading.Thread(target=srv.serve_forever, daemon=True)
|
||
thread.start()
|
||
return srv
|
||
|
||
|
||
class TestProbe(unittest.TestCase):
|
||
"""探针统计与阈值判断。"""
|
||
|
||
def test_stable_endpoint_100pct(self):
|
||
srv = start_server()
|
||
try:
|
||
url = f"http://127.0.0.1:{srv.server_port}"
|
||
probe = AvailabilityProbe([url], timeout_seconds=2)
|
||
report = probe.run(rounds=10, interval=0)
|
||
finally:
|
||
srv.shutdown()
|
||
self.assertEqual(report.total, 10)
|
||
self.assertEqual(report.success, 10)
|
||
self.assertEqual(report.availability, 1.0)
|
||
self.assertTrue(report.meets_target())
|
||
|
||
def test_single_failure_below_threshold(self):
|
||
srv = start_server(fail_once=True)
|
||
try:
|
||
url = f"http://127.0.0.1:{srv.server_port}"
|
||
probe = AvailabilityProbe([url], timeout_seconds=2)
|
||
report = probe.run(rounds=500, interval=0) # 1/500 = 0.2%
|
||
finally:
|
||
srv.shutdown()
|
||
self.assertEqual(len(report.failures), 1)
|
||
# 1/500 = 0.002 = 99.8% —— 恰好达标（>= target 判定通过）
|
||
self.assertGreaterEqual(report.availability, report.target)
|
||
self.assertTrue(report.meets_target())
|
||
# urllib 对非 200 抛 HTTPError → 记为不可用
|
||
self.assertFalse(report.failures[0].ok)
|
||
self.assertIn("503", report.failures[0].detail)
|
||
|
||
def test_high_failure_rate_fails_threshold(self):
|
||
# 构造 50% 失败的端点（fail_mode 每次循环交替：用两个处理器状态不可行，
|
||
# 改为直接 mock probe_once）
|
||
probe = AvailabilityProbe(["http://x"], timeout_seconds=1)
|
||
results = [probe.probe_once.__self__.__class__] # noqa - 占位
|
||
from unittest import mock
|
||
ok_result = mock.MagicMock()
|
||
ok_result.ok = True
|
||
bad_result = mock.MagicMock()
|
||
bad_result.ok = False
|
||
bad_result.url = "http://x"
|
||
bad_result.detail = "http-503"
|
||
bad_result.latency_ms = 1.0
|
||
# 交替 200/503 → 6 次探测 3 成功 → 50% 可用率，远低于 99.8%
|
||
with mock.patch.object(probe, "probe_once",
|
||
side_effect=[ok_result, bad_result] * 3):
|
||
report = probe.run(rounds=6, interval=0)
|
||
self.assertEqual(report.availability, 0.5)
|
||
self.assertFalse(report.meets_target())
|
||
self.assertEqual(len(report.failures), 3)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|