File size: 6,715 Bytes
2e818da
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
"""Status-vocabulary and aggregation tests for the Control Room observability
projection (`app.services.observability_summary.observability_summary`).

Uses real `bootstrap.initialize_observability(...)` + real
`LocalObservationStore` instances against a `tmp_path` artifact root -- the
same isolation pattern `tests/observability/test_rag_spans.py` and
`test_operation.py` already use -- rather than mocking the observability
layer.
"""

from __future__ import annotations

import pytest

from app.observability import bootstrap
from app.observability.config import ObservabilityConfig
from app.observability.operation import observe_operation
from app.services.observability_summary import observability_summary


@pytest.fixture(autouse=True)
def _shutdown_after():
    bootstrap.reset_suppressed_failures()
    yield
    bootstrap.shutdown_observability(timeout_seconds=2.0)


def _config(tmp_path, mode: str) -> ObservabilityConfig:
    return ObservabilityConfig(
        enabled=mode != "disabled",
        mode=mode,  # type: ignore[arg-type]
        otlp_endpoint="http://127.0.0.1:4318",
        signoz_ui_url="http://127.0.0.1:8080",
        artifact_root=tmp_path,
    )


# --- not_instrumented ---------------------------------------------------


def test_not_instrumented_when_mode_disabled(tmp_path):
    bootstrap.initialize_observability(config=_config(tmp_path, "disabled"))

    summary = observability_summary()

    assert summary["status"] == "not_instrumented"
    assert summary["latest_trace_id"] is None
    assert summary["pipeline_version"] is None
    assert summary["stages"] == {}
    assert summary["retrieval"] == {"success_count": 0, "empty_count": 0, "error_count": 0}
    assert summary["exporter"] == {"mode": "disabled", "suppressed_failures": 0}
    assert summary["signoz_url"] is None


# --- not_observed ---------------------------------------------------


def test_not_observed_when_store_empty(tmp_path):
    bootstrap.initialize_observability(config=_config(tmp_path, "local"))

    summary = observability_summary()

    assert summary["status"] == "not_observed"
    assert summary["latest_trace_id"] is None
    assert summary["stages"] == {}
    assert summary["retrieval"] == {"success_count": 0, "empty_count": 0, "error_count": 0}
    assert summary["exporter"]["mode"] == "local"


def test_not_observed_takes_priority_over_signoz_url_in_full_mode(tmp_path):
    bootstrap.initialize_observability(config=_config(tmp_path, "full"))

    summary = observability_summary()

    assert summary["status"] == "not_observed"
    # full mode always surfaces the SigNoz link, even with nothing observed yet.
    assert summary["signoz_url"] == "http://127.0.0.1:8080"


# --- degraded ---------------------------------------------------


def test_degraded_when_suppressed_failures_recorded_even_with_data(tmp_path):
    bootstrap.initialize_observability(config=_config(tmp_path, "local"))
    with observe_operation("rag.retrieval"):
        pass
    bootstrap.note_suppressed_failure()

    summary = observability_summary()

    assert summary["status"] == "degraded"
    assert summary["exporter"]["suppressed_failures"] >= 1


# --- observed ---------------------------------------------------


def test_observed_with_stage_percentiles_and_retrieval_counts(tmp_path):
    from app.observability.contracts import RetrievalOutcome

    bootstrap.initialize_observability(config=_config(tmp_path, "local"))

    # Three vector_search rows with distinct durations to exercise p50/p95.
    for _ in range(3):
        with observe_operation("chroma.vector_search"):
            pass

    with observe_operation("rag.retrieval") as op:
        op.set_retrieval(RetrievalOutcome.success([{"text": "a"}]))
    with observe_operation("rag.retrieval") as op:
        op.set_retrieval(RetrievalOutcome.success([]))
    with observe_operation("rag.retrieval") as op:
        op.set_retrieval(RetrievalOutcome.failure("vector_search_failed"))
        op.mark_error("vector_search_failed")

    summary = observability_summary()

    assert summary["status"] == "observed"
    assert summary["latest_trace_id"] is not None
    assert "vector_search_ms" in summary["stages"]
    assert summary["stages"]["vector_search_ms"]["count"] == 3
    assert summary["stages"]["vector_search_ms"]["p50"] >= 0
    assert summary["stages"]["vector_search_ms"]["p95"] >= summary["stages"]["vector_search_ms"]["p50"]
    assert "retrieval_ms" in summary["stages"]
    assert summary["stages"]["retrieval_ms"]["count"] == 3
    assert summary["retrieval"] == {"success_count": 1, "empty_count": 1, "error_count": 1}
    assert summary["exporter"] == {"mode": "local", "suppressed_failures": 0}


def test_missing_duration_never_counted_as_zero(tmp_path):
    """A row recorded via a store that never got a real duration (e.g. an
    injected store bypassing the timing boundary) must not pollute p50/p95.

    Exercised here by directly appending a row through the store with
    `duration_ms=None`, the way a degraded/failed timing path could produce
    one, and confirming it contributes zero to the stage's count.
    """
    from app.observability.contracts import OperationObservation
    from app.observability.local_store import LocalObservationStore

    bootstrap.initialize_observability(config=_config(tmp_path, "local"))
    store = bootstrap.get_state().local_store
    assert isinstance(store, LocalObservationStore)
    store.record(
        OperationObservation(operation="chroma.vector_search", status="success", duration_ms=None)
    )

    summary = observability_summary()

    assert summary["status"] == "observed"
    assert summary["stages"] == {}


def test_pipeline_version_taken_from_latest_experiment(tmp_path):
    from app.observability.contracts import ExperimentIdentity

    bootstrap.initialize_observability(config=_config(tmp_path, "local"))
    with observe_operation("rag.retrieval") as op:
        op.set_experiment(ExperimentIdentity(pipeline_version="rag-naive-v1"))
    with observe_operation("rag.retrieval"):
        pass  # no experiment identity -- ordinary traffic

    summary = observability_summary()

    # Most recent row wins; the plain ordinary-traffic row carries no
    # experiment identity, so the last *tagged* row's version is used.
    assert summary["pipeline_version"] == "rag-naive-v1"


def test_signoz_url_only_present_in_full_mode(tmp_path):
    bootstrap.initialize_observability(config=_config(tmp_path, "full"))
    with observe_operation("rag.retrieval"):
        pass

    summary = observability_summary()

    assert summary["signoz_url"] == "http://127.0.0.1:8080"
    assert summary["exporter"]["mode"] == "full"