wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

fix(cd): retry alert chain api health smoke
All checks were successful
CD Pipeline / tests (push) Successful in 1m21s
Details
Code Review / ai-code-review (push) Successful in 14s
Details
CD Pipeline / build-and-deploy (push) Successful in 3m40s
Details
CD Pipeline / post-deploy-checks (push) Successful in 1m40s
Details

Browse Source
This commit is contained in:
Your Name
2026-06-01 18:38:48 +08:00
parent cc92eb0294
commit 0746543b0a
3 changed files with 192 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
apps/api/tests/test_alert_chain_smoke_metric.py
View File

@@ -47,6 +47,78 @@ class AlertChainSmokeMetricTest(unittest.TestCase):
self.assertTrue(result.passed)
self.assertIn("非阻塞降級: ollama", result.message)
def test_api_health_retries_transient_connection_failure(self):
calls = []
def fake_get(url, *, params=None, timeout=None):
self.assertTrue(url.endswith("/api/v1/health"))
calls.append({"url": url, "timeout": timeout})
if len(calls) == 1:
raise alert_chain_smoke_test.URLError("timed out")
return alert_chain_smoke_test.HttpGetResult(
200,
json.dumps(
{
"status": "healthy",
"environment": "prod",
"components": {
"api": {"status": "up"},
"postgresql": {"status": "up"},
"redis": {"status": "up"},
},
}
),
)
original_get = alert_chain_smoke_test.http_get
original_attempts = alert_chain_smoke_test.API_HEALTH_ATTEMPTS
original_timeout = alert_chain_smoke_test.API_HEALTH_TIMEOUT
original_delay = alert_chain_smoke_test.API_HEALTH_RETRY_DELAY
try:
alert_chain_smoke_test.http_get = fake_get
alert_chain_smoke_test.API_HEALTH_ATTEMPTS = 3
alert_chain_smoke_test.API_HEALTH_TIMEOUT = 20
alert_chain_smoke_test.API_HEALTH_RETRY_DELAY = 0
result = alert_chain_smoke_test.check_api_health("http://api")
finally:
alert_chain_smoke_test.http_get = original_get
alert_chain_smoke_test.API_HEALTH_ATTEMPTS = original_attempts
alert_chain_smoke_test.API_HEALTH_TIMEOUT = original_timeout
alert_chain_smoke_test.API_HEALTH_RETRY_DELAY = original_delay
self.assertTrue(result.passed)
self.assertEqual(len(calls), 2)
self.assertEqual({call["timeout"] for call in calls}, {20})
def test_api_health_reports_attempts_after_retry_exhaustion(self):
calls = []
def fake_get(url, *, params=None, timeout=None):
self.assertTrue(url.endswith("/api/v1/health"))
calls.append(timeout)
raise TimeoutError("timed out")
original_get = alert_chain_smoke_test.http_get
original_attempts = alert_chain_smoke_test.API_HEALTH_ATTEMPTS
original_timeout = alert_chain_smoke_test.API_HEALTH_TIMEOUT
original_delay = alert_chain_smoke_test.API_HEALTH_RETRY_DELAY
try:
alert_chain_smoke_test.http_get = fake_get
alert_chain_smoke_test.API_HEALTH_ATTEMPTS = 2
alert_chain_smoke_test.API_HEALTH_TIMEOUT = 7
alert_chain_smoke_test.API_HEALTH_RETRY_DELAY = 0
result = alert_chain_smoke_test.check_api_health("http://api")
finally:
alert_chain_smoke_test.http_get = original_get
alert_chain_smoke_test.API_HEALTH_ATTEMPTS = original_attempts
alert_chain_smoke_test.API_HEALTH_TIMEOUT = original_timeout
alert_chain_smoke_test.API_HEALTH_RETRY_DELAY = original_delay
self.assertFalse(result.passed)
self.assertEqual(calls, [7, 7])
self.assertIn("attempts=2", result.message)
self.assertIn("timeout=7s", result.message)
def test_api_health_fails_when_core_component_is_down(self):
def fake_get(url, *, params=None, timeout=None):
self.assertTrue(url.endswith("/api/v1/health"))