EmbeddingIndexManager should be initialized if seed index exists.
(tmp_path)
| 135 | |
| 136 | |
| 137 | def test_index_manager_initialized(tmp_path): |
| 138 | """EmbeddingIndexManager should be initialized if seed index exists.""" |
| 139 | _reset_lifecycle() |
| 140 | # Create a minimal seed index |
| 141 | import numpy as np |
| 142 | import json |
| 143 | splits = tmp_path / "v2_splits" |
| 144 | splits.mkdir() |
| 145 | embs = np.random.randn(5, 384).astype(np.float32) |
| 146 | embs = embs / np.linalg.norm(embs, axis=1, keepdims=True) |
| 147 | np.save(splits / "seed_embeddings.npy", embs) |
| 148 | with open(splits / "seed_labels.json", "w") as f: |
| 149 | json.dump([0, 1, 2, 3, 0], f) |
| 150 | |
| 151 | lc.on_startup(tmp_path) |
| 152 | assert lc._index_manager is not None |
| 153 | assert lc._index_manager.size == 5 |
| 154 | |
| 155 | # Test growth via on_route_complete with high confidence + unanimous signals |
| 156 | new_vec = np.random.randn(384).astype(np.float32) |
| 157 | new_vec = new_vec / np.linalg.norm(new_vec) |
| 158 | lc.on_route_complete( |
| 159 | request_id="growth_test", tier_id=1, model="test", method="direct", |
| 160 | confidence=0.9, # >= 0.7 threshold |
| 161 | signal_a_tier=1, signal_a_conf=0.9, |
| 162 | signal_b_tier=1, signal_b_conf=0.7, |
| 163 | signal_c_tier=1, signal_c_conf=0.9, # signal_a == signal_c → unanimous |
| 164 | query_embedding=new_vec, |
| 165 | ) |
| 166 | assert lc._index_manager.size == 6 |
| 167 | |
| 168 | # Test save on shutdown |
| 169 | lc.on_shutdown() |
| 170 | assert (splits / "seed_embeddings.npy").exists() |
| 171 | _reset_lifecycle() |
| 172 | |
| 173 | |
| 174 | def test_idempotent_startup(tmp_path): |
nothing calls this directly
no test coverage detected