han-app/VM2_services/codebase/services/bitrix-sync/tests/test_telemetry.py

from __future__ import annotations

import json
import logging
from typing import Any

import pytest
from opentelemetry.sdk.trace import ReadableSpan
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter

from app import telemetry


@pytest.mark.parametrize(
    "canary",
    [
        "Bearer canary-authorization-value",
        "person@example.test",
        "+7 (999) 123-45-67",
        "https://portal.example/path?token=canary",
        "aaaaaaaaaaaaaaaaaaaa.bbbbbbbbbbbbbbbbbbbb.cccccccccc",
    ],
)
def test_redaction_removes_canaries_from_json_stdout(canary: str) -> None:
    record = logging.LogRecord(
        "test",
        logging.INFO,
        __file__,
        1,
        "safe event %s",
        (canary,),
        None,
    )
    record.event_name = "test.canary"
    record.telemetry_attributes = {
        "authorization": canary,
        "nested": {"email": canary, "safe": canary},
    }

    telemetry.RedactionFilter().filter(record)
    output = telemetry.JsonFormatter().format(record)

    assert canary not in output
    assert json.loads(output)["event"] == "test.canary"


def test_redaction_blocks_sensitive_keys_and_bounds_collections() -> None:
    result = telemetry.redact(
        {
            "db.statement": "SELECT secret FROM users",
            "object_key": "private/file.txt",
            "safe": list(range(100)),
        }
    )

    assert result["db.statement"] == "[REDACTED]"
    assert result["object_key"] == "[REDACTED]"
    assert len(result["safe"]) == 32


def test_redaction_drops_exception_text_before_otlp() -> None:
    try:
        raise RuntimeError("secret-token-in-exception")
    except RuntimeError:
        record = logging.LogRecord(
            "test",
            logging.ERROR,
            __file__,
            1,
            "operation failed",
            (),
            exc_info=__import__("sys").exc_info(),
        )

    telemetry.RedactionFilter().filter(record)

    assert record.exc_info is None
    assert record.telemetry_attributes["error.type"] == "RuntimeError"
    assert "secret-token-in-exception" not in telemetry.JsonFormatter().format(record)


def test_span_processor_replaces_immutable_sdk_attributes() -> None:
    exporter = InMemorySpanExporter()
    processor = telemetry.RedactingBatchSpanProcessor(exporter)
    span = ReadableSpan(
        name="crm.request",
        attributes={"url.full": "https://crm.example/?token=secret", "safe.outcome": "success"},
    )

    processor.on_end(span)
    processor.shutdown()

    assert span.attributes["url.full"] == "[REDACTED]"
    assert span.attributes["safe.outcome"] == "success"


def test_init_without_endpoint_is_backward_compatible(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("OTEL_EXPORTER_OTLP_ENDPOINT", raising=False)
    monkeypatch.setattr(telemetry, "_runtime", None)

    runtime = telemetry.init_telemetry("bitrix-sync-worker")

    assert runtime.tracer_provider is None
    assert runtime.meter_provider is None
    assert runtime.logger_provider is None
    telemetry.shutdown_telemetry()


def test_each_process_has_distinct_service_resource() -> None:
    names = {
        telemetry._resource(name).attributes["service.name"]  # noqa: SLF001
        for name in telemetry.SERVICE_NAMES
    }

    assert names == {
        "bitrix-sync-api",
        "bitrix-sync-worker",
        "bitrix-sync-reconciliation",
    }


def test_init_is_fail_open_when_exporter_construction_fails(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector:4317")
    monkeypatch.setattr(telemetry, "_runtime", None)

    def fail_exporter(**_kwargs: Any) -> None:
        raise RuntimeError("collector unavailable")

    monkeypatch.setattr(telemetry, "OTLPSpanExporter", fail_exporter)

    runtime = telemetry.init_telemetry("bitrix-sync-reconciliation")

    assert runtime.tracer_provider is None
    telemetry.shutdown_telemetry()


def test_business_metric_attributes_are_bounded(monkeypatch: pytest.MonkeyPatch) -> None:
    captured: list[dict[str, str]] = []

    class Instrument:
        def add(self, _value: int, attributes: dict[str, str]) -> None:
            captured.append(attributes)

        def record(self, _value: float, attributes: dict[str, str]) -> None:
            captured.append(attributes)

    class Meter:
        def create_counter(self, *_args: Any, **_kwargs: Any) -> Instrument:
            return Instrument()

        def create_histogram(self, *_args: Any, **_kwargs: Any) -> Instrument:
            return Instrument()

    monkeypatch.setattr(telemetry.metrics, "get_meter", lambda _name: Meter())

    telemetry.record_workflow("user-2f3c0a5e-identifier", "novel-state", 0.5)
    telemetry.record_crm("unknown.dynamic.method", "novel-state", 0.1)

    assert captured
    assert all("user-2f3c0a5e-identifier" not in values.values() for values in captured)
    assert all("novel-state" not in values.values() for values in captured)


def test_business_metrics_are_fail_open(monkeypatch: pytest.MonkeyPatch) -> None:
    def unavailable(_name: str) -> None:
        raise RuntimeError("metrics unavailable")

    monkeypatch.setattr(telemetry.metrics, "get_meter", unavailable)

    telemetry.record_claim("task", 1)