from __future__ import annotations import json from typing import Any import httpx from ariadne.services import hermes_autotriage_logs as logs_module WINDOW_START = "2026-08-05T10:00:00Z" WINDOW_END = "2026-08-05T10:30:00Z" PADDED_FROM = "2026-08-05T09:55:00Z" PADDED_TO = "2026-08-05T10:35:00Z" INCIDENT_ID = "INC-4242" DEFAULT_SIZE = 50 HARD_CAP = 100 DEFAULT_TIMEOUT = 5.0 TWO_PASSES = 2 def _config(**overrides: Any) -> dict: values: dict[str, Any] = { "opensearch_url": "http://opensearch:9200", "namespace": "hermes", "extra_namespaces": ["jenkins"], } values.update(overrides) return values def _hit(message: str = "ready", pod: str = "hermes-0") -> dict: return { "_index": "kube-2026.08.05", "_source": { "@timestamp": "2026-08-05T10:01:00Z", "message": message, "stream": "stdout", "kubernetes": { "namespace_name": "hermes", "pod_name": pod, "container_name": "app", }, }, } def _payload(hits: list, total: Any = None) -> dict: total_value = {"value": len(hits)} if total is None else total return {"hits": {"total": total_value, "hits": hits}} class FakeResponse: def __init__(self, payload: Any = None, status_code: int = 200, invalid_json: bool = False): self.status_code = status_code self._payload = payload self._invalid_json = invalid_json def json(self): # type: ignore[no-untyped-def] if self._invalid_json: raise ValueError("bad json") return self._payload class FakeClient: def __init__(self, script: list, calls: list): self._script = script self._calls = calls def __enter__(self): return self def __exit__(self, *args) -> None: # type: ignore[no-untyped-def] return None def post(self, url, json=None): # type: ignore[no-untyped-def] # noqa: A002 self._calls.append((url, json)) step = self._script.pop(0) if isinstance(step, Exception): raise step return step def _install(monkeypatch, script: list) -> tuple[list, list]: # type: ignore[no-untyped-def] calls: list[tuple[str, dict]] = [] client_kwargs: list[dict] = [] def factory(**kwargs): # type: ignore[no-untyped-def] client_kwargs.append(kwargs) return FakeClient(script, calls) monkeypatch.setattr(logs_module.httpx, "Client", factory) return calls, client_kwargs def test_first_pass_matches_incident_and_builds_bounded_query(monkeypatch) -> None: calls, client_kwargs = _install( monkeypatch, [FakeResponse(_payload([_hit(f"boom {INCIDENT_ID}")]))] ) result = logs_module.collect_log_evidence( _config(), INCIDENT_ID, WINDOW_START, WINDOW_END ) assert result["query_window"] == { "from": PADDED_FROM, "to": PADDED_TO, "correlation": "incident_id", } assert result["records"] == [ { "index": "kube-2026.08.05", "timestamp": "2026-08-05T10:01:00Z", "namespace": "hermes", "pod": "hermes-0", "container": "app", "message": f"boom {INCIDENT_ID}", } ] assert result["truncated"] is False assert result["error"] is None assert client_kwargs == [{"timeout": DEFAULT_TIMEOUT}] url, body = calls[0] assert url == "http://opensearch:9200/kube-*/_search" assert body["size"] == DEFAULT_SIZE assert body["sort"] == [{"@timestamp": {"order": "asc"}}] assert body["_source"] == logs_module._SOURCE_FIELDS # noqa: SLF001 assert body["query"]["bool"]["filter"] == [ {"terms": {"kubernetes.namespace_name": ["hermes", "jenkins"]}}, {"range": {"@timestamp": {"gte": PADDED_FROM, "lte": PADDED_TO}}}, ] assert body["query"]["bool"]["must"] == [{"match_phrase": {"message": INCIDENT_ID}}] def test_second_pass_drops_incident_filter_when_first_is_empty(monkeypatch) -> None: calls, _ = _install( monkeypatch, [FakeResponse(_payload([])), FakeResponse(_payload([_hit()]))] ) result = logs_module.collect_log_evidence( _config(), INCIDENT_ID, WINDOW_START, WINDOW_END ) assert len(calls) == TWO_PASSES assert "must" in calls[0][1]["query"]["bool"] assert "must" not in calls[1][1]["query"]["bool"] assert result["query_window"]["correlation"] == "window" assert result["records"][0]["pod"] == "hermes-0" assert result["error"] is None def test_empty_both_passes_is_success(monkeypatch) -> None: calls, _ = _install( monkeypatch, [FakeResponse(_payload([])), FakeResponse(_payload([]))] ) result = logs_module.collect_log_evidence( _config(), INCIDENT_ID, WINDOW_START, WINDOW_END ) assert len(calls) == TWO_PASSES assert result["records"] == [] assert result["truncated"] is False assert result["error"] is None assert result["query_window"]["correlation"] == "window" def test_timeout_returns_error_without_second_pass(monkeypatch) -> None: calls, _ = _install(monkeypatch, [httpx.TimeoutException("slow")]) result = logs_module.collect_log_evidence( _config(), INCIDENT_ID, WINDOW_START, WINDOW_END ) assert len(calls) == 1 assert result["error"] == "opensearch timeout" assert result["records"] == [] assert result["truncated"] is False assert result["query_window"]["correlation"] == "incident_id" def test_http_error_and_generic_failure_set_error(monkeypatch) -> None: _install(monkeypatch, [FakeResponse(status_code=500)]) result = logs_module.collect_log_evidence( _config(), INCIDENT_ID, WINDOW_START, WINDOW_END ) assert result["error"] == "opensearch http 500" assert result["records"] == [] _install(monkeypatch, [RuntimeError("connection refused")]) result = logs_module.collect_log_evidence( _config(), INCIDENT_ID, WINDOW_START, WINDOW_END ) assert result["error"] == "opensearch request failed: connection refused" def test_invalid_json_and_malformed_payload_set_error(monkeypatch) -> None: _install(monkeypatch, [FakeResponse(invalid_json=True)]) result = logs_module.collect_log_evidence( _config(), INCIDENT_ID, WINDOW_START, WINDOW_END ) assert result["error"] == "opensearch returned invalid json" _install(monkeypatch, [FakeResponse({"unexpected": True})]) result = logs_module.collect_log_evidence( _config(), INCIDENT_ID, WINDOW_START, WINDOW_END ) assert result["error"] == "malformed opensearch payload" assert result["records"] == [] def test_truncated_when_opensearch_reports_more_hits(monkeypatch) -> None: _install(monkeypatch, [FakeResponse(_payload([_hit()], total={"value": 500}))]) result = logs_module.collect_log_evidence( _config(), INCIDENT_ID, WINDOW_START, WINDOW_END ) assert result["truncated"] is True assert len(result["records"]) == 1 assert result["error"] is None _install(monkeypatch, [FakeResponse(_payload([_hit()], total=7))]) result = logs_module.collect_log_evidence( _config(), INCIDENT_ID, WINDOW_START, WINDOW_END ) assert result["truncated"] is True def test_truncated_by_byte_cap(monkeypatch) -> None: record = { "index": "kube-2026.08.05", "timestamp": "2026-08-05T10:01:00Z", "namespace": "hermes", "pod": "hermes-0", "container": "app", "message": "ready", } record_bytes = len(json.dumps(record, separators=(",", ":")).encode("utf-8")) hits = [_hit(), _hit(), _hit()] _install(monkeypatch, [FakeResponse(_payload(hits))]) result = logs_module.collect_log_evidence( _config(max_response_bytes=record_bytes + 1), INCIDENT_ID, WINDOW_START, WINDOW_END, ) assert result["records"] == [record] assert result["truncated"] is True assert result["error"] is None def test_zero_byte_budget_keeps_no_records_without_fallback(monkeypatch) -> None: calls, _ = _install(monkeypatch, [FakeResponse(_payload([_hit()]))]) result = logs_module.collect_log_evidence( _config(max_response_bytes=1), INCIDENT_ID, WINDOW_START, WINDOW_END ) assert len(calls) == 1 assert result["records"] == [] assert result["truncated"] is True assert result["query_window"]["correlation"] == "incident_id" def test_max_records_hard_cap_and_bad_values(monkeypatch) -> None: calls, client_kwargs = _install( monkeypatch, [FakeResponse(_payload([_hit()])), FakeResponse(_payload([_hit()]))] ) logs_module.collect_log_evidence( _config(max_records=250, timeout_seconds=2.5), INCIDENT_ID, WINDOW_START, WINDOW_END, ) logs_module.collect_log_evidence( _config(max_records="bad"), INCIDENT_ID, WINDOW_START, WINDOW_END ) assert calls[0][1]["size"] == HARD_CAP assert client_kwargs[0] == {"timeout": 2.5} assert calls[1][1]["size"] == DEFAULT_SIZE assert client_kwargs[1] == {"timeout": DEFAULT_TIMEOUT} def test_invalid_window_timestamps_skip_search(monkeypatch) -> None: calls, _ = _install(monkeypatch, []) result = logs_module.collect_log_evidence( _config(), INCIDENT_ID, "not-a-time", WINDOW_END ) assert calls == [] assert result == { "query_window": { "from": "not-a-time", "to": WINDOW_END, "correlation": "incident_id", }, "records": [], "truncated": False, "error": "invalid window timestamps", } def test_window_arithmetic_pads_five_minutes_and_assumes_utc() -> None: padded = logs_module._padded_window( # noqa: SLF001 "2026-08-05T10:00:00", "2026-08-05T10:30:00" ) assert padded == (PADDED_FROM, PADDED_TO) assert logs_module._padded_window(None, WINDOW_END) is None # noqa: SLF001 def test_malformed_hits_are_skipped_and_missing_total_is_success(monkeypatch) -> None: bad_source = {"_index": "kube-2026.08.05", "_source": "oops"} no_kubernetes = {"_index": "kube-2026.08.05", "_source": {"message": "plain"}} _install( monkeypatch, [FakeResponse({"hits": {"hits": ["junk", bad_source, no_kubernetes, _hit()]}})], ) result = logs_module.collect_log_evidence( _config(), INCIDENT_ID, WINDOW_START, WINDOW_END ) assert result["error"] is None assert result["truncated"] is False assert [item["pod"] for item in result["records"]] == ["", "hermes-0"] assert result["records"][0] == { "index": "kube-2026.08.05", "timestamp": "", "namespace": "", "pod": "", "container": "", "message": "plain", } def test_sanitize_masks_headers_and_tokens() -> None: sanitize = logs_module._sanitize # noqa: SLF001 assert sanitize("Authorization: Bearer abc.def") == "Authorization: [REDACTED]" assert sanitize("authorization:Basic Zm9vOmJhcg==") == "authorization:[REDACTED]" assert sanitize("retry with Bearer eyJhbGci.sig") == "retry with Bearer [REDACTED]" assert sanitize("Cookie: sid=abc; theme=dark") == "Cookie: [REDACTED]" assert sanitize("Set-Cookie: id=1; Path=/") == "Set-Cookie: [REDACTED]" assert sanitize("listening on port 8080") == "listening on port 8080" def test_sanitize_masks_credential_assignments() -> None: sanitize = logs_module._sanitize # noqa: SLF001 assert sanitize("password=hunter2 ok") == "password=[REDACTED] ok" assert sanitize("passwd: hunter2") == "passwd: [REDACTED]" assert sanitize("db_secret=s3cr3t") == "db_secret=[REDACTED]" assert sanitize("token: abc123") == "token: [REDACTED]" assert sanitize("api_key=k1") == "api_key=[REDACTED]" assert sanitize("apikey: k2") == "apikey: [REDACTED]" assert sanitize("access_key=AKIA123") == "access_key=[REDACTED]" assert sanitize("private_key=xyz") == "private_key=[REDACTED]" assert sanitize("access_token=opaque") == "access_token=[REDACTED]" assert sanitize("session=deadbeef;") == "session=[REDACTED];" assert sanitize("sessionid: deadbeef") == "sessionid: [REDACTED]" assert sanitize('"password": "hunter2"') == '"password": "[REDACTED]"' def test_sanitize_masks_pem_private_key_blocks() -> None: text = ( "before\n" "-----BEGIN RSA PRIVATE KEY-----\n" "MIIEowIBAAKCAQEA\nabcdef\n" "-----END RSA PRIVATE KEY-----\n" "after" ) masked = logs_module._sanitize(text) # noqa: SLF001 assert masked == ( "before\n" "-----BEGIN RSA PRIVATE KEY-----" "[REDACTED]" "-----END RSA PRIVATE KEY-----\n" "after" )