| 197 | async def test_custom_resolver_model_used(self) -> None: |
| 198 | captured: dict[str, object] = {} |
| 199 | |
| 200 | def _capture_agent(*_a: object, **kwargs: object) -> object: |
| 201 | captured.update(kwargs) |
| 202 | return MagicMock() |
| 203 | |
| 204 | resolved = ResolvedFlowConfig[NewsWindow, NewsCollectionCfg]( |
| 205 | input_obj=_news_window(), |
| 206 | cfg_obj=NewsCollectionCfg(), |
| 207 | ) |
| 208 | fake_result = MagicMock() |
| 209 | fake_result.final_output = resolved |
| 210 | with ( |
| 211 | patch("quantmind.magic.Agent", side_effect=_capture_agent), |
| 212 | patch( |
| 213 | "quantmind.magic.Runner.run", |
| 214 | new=AsyncMock(return_value=fake_result), |
| 215 | ), |
| 216 | ): |
| 217 | await resolve_magic_input( |
| 218 | "x", |
| 219 | target_flow=collect_news, |
| 220 | resolver_model="custom-resolver-model", |
| 221 | ) |
| 222 | self.assertEqual(captured["model"], "custom-resolver-model") |
| 223 | |
| 224 | |
| 225 | class PreviewResolveTests(unittest.IsolatedAsyncioTestCase): |
nothing calls this directly
no outgoing calls
no test coverage detected