Class ResolveMagicInputTests

tests/test_magic.py:106–196

    def test_basemodel_with_clean_schema(self) -> None:
        class Clean(BaseModel):
            x: int = 0

        out = _pydantic_schema_str(Clean)
        parsed = json.loads(out)
        # Clean schema path -> emits standard "properties" key.
        self.assertIn("properties", parsed)

    def test_annotated_union_emits_one_of(self) -> None:
        out = _pydantic_schema_str(PaperInput)
        parsed = json.loads(out)
        self.assertIn("oneOf", parsed)
        # PaperInput has 5 variants; not all need schema-rendering, but
        # the rendered list should be non-empty.
        self.assertGreater(len(parsed["oneOf"]), 0)

    def test_baseinput_subclass_directly(self) -> None:
        out = _pydantic_schema_str(ArxivIdentifier)
        parsed = json.loads(out)
        self.assertEqual(parsed["properties"]["type"]["default"], "arxiv")

    def test_fallback_for_unknown_type(self) -> None:
        out = _pydantic_schema_str(int)
        # Falls back to repr.
        self.assertEqual(out, repr(int))


class ResolveMagicInputTests(unittest.IsolatedAsyncioTestCase):
    async def test_happy_path_returns_tuple(self) -> None:
        captured: dict[str, object] = {}

        def _capture_agent(*_a: object, **kwargs: object) -> object:
            captured.update(kwargs)
            return MagicMock(name="agent")

        # Build a fake resolver result whose final_output is a populated
        # ResolvedFlowConfig.
        resolved = ResolvedFlowConfig[NewsWindow, NewsCollectionCfg](
            input_obj=_news_window(),
            cfg_obj=NewsCollectionCfg(retain_raw_html=True),
        )
        fake_result = MagicMock()
        fake_result.final_output = resolved
        with (
            patch("quantmind.magic.Agent", side_effect=_capture_agent),
            patch(
                "quantmind.magic.Runner.run",
                new=AsyncMock(return_value=fake_result),
            ),
        ):
            inp, cfg = await resolve_magic_input(
                "collect the last day of pr-newswire",
                target_flow=collect_news,
            )
        self.assertIs(inp, resolved.input_obj)
        self.assertIs(cfg, resolved.cfg_obj)
        # Resolver agent was given a name derived from the flow.
        self.assertEqual(captured["name"], "magic_resolver_collect_news")
        self.assertEqual(captured["model"], "gpt-5.6-luna")