Code / services/cj_assessment_service/tests/unit/test_bt_se_computation.py
services/cj_assessment_service/tests/unit/test_bt_se_computation.py
1
"""Tests for raw Bradley-Terry Fisher information state.
2
3
Purpose:
4
Protect the ADR-0046 split between raw optimizer-facing Fisher state and
5
capped standard-error reporting. The analytical BT information boundary is
6
intentionally exercised only through id-typed comparison outcomes.
7
"""
8
9
from __future__ import annotations
10
11
import importlib
12
import math
13
import sys
14
15
import numpy as np
16
import pytest
17
18
from services.cj_assessment_service.cj_core_logic import bt_inference
19
from services.cj_assessment_service.cj_core_logic.bt_inference import (
20
BTInformationFailureReason,
21
ComparisonOutcome,
22
EssayId,
23
)
24
25
26
def _essay_id(value: str) -> EssayId:
27
return EssayId(value)
28
29
30
def test_bt_inference_imports_without_service_infrastructure(
31
monkeypatch: pytest.MonkeyPatch,
32
) -> None:
33
"""BT Fisher primitives retain their import boundary without service packages."""
34
35
module_name = "services.cj_assessment_service.cj_core_logic.bt_inference"
36
monkeypatch.delitem(sys.modules, module_name, raising=False)
37
monkeypatch.setitem(sys.modules, "huleedu_service_libs", None)
38
monkeypatch.setitem(sys.modules, "structlog", None)
39
40
imported_module = importlib.import_module(module_name)
41
42
result = imported_module.compute_raw_bt_information_state(
43
ordered_ids=(EssayId("a"), EssayId("reference")),
44
pairs=((EssayId("a"), EssayId("reference")),),
45
theta_by_id={EssayId("a"): 0.0, EssayId("reference"): 0.0},
46
)
47
48
assert result.is_ok
49
assert result.value.reference_id == EssayId("reference")
50
51
52
class TestRawBTInformationState:
53
"""Tests for id-typed raw Fisher information-state construction."""
54
55
def test_reference_handling_preserves_last_anchor_convention(self) -> None:
56
"""The last anchor is the reference and excluded from active coordinates."""
57
ids = tuple(
58
_essay_id(value) for value in ("student-1", "student-2", "anchor-a", "anchor-b")
59
)
60
pairs: tuple[ComparisonOutcome, ...] = (
61
(ids[0], ids[1]),
62
(ids[1], ids[2]),
63
(ids[2], ids[3]),
64
)
65
theta_by_id = {essay_id: float(idx) for idx, essay_id in enumerate(ids)}
66
67
result = bt_inference.compute_raw_bt_information_state(
68
ordered_ids=ids,
69
pairs=pairs,
70
theta_by_id=theta_by_id,
71
anchor_ids=(ids[2], ids[3]),
72
)
73
74
assert result.is_ok
75
state = result.value
76
assert state.reference_id == ids[3]
77
assert ids[3] not in state.active_ids
78
assert ids[3] not in state.active_index_by_id
79
assert state.raw_standard_error_by_id[ids[3]] == 0.0
80
assert state.active_ids == (ids[0], ids[1], ids[2])
81
82
def test_reference_star_graph_is_connected_before_reference_removal(self) -> None:
83
"""Active essays connected through the reference produce valid covariance."""
84
ids = tuple(_essay_id(value) for value in ("a", "b", "reference"))
85
pairs: tuple[ComparisonOutcome, ...] = ((ids[0], ids[2]), (ids[1], ids[2]))
86
theta_by_id = {essay_id: 0.0 for essay_id in ids}
87
88
result = bt_inference.compute_raw_bt_information_state(
89
ordered_ids=ids,
90
pairs=pairs,
91
theta_by_id=theta_by_id,
92
anchor_ids=(ids[2],),
93
)
94
95
assert result.is_ok
96
state = result.value
97
assert state.reference_id == ids[2]
98
assert np.all(np.isfinite(state.covariance_active))
99
assert np.all(np.isfinite(tuple(state.raw_standard_error_by_id.values())))
100
101
def test_ridge_covariance_applied_exactly_once(self) -> None:
102
"""A zero-information connected graph is invertible after one ridge."""
103
ids = tuple(_essay_id(value) for value in ("a", "b", "reference"))
104
pairs: tuple[ComparisonOutcome, ...] = ((ids[0], ids[1]), (ids[1], ids[2]))
105
theta_by_id = {ids[0]: 20.0, ids[1]: 0.0, ids[2]: -20.0}
106
107
result = bt_inference.compute_raw_bt_information_state(
108
ordered_ids=ids,
109
pairs=pairs,
110
theta_by_id=theta_by_id,
111
ridge=bt_inference.FISHER_RIDGE,
112
)
113
114
assert result.is_ok
115
state = result.value
116
expected_regularized = (
117
state.observed_fisher_active_unregularized
118
+ np.eye(len(state.active_ids), dtype=np.float64) * bt_inference.FISHER_RIDGE
119
)
120
np.testing.assert_allclose(state.observed_fisher_active_regularized, expected_regularized)
121
assert np.all(np.isfinite(np.diag(state.covariance_active)))
122
assert np.all(np.diag(state.covariance_active) >= 0.0)
123
124
def test_disconnected_graph_fails_closed_without_matrices(self) -> None:
125
"""Two active comparison islands return a typed disconnected failure."""
126
ids = tuple(_essay_id(value) for value in ("a", "b", "c", "reference"))
127
pairs: tuple[ComparisonOutcome, ...] = ((ids[0], ids[1]), (ids[2], ids[3]))
128
theta_by_id = {essay_id: 0.0 for essay_id in ids}
129
130
result = bt_inference.compute_raw_bt_information_state(
131
ordered_ids=ids,
132
pairs=pairs,
133
theta_by_id=theta_by_id,
134
)
135
136
assert result.is_err
137
assert result.error.failure_reason is BTInformationFailureReason.DISCONNECTED_GRAPH
138
139
def test_condition_number_is_reported_but_not_gated(self) -> None:
140
"""The builder reports high condition numbers without rejecting them."""
141
ids = tuple(_essay_id(value) for value in ("a", "b", "reference"))
142
pairs: tuple[ComparisonOutcome, ...] = ((ids[0], ids[1]), (ids[1], ids[2]))
143
theta_by_id = {ids[0]: 20.0, ids[1]: 0.0, ids[2]: -20.0}
144
145
result = bt_inference.compute_raw_bt_information_state(
146
ordered_ids=ids,
147
pairs=pairs,
148
theta_by_id=theta_by_id,
149
ridge=bt_inference.FISHER_RIDGE,
150
)
151
152
assert result.is_ok
153
state = result.value
154
assert state.condition_number_regularized == pytest.approx(
155
float(np.linalg.cond(state.observed_fisher_active_regularized))
156
)
157
158
def test_capped_and_raw_standard_errors_are_separate(self) -> None:
159
"""Sparse connected raw SE can exceed the reporting cap."""
160
ids = tuple(_essay_id(value) for value in ("a", "b", "reference"))
161
pairs: tuple[ComparisonOutcome, ...] = ((ids[0], ids[1]), (ids[1], ids[2]))
162
theta_by_id = {ids[0]: 20.0, ids[1]: 0.0, ids[2]: -20.0}
163
164
raw_result = bt_inference.compute_raw_bt_information_state(
165
ordered_ids=ids,
166
pairs=pairs,
167
theta_by_id=theta_by_id,
168
)
169
capped = bt_inference.report_capped_bt_standard_errors(
170
ordered_ids=ids,
171
pairs=pairs,
172
theta_by_id=theta_by_id,
173
)
174
175
assert raw_result.is_ok
176
raw_state = raw_result.value
177
assert raw_state.raw_standard_error_by_id[ids[0]] > bt_inference.BT_STANDARD_ERROR_MAX
178
assert capped[0] == bt_inference.BT_STANDARD_ERROR_MAX
179
180
def test_raw_standard_error_identity_matches_covariance_diagonal(self) -> None:
181
"""Active raw SEs are sqrt of covariance diagonal; reference is zero."""
182
ids = tuple(_essay_id(value) for value in ("a", "b", "reference"))
183
pairs: tuple[ComparisonOutcome, ...] = ((ids[0], ids[1]), (ids[1], ids[2]))
184
theta_by_id = {essay_id: 0.0 for essay_id in ids}
185
186
result = bt_inference.compute_raw_bt_information_state(
187
ordered_ids=ids,
188
pairs=pairs,
189
theta_by_id=theta_by_id,
190
)
191
192
assert result.is_ok
193
state = result.value
194
for essay_id in state.active_ids:
195
idx = state.active_index_by_id[essay_id]
196
expected = math.sqrt(max(float(state.covariance_active[idx, idx]), 0.0))
197
assert state.raw_standard_error_by_id[essay_id] == pytest.approx(expected)
198
assert state.raw_standard_error_by_id[state.reference_id] == 0.0
199
200
def test_non_finite_theta_fails_closed(self) -> None:
201
"""NaN theta values return the typed non-finite theta failure."""
202
ids = tuple(_essay_id(value) for value in ("a", "b", "reference"))
203
pairs: tuple[ComparisonOutcome, ...] = ((ids[0], ids[1]), (ids[1], ids[2]))
204
theta_by_id = {ids[0]: 0.0, ids[1]: math.nan, ids[2]: 0.0}
205
206
result = bt_inference.compute_raw_bt_information_state(
207
ordered_ids=ids,
208
pairs=pairs,
209
theta_by_id=theta_by_id,
210
)
211
212
assert result.is_err
213
assert result.error.failure_reason is BTInformationFailureReason.NON_FINITE_THETA
214
215
216
class TestReportCappedBTSE:
217
"""Tests for the capped reporting adapter."""
218
219
def test_cap_math_passes_through_and_clips(self) -> None:
220
"""Values below the cap pass through; values above it clip exactly."""
221
ids = tuple(_essay_id(value) for value in ("a", "b", "reference"))
222
pairs: tuple[ComparisonOutcome, ...] = ((ids[0], ids[1]), (ids[1], ids[2]))
223
stable_theta_by_id = {essay_id: 0.0 for essay_id in ids}
224
sparse_theta_by_id = {ids[0]: 20.0, ids[1]: 0.0, ids[2]: -20.0}
225
226
passed_through = bt_inference.report_capped_bt_standard_errors(
227
ordered_ids=ids,
228
pairs=pairs,
229
theta_by_id=stable_theta_by_id,
230
ridge=1.0,
231
)
232
clipped = bt_inference.report_capped_bt_standard_errors(
233
ordered_ids=ids,
234
pairs=pairs,
235
theta_by_id=sparse_theta_by_id,
236
)
237
238
assert float(np.max(passed_through)) < bt_inference.BT_STANDARD_ERROR_MAX
239
assert clipped[0] == bt_inference.BT_STANDARD_ERROR_MAX
240
241
def test_builder_failure_returns_nan_sentinels_and_logs_reason(
242
self,
243
monkeypatch: pytest.MonkeyPatch,
244
) -> None:
245
"""Adapter failures return NaNs and log the typed failure reason."""
246
ids = tuple(_essay_id(value) for value in ("a", "b", "reference"))
247
pairs: tuple[ComparisonOutcome, ...] = ()
248
theta_by_id = {essay_id: 0.0 for essay_id in ids}
249
log_records: list[dict[str, object]] = []
250
251
def fake_warning(message: str, *, extra: dict[str, object]) -> None:
252
log_records.append({"message": message, **extra})
253
254
monkeypatch.setattr(bt_inference.logger, "warning", fake_warning)
255
256
se = bt_inference.report_capped_bt_standard_errors(
257
ordered_ids=ids,
258
pairs=pairs,
259
theta_by_id=theta_by_id,
260
)
261
262
assert np.all(np.isnan(se))
263
assert log_records
264
assert (
265
log_records[0]["failure_reason"] == BTInformationFailureReason.INVALID_COMPARISONS.value
266
)