108 lines
3.4 KiB
Python
108 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
from src.services.channel_hub import (
|
|
build_grouped_alert_provider_event_id,
|
|
build_grouped_alert_run_id,
|
|
ensure_completed_shadow_run,
|
|
format_grouped_alert_digest_text,
|
|
format_grouped_alert_event_content,
|
|
)
|
|
|
|
|
|
def test_build_grouped_alert_provider_event_id_is_deterministic() -> None:
|
|
event_id = build_grouped_alert_provider_event_id(
|
|
"INC-20260507-ABCD12",
|
|
"1234567890abcdef" * 4,
|
|
)
|
|
|
|
assert event_id == "alert-group:INC-20260507-ABCD12:1234567890abcdef1234567890abcdef"
|
|
assert len(event_id) < 256
|
|
|
|
|
|
def test_build_grouped_alert_run_id_is_stable() -> None:
|
|
provider_event_id = build_grouped_alert_provider_event_id(
|
|
"INC-20260507-ABCD12",
|
|
"1234567890abcdef" * 4,
|
|
)
|
|
|
|
first = build_grouped_alert_run_id("awoooi", provider_event_id)
|
|
second = build_grouped_alert_run_id("awoooi", provider_event_id)
|
|
other_project = build_grouped_alert_run_id("ewoooc", provider_event_id)
|
|
|
|
assert first == second
|
|
assert first != other_project
|
|
|
|
|
|
def test_format_grouped_alert_event_content_keeps_operator_context() -> None:
|
|
content = format_grouped_alert_event_content(
|
|
alert_id="INC-20260507-ABCD12",
|
|
alertname="DockerContainerRestartSpike",
|
|
severity="critical",
|
|
namespace="default",
|
|
target_resource="sentry-self-hosted-events-consumer-1",
|
|
group_key="DockerContainerRestartSpike:default",
|
|
count=4,
|
|
parent_fingerprint="parent-fp",
|
|
fingerprint="child-fp",
|
|
)
|
|
|
|
assert "告警已收斂，不發 Telegram" in content
|
|
assert "Alert: DockerContainerRestartSpike" in content
|
|
assert "Target: sentry-self-hosted-events-consumer-1" in content
|
|
assert "Group Count: 4" in content
|
|
assert "Parent Fingerprint: parent-fp" in content
|
|
|
|
|
|
def test_format_grouped_alert_digest_text_is_html_safe() -> None:
|
|
content = format_grouped_alert_digest_text(
|
|
alertname="Docker<Restart>",
|
|
severity="critical",
|
|
namespace="default",
|
|
target_resource="sentry&snuba",
|
|
group_key="Docker<Restart>:default",
|
|
count=7,
|
|
)
|
|
|
|
assert "告警已收斂到父卡" in content
|
|
assert "Docker<Restart>" in content
|
|
assert "sentry&snuba" in content
|
|
assert "<b>7</b> 筆同組告警" in content
|
|
assert "AwoooP Run 監控" in content
|
|
|
|
|
|
class _FakeResult:
|
|
def fetchone(self) -> tuple[str] | None:
|
|
return ("created",)
|
|
|
|
|
|
class _FakeSession:
|
|
def __init__(self) -> None:
|
|
self.statement = ""
|
|
self.params = {}
|
|
|
|
async def execute(self, statement, params): # noqa: ANN001
|
|
self.statement = str(statement)
|
|
self.params = params
|
|
return _FakeResult()
|
|
|
|
|
|
async def test_completed_shadow_run_sets_run_state_not_null_defaults() -> None:
|
|
session = _FakeSession()
|
|
run_id = build_grouped_alert_run_id("awoooi", "provider-event")
|
|
|
|
inserted = await ensure_completed_shadow_run(
|
|
session, # type: ignore[arg-type]
|
|
project_id="awoooi",
|
|
run_id=run_id,
|
|
agent_id="legacy-telegram-gateway",
|
|
trigger_type="legacy_outbound",
|
|
trigger_ref="12713",
|
|
input_payload={"message_type": "final"},
|
|
)
|
|
|
|
assert inserted is True
|
|
assert "attempt_count, max_attempts, cost_usd, step_count" in session.statement
|
|
assert "0, 3, 0.0000, 0" in session.statement
|
|
assert session.params["project_id"] == "awoooi"
|
|
assert session.params["run_id"] == run_id
|