from __future__ import annotations import json import logging import uuid from app.utils import logging as logging_utils from app.utils import structured_otlp from app.utils import structured_pubsub class _FakeFuture: def add_done_callback(self, callback): callback(self) def result(self): return "message-id" class _FakePublisher: def __init__(self) -> None: self.calls = [] def publish(self, *args, **kwargs): self.calls.append((args, kwargs)) return _FakeFuture() def topic_path(self, project_id: str, topic: str) -> str: return f"projects/{project_id}/topics/{topic}" def test_log_structured_event_publishes_to_pubsub_when_configured(monkeypatch) -> None: published_events = [] monkeypatch.setenv("STRUCTURED_EVENT_LOG_ENABLED", "1") monkeypatch.setattr(logging_utils, "publish_structured_event", published_events.append) logger = logging.getLogger("test.logging_pubsub") logger.handlers = [logging.NullHandler()] logger.setLevel(logging.INFO) event = logging_utils.log_structured_event( logger, logging_utils.StructuredLogContext( callid="call-1", session_id="session-1", num_telefone="5511999999999", cod_ani="1234", nome_agente="conta", ), tipo_evento=logging_utils.EVENT_RECEBIMENTO_MSG, message_id="message-1", inicio_ns=1_700_000_000_000_000_000, ) assert event is not None assert published_events == [event] def test_log_structured_event_publishes_to_pubsub_and_otlp_when_trace_id_is_valid(monkeypatch) -> None: published_events = [] published_spans = [] monkeypatch.setenv("STRUCTURED_EVENT_LOG_ENABLED", "1") monkeypatch.setattr(logging_utils, "publish_structured_event", published_events.append) monkeypatch.setattr(logging_utils, "publish_structured_span", published_spans.append) logger = logging.getLogger("test.logging_pubsub_otlp") logger.handlers = [logging.NullHandler()] logger.setLevel(logging.INFO) event = logging_utils.log_structured_event( logger, logging_utils.StructuredLogContext( callid="call-1", session_id="550e8400-e29b-41d4-a716-446655440101", num_telefone="5511999999999", cod_ani="1234", nome_agente="conta", ), tipo_evento=logging_utils.EVENT_RECEBIMENTO_MSG, message_id="message-1", inicio_ns=1_700_000_000_000_000_000, ) assert event is not None assert published_events == [event] assert published_spans == [event] def test_publish_structured_event_sends_json_to_configured_topic(monkeypatch) -> None: fake_publisher = _FakePublisher() monkeypatch.setenv("GCP_PROJECT_ID", "project-1") monkeypatch.setenv("AGENT_PUBSUB_TOPIC", "agent-logs") monkeypatch.setattr(structured_pubsub, "_PUBLISHER", fake_publisher) monkeypatch.setattr(structured_pubsub, "_TOPIC_PATH", "projects/project-1/topics/agent-logs") event = { "tipo_evento": "recebimento msg", "dat_hora_inicio": "2026-05-11T22:07:29.965Z", "dat_hora_fim": "2026-05-11T22:07:30.965Z", "callid": "call-1", "session_id": "session-1", "nome_agente": "conta", "cod_ani": "1234", "message_id": "message-1", "original_message_id": "original-message-1", "http_cod_status": 200, "http_cod_desc": "OK", } structured_pubsub.publish_structured_event(event) assert len(fake_publisher.calls) == 1 args, kwargs = fake_publisher.calls[0] assert args[0] == "projects/project-1/topics/agent-logs" payload = json.loads(args[1].decode("utf-8")) assert payload == { "tipo_evento": "recebimento msg", "dat_hora_inicio": "11/05/2026 19:07:29,965000000", "dat_hora_termino": "11/05/2026 19:07:30,965000000", "callid": "call-1", "sessionId": "session-1", "nome_agente": "conta", "cod_ani": "1234", "message_id": "message-1", "http_cod_status": 200, "http_cod_desc": "OK", } assert "session_id" not in payload assert "dat_hora_fim" not in payload assert "original_message_id" not in payload assert kwargs == { "tipo_evento": "recebimento msg", "dat_hora_inicio": "11/05/2026 19:07:29,965000000", "dat_hora_termino": "11/05/2026 19:07:30,965000000", "sessionId": "session-1", "callid": "call-1", "nome_agente": "conta", "cod_ani": "1234", "http_cod_status": "200", "http_cod_desc": "OK", } def test_log_structured_event_skips_pubsub_without_required_env(monkeypatch) -> None: fake_publisher = _FakePublisher() monkeypatch.delenv("GCP_PROJECT_ID", raising=False) monkeypatch.delenv("AGENT_PUBSUB_TOPIC", raising=False) monkeypatch.setattr(structured_pubsub, "_PUBLISHER", fake_publisher) monkeypatch.setattr(structured_pubsub, "_TOPIC_PATH", "projects/project-1/topics/agent-logs") structured_pubsub.publish_structured_event({"tipo_evento": "envio msg"}) assert fake_publisher.calls == [] def test_build_structured_event_includes_session_id_and_lowercase_nome_agente() -> None: event = logging_utils.build_structured_event( logging_utils.StructuredLogContext( callid="call-1", session_id="session-1", num_telefone="5511999999999", cod_ani="1234", nome_agente="conta", ), tipo_evento=logging_utils.EVENT_RECEBIMENTO_MSG, message_id="message-1", inicio_ns=1_700_000_000_000_000_000, ) assert event["session_id"] == "session-1" assert event["nome_agente"] == "conta" assert event["dat_hora_fim"] == "" assert event["latencia_total_STT_TTS"] == "" assert event["latencia_TFFB_STT_TTS"] == "" assert event["duracao_audio"] == "" assert event["tts_max_gap_ms"] == "" assert event["tts_max_underrun_0ms"] == "" assert event["tts_underflow_count"] == "" assert event["tts_avg_underflow_ms"] == "" assert "tts_clear_rtt_ms" not in event assert "tts_connection_queue_wait_ms" not in event assert "tts_clear_stale_messages" not in event assert "tts_clear_stale_audio_bytes" not in event assert event["interrupcao"] == "" assert event["erro_msg"] == "" assert event["erro_detalhe"] == "" assert event["http_cod_status"] == "" assert event["http_cod_desc"] == "" assert event["finalizacao"] == "" assert all(value is not None for value in event.values()) assert "Nome_agente" not in event def test_build_structured_event_includes_audio_duration_and_interruption_flag() -> None: event = logging_utils.build_structured_event( logging_utils.StructuredLogContext( callid="call-1", session_id="session-1", num_telefone="5511999999999", cod_ani="1234", nome_agente="conta", ), tipo_evento=logging_utils.EVENT_ENVIO_MSG, message_id="message-1", inicio_ns=1_700_000_000_000_000_000, duracao_audio_ms=2400, tts_max_gap_ms=840, tts_max_underrun_0ms=125, interrupcao=True, ) assert event["duracao_audio"] == 2400 assert event["tts_max_gap_ms"] == 840 assert event["tts_max_underrun_0ms"] == 125 assert event["interrupcao"] == 1 def test_build_structured_event_uses_uuid_when_message_id_is_missing(monkeypatch) -> None: generated = uuid.UUID("12345678-1234-4234-9234-123456789abc") monkeypatch.setattr(logging_utils.uuid, "uuid4", lambda: generated) event = logging_utils.build_structured_event( logging_utils.StructuredLogContext( callid="call-1", session_id="session-1", num_telefone="5511999999999", cod_ani="1234", nome_agente="conta", ), tipo_evento=logging_utils.EVENT_ENVIO_MSG, inicio_ns=1_700_000_000_000_000_000, ) assert event["message_id"] == "12345678-1234-4234-9234-123456789abc" def test_error_message_from_resource_uses_event_specific_contract() -> None: assert ( logging_utils.error_message_from_resource( resource="agent_runtime", tipo_evento=logging_utils.EVENT_RECEBIMENTO_MSG, ) == "Falha TIA" ) assert ( logging_utils.error_message_from_resource( resource="stt", tipo_evento=logging_utils.EVENT_RECEBIMENTO_MSG, ) == "Falha STT" ) assert ( logging_utils.error_message_from_resource( resource="bridge", tipo_evento=logging_utils.EVENT_RECEBIMENTO_MSG, ) == "Falha comunicacao" ) assert ( logging_utils.error_message_from_resource( status="stop_silencio_longo", tipo_evento=logging_utils.EVENT_RECEBIMENTO_MSG, ) == "Silencio Longo" ) assert ( logging_utils.error_message_from_resource( resource="agent_backend", tipo_evento=logging_utils.EVENT_ENVIO_MSG, ) == "Falha comunicacao" ) assert ( logging_utils.error_message_from_resource( resource="tts", tipo_evento=logging_utils.EVENT_ENVIO_MSG, ) == "Falha TTS" ) assert ( logging_utils.error_message_from_resource( status="transferred", tipo_evento=logging_utils.EVENT_ENVIO_MSG, ) == "Transferido" ) def test_format_event_timestamp_uses_iso_utc_milliseconds() -> None: assert logging_utils.format_event_timestamp(1_700_000_001_234_567_890) == ( "2023-11-14T22:13:21.234Z" ) def test_structured_context_from_start_data_keeps_session_id_empty_when_not_provided() -> None: context = logging_utils.structured_context_from_start_data( { "callIdGed": "ged-1", "gsm": "5511999999999", "ani": "1234", "agent": "conta", } ) assert context.callid == "ged-1" assert context.session_id == "" def test_topic_path_accepts_short_topic_or_full_resource_name() -> None: fake_publisher = _FakePublisher() assert structured_pubsub._topic_path(fake_publisher, "project-1", "agent-logs") == ( "projects/project-1/topics/agent-logs" ) assert structured_pubsub._topic_path( fake_publisher, "project-1", "projects/another-project/topics/agent-logs", ) == "projects/another-project/topics/agent-logs" def test_trace_id_from_session_id_sanitizes_uuid_and_rejects_invalid_values() -> None: assert structured_otlp.trace_id_from_session_id( "550e8400-e29b-41d4-a716-446655440101" ) == "550e8400e29b41d4a716446655440101" assert structured_otlp.trace_id_from_session_id("sess-001") == "" assert structured_otlp.trace_id_from_session_id("00000000-0000-0000-0000-000000000000") == "" def _install_memory_tracer(monkeypatch): from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter endpoint = "http://otel.example/v1/traces" exporter = InMemorySpanExporter() provider = TracerProvider(id_generator=structured_otlp._ID_GENERATOR) provider.add_span_processor(SimpleSpanProcessor(exporter)) monkeypatch.setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", endpoint) monkeypatch.setattr(structured_otlp, "_TRACER_PROVIDER", provider) monkeypatch.setattr(structured_otlp, "_TRACER", provider.get_tracer("test.structured_otlp")) monkeypatch.setattr(structured_otlp, "_TRACER_ENDPOINT", endpoint) return exporter def test_publish_structured_span_exports_otlp_span_with_timestamps_and_attributes(monkeypatch) -> None: exporter = _install_memory_tracer(monkeypatch) start_ns = 1_700_000_000_000_000_000 end_ns = 1_700_000_001_234_000_000 event = { "tipo_evento": "envio msg", "message_id": "message-1", "dat_hora_inicio": logging_utils.format_event_timestamp(start_ns), "dat_hora_fim": logging_utils.format_event_timestamp(end_ns), "callid": "call-1", "session_id": "550e8400-e29b-41d4-a716-446655440101", "num_telefone": "5511999999999", "cod_ani": "1234", "latencia_total_STT_TTS": 1234, "latencia_TFFB_STT_TTS": 123, "erro_msg": "Falha TTS", "erro_detalhe": "", "http_cod_status": "", "http_cod_desc": "", "finalizacao": "", "nome_agente": "conta", "original_message_id": "original-message-1", } structured_otlp.publish_structured_span(event) spans = exporter.get_finished_spans() assert len(spans) == 1 span = spans[0] assert span.name == "structured_log.envio msg" assert span.context.trace_id == int("550e8400e29b41d4a716446655440101", 16) assert span.context.span_id != 0 assert span.parent is None assert span.start_time == start_ns assert span.end_time == end_ns assert span.attributes["tipo_evento"] == "envio msg" assert span.attributes["message_id"] == "message-1" assert span.attributes["latencia_total_STT_TTS"] == 1234 assert span.attributes["erro_msg"] == "Falha TTS" assert span.attributes["erro_detalhe"] == "" assert span.attributes["http_cod_status"] == "" assert span.attributes["http_cod_desc"] == "" assert span.attributes["finalizacao"] == "" assert "original_message_id" not in span.attributes assert span.status.status_code.name == "ERROR" def test_publish_structured_span_keeps_same_trace_id_and_lets_sdk_generate_span_id(monkeypatch) -> None: exporter = _install_memory_tracer(monkeypatch) trace_id = int("550e8400e29b41d4a716446655440101", 16) base_event = { "dat_hora_inicio": logging_utils.format_event_timestamp(1_700_000_000_000_000_000), "session_id": "550e8400-e29b-41d4-a716-446655440101", } structured_otlp.publish_structured_span({**base_event, "tipo_evento": "recebimento msg"}) structured_otlp.publish_structured_span({**base_event, "tipo_evento": "envio msg"}) spans = exporter.get_finished_spans() assert len(spans) == 2 assert {span.context.trace_id for span in spans} == {trace_id} assert spans[0].context.span_id != spans[1].context.span_id assert spans[0].parent is None assert spans[1].parent is None def test_publish_structured_span_omits_empty_tffb_attribute(monkeypatch) -> None: exporter = _install_memory_tracer(monkeypatch) structured_otlp.publish_structured_span( { "tipo_evento": "envio msg", "dat_hora_inicio": logging_utils.format_event_timestamp(1_700_000_000_000_000_000), "session_id": "550e8400-e29b-41d4-a716-446655440101", "latencia_TFFB_STT_TTS": "", } ) spans = exporter.get_finished_spans() assert len(spans) == 1 assert "latencia_TFFB_STT_TTS" not in spans[0].attributes def test_publish_structured_span_skips_invalid_session_id(monkeypatch) -> None: exporter = _install_memory_tracer(monkeypatch) structured_otlp.publish_structured_span( { "tipo_evento": "recebimento msg", "dat_hora_inicio": logging_utils.format_event_timestamp(1_700_000_000_000_000_000), "session_id": "sess-001", } ) assert exporter.get_finished_spans() == () def test_publish_structured_span_skips_without_otlp_endpoint(monkeypatch) -> None: monkeypatch.delenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", raising=False) monkeypatch.setattr(structured_otlp, "_TRACER", None) monkeypatch.setattr(structured_otlp, "_TRACER_ENDPOINT", "") structured_otlp.publish_structured_span( { "tipo_evento": "recebimento msg", "dat_hora_inicio": logging_utils.format_event_timestamp(1_700_000_000_000_000_000), "session_id": "550e8400-e29b-41d4-a716-446655440101", } )