1from aiatlas.connectors.labs.qwen import FEED, QwenConnector2from aiatlas.sdk.facts import Target3from tests.conftest import extract_from_fixture, fixture_path456async def test_feed():7 facts = await extract_from_fixture(QwenConnector(), Target(url=FEED, doc_type="feed", key="feed"), fixture_path("qwen", "blog.rss.xml"), content_type="application/xml")8 assert len(facts.events) == 449 guard = facts.events[0]10 assert guard.summary == "Qwen: Qwen3Guard: Real-time Safety for Your Token Stream" and guard.category == "release"11 assert guard.effective_at.date().isoformat() == "2025-09-22" and guard.meta["models_mentioned"] == ["Qwen3Guard"]12 assert guard.dedupe_key == "ANNOUNCEMENT:https://qwenlm.github.io/blog/qwen3guard/"13 gspo = next(e for e in facts.events if "GSPO" in e.summary)14 assert gspo.category == "company"15 follows = [t for t in facts.targets if t.doc_type == "news"]16 assert len(follows) == 30 and follows[0].needs_llm is True and follows[0].meta["llm_task"] == "release_announcement"17 assert next(t for t in follows if "gspo" in t.url).needs_llm is False18