8aa8967c2b
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / build-and-deploy (push) Has been cancelled
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / tests (push) Has been cancelled
101 lines
3.0 KiB
Python
101 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from src.models.incident import Severity
|
|
from src.services import signal_observation_service as signal_observation_module
|
|
from src.services.signal_observation_service import (
|
|
build_signal_worker_provider_event_id,
|
|
enrich_incident_for_signal_observation,
|
|
format_signal_worker_event_content,
|
|
)
|
|
|
|
|
|
def _host_incident() -> SimpleNamespace:
|
|
signal = SimpleNamespace(
|
|
alert_name="HostErrorLogFlood",
|
|
labels={"error_count": "30"},
|
|
annotations={"summary": "30 ERROR log entries in last 5min"},
|
|
fingerprint="b000a87cfe4bc658",
|
|
)
|
|
return SimpleNamespace(
|
|
incident_id="INC-20260518-TEST01",
|
|
severity=Severity.P2,
|
|
affected_services=["ollama"],
|
|
signals=[signal],
|
|
)
|
|
|
|
|
|
def test_enrich_incident_for_signal_observation_restores_host_context() -> None:
|
|
incident = _host_incident()
|
|
|
|
labels = enrich_incident_for_signal_observation(
|
|
incident,
|
|
{
|
|
"source": "journal",
|
|
"namespace": "infra",
|
|
"target": "ollama",
|
|
"sensor_host": "ollama",
|
|
"sensor_ip": "192.168.0.188",
|
|
},
|
|
)
|
|
|
|
assert labels["alertname"] == "HostErrorLogFlood"
|
|
assert labels["namespace"] == "infra"
|
|
assert labels["target"] == "ollama"
|
|
assert labels["host"] == "192.168.0.188"
|
|
assert labels["sensor_host"] == "ollama"
|
|
assert incident.signals[0].labels["host"] == "192.168.0.188"
|
|
|
|
|
|
def test_signal_worker_provider_event_id_is_stable_and_bounded() -> None:
|
|
event_id = build_signal_worker_provider_event_id(
|
|
"1747528924000-0",
|
|
"b000a87cfe4bc658",
|
|
)
|
|
|
|
assert event_id == "signal-worker:received:1747528924000-0:b000a87cfe4bc658"
|
|
assert len(event_id) < 256
|
|
|
|
|
|
def test_signal_worker_event_content_is_operator_readable() -> None:
|
|
content = format_signal_worker_event_content(
|
|
incident_id="INC-20260518-TEST01",
|
|
alertname="HostErrorLogFlood",
|
|
severity="P2",
|
|
namespace="infra",
|
|
target="ollama",
|
|
fingerprint="b000a87cfe4bc658",
|
|
message_id="1747528924000-0",
|
|
signal_source="journal",
|
|
)
|
|
|
|
assert "Incident: INC-20260518-TEST01" in content
|
|
assert "Alert: HostErrorLogFlood" in content
|
|
assert "Telegram: not sent" in content
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_signal_observation_timeline_uses_explicit_project_context(monkeypatch) -> None:
|
|
calls: list[dict] = []
|
|
|
|
class FakeTimelineService:
|
|
async def add_event(self, **kwargs):
|
|
calls.append(kwargs)
|
|
|
|
monkeypatch.setattr(
|
|
"src.services.approval_db.get_timeline_service",
|
|
lambda: FakeTimelineService(),
|
|
)
|
|
|
|
await signal_observation_module._add_observation_timeline(
|
|
incident=_host_incident(),
|
|
signal_data={"alert_name": "HostErrorLogFlood"},
|
|
labels={"target": "ollama"},
|
|
raw_evidence_created=True,
|
|
)
|
|
|
|
assert calls[0]["project_id"] == "awoooi"
|