spb/ai-atlas

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