osint-dashboard/tests/test_new_sources.py

"""NASA EONET + CISA KEV parsers (no network)."""

from __future__ import annotations


def test_parse_eonet_keeps_stable_ids_and_points():
    from sources import parse_eonet_events

    payload = {
        "events": [
            {
                "id": "EONET_6363",
                "title": "Etna Volcano",
                "categories": [{"id": "volcanoes", "title": "Volcanoes"}],
                "geometry": [
                    {"date": "2024-01-01T00:00:00Z", "type": "Point", "coordinates": [15.0, 37.7]},
                ],
                "link": "https://eonet.gsfc.nasa.gov/api/v3/events/EONET_6363",
            },
            {
                "id": "EONET_skip",
                "title": "No geometry",
                "categories": [],
                "geometry": [],
                "link": "https://eonet.gsfc.nasa.gov/api/v3/events/EONET_skip",
            },
        ]
    }
    events = parse_eonet_events(payload)
    assert len(events) == 1
    ev = events[0]
    assert ev["url"] == "https://eonet.gsfc.nasa.gov/api/v3/events/EONET_6363"
    assert ev["source_type"] == "disaster"
    assert ev["location_lat"] == 37.7
    assert ev["location_lon"] == 15.0
    assert ev["raw"]["eonet_id"] == "EONET_6363"
    assert "volcanoes" in ev["tags"]


def test_parse_cisa_kev_emits_cve_url_no_coords():
    from sources import parse_cisa_kev

    payload = {
        "vulnerabilities": [
            {
                "cveID": "CVE-2024-1234",
                "vendorProject": "Acme",
                "product": "Widget",
                "vulnerabilityName": "RCE",
                "dateAdded": "2024-06-01",
                "shortDescription": "Remote code execution",
                "requiredAction": "Apply updates",
                "dueDate": "2024-06-22",
                "knownRansomwareCampaignUse": "Known",
            }
        ]
    }
    events = parse_cisa_kev(payload)
    assert len(events) == 1
    ev = events[0]
    assert ev["url"] == "https://nvd.nist.gov/vuln/detail/CVE-2024-1234"
    assert ev["location_lat"] is None
    assert ev["location_lon"] is None
    assert "cisa-kev" in ev["tags"]
    assert "CVE-2024-1234" in ev["tags"]
    assert ev["raw"]["cveID"] == "CVE-2024-1234"


def test_ingest_cisa_kev_does_not_republish_known_nist_urls(monkeypatch):
    """Producer must not push the whole KEV catalog to NATS every cycle."""
    import asyncio

    from sources import ingest_cisa_kev

    payload = {
        "vulnerabilities": [
            {
                "cveID": "CVE-2024-1111",
                "vulnerabilityName": "old",
                "dateAdded": "2024-01-01",
                "shortDescription": "already in db",
            },
            {
                "cveID": "CVE-2024-2222",
                "vulnerabilityName": "new",
                "dateAdded": "2024-06-01",
                "shortDescription": "not in db yet",
            },
        ]
    }

    class FakeResp:
        def raise_for_status(self):
            pass

        def json(self):
            return payload

    class FakeClient:
        def __init__(self, **kw):
            pass

        async def __aenter__(self):
            return self

        async def __aexit__(self, *exc):
            return False

        async def get(self, url):
            return FakeResp()

    published: list[str] = []

    async def fake_publish(subject, event):
        published.append(event["url"])

    known = {"https://nvd.nist.gov/vuln/detail/CVE-2024-1111"}

    async def fake_existing(urls):
        return {u for u in urls if u in known}

    monkeypatch.setattr("sources.httpx.AsyncClient", FakeClient)
    monkeypatch.setattr("sources.publish_event", fake_publish)
    monkeypatch.setattr("sources.existing_event_urls", fake_existing, raising=False)

    n = asyncio.run(ingest_cisa_kev())
    assert n == 1
    assert published == ["https://nvd.nist.gov/vuln/detail/CVE-2024-2222"]