Merge remote-tracking branch 'gitea/main' into codex/p0-006-ai-log-triage-20260709
This commit is contained in:
@@ -1,9 +1,12 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from src.api.v1.iwooos import router
|
||||
from src.core.metrics import ALERT_CHAIN_LAST_SUCCESS
|
||||
from src.services.iwooos_security_operating_system import (
|
||||
load_latest_iwooos_security_operating_system,
|
||||
validate_iwooos_security_operation_packet,
|
||||
@@ -151,6 +154,52 @@ def test_iwooos_security_operating_system_api_counts_alertmanager_receipts(
|
||||
assert data["summary"]["runtime_gate_count"] == 0
|
||||
|
||||
|
||||
def test_iwooos_security_operating_system_api_uses_alert_chain_metric_fallback(
|
||||
monkeypatch,
|
||||
) -> None:
|
||||
async def recent_events_timeout(**_: object) -> dict[str, object]:
|
||||
raise TimeoutError
|
||||
|
||||
monkeypatch.setattr(
|
||||
"src.api.v1.iwooos.list_recent_channel_events",
|
||||
recent_events_timeout,
|
||||
)
|
||||
ALERT_CHAIN_LAST_SUCCESS.labels(source="alertmanager").set(time.time())
|
||||
|
||||
response = _client().get("/api/v1/iwooos/security-operating-system")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["summary"]["alert_receipt_accepted_count"] == 1
|
||||
assert data["summary"]["alert_receipt_observed_count"] == 1
|
||||
assert data["summary"]["alert_receipt_source_status"] == "ready"
|
||||
assert data["summary"]["alert_receipt_source_error"] is None
|
||||
|
||||
|
||||
def test_iwooos_security_operating_system_api_rejects_stale_metric_fallback(
|
||||
monkeypatch,
|
||||
) -> None:
|
||||
async def recent_events_timeout(**_: object) -> dict[str, object]:
|
||||
raise TimeoutError
|
||||
|
||||
monkeypatch.setattr(
|
||||
"src.api.v1.iwooos.list_recent_channel_events",
|
||||
recent_events_timeout,
|
||||
)
|
||||
ALERT_CHAIN_LAST_SUCCESS.labels(source="alertmanager").set(
|
||||
time.time() - (3 * 60 * 60)
|
||||
)
|
||||
|
||||
response = _client().get("/api/v1/iwooos/security-operating-system")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["summary"]["alert_receipt_accepted_count"] == 0
|
||||
assert data["summary"]["alert_receipt_observed_count"] == 0
|
||||
assert data["summary"]["alert_receipt_source_status"] == "source_unavailable"
|
||||
assert data["summary"]["alert_receipt_source_error"] == "TimeoutError"
|
||||
|
||||
|
||||
def test_iwooos_security_operation_packet_validator_accepts_redacted_loop() -> None:
|
||||
payload = validate_iwooos_security_operation_packet(_valid_operation_packet())
|
||||
|
||||
|
||||
Reference in New Issue
Block a user