5ea9a4e879
Phase 1: in-memory ConnectionManager viewport fan-out, 500ms map debounce, cachetools TTLCache, background masscan/ffmpeg, compose memory caps. Phase 2: PostGIS geofences + ST_Intersects alerts, Timescale 1-min CAGGs and timestamp playback, FIRMS/WFIGS x firefighting ADS-B within 20 miles. No Redis/Kafka/Celery.
104 lines
2.8 KiB
Python
104 lines
2.8 KiB
Python
"""FIRMS / RSS in-process TTLCache — no Redis."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
from cachetools import TTLCache
|
|
|
|
from fire_sources import ingest_fires
|
|
from sources import ingest_rss_feed
|
|
from test_fire_sources import SAMPLE_CSV, _async_return
|
|
from upstream_cache import firms_cache, rss_cache
|
|
|
|
|
|
def test_firms_and_rss_caches_are_ttlcache():
|
|
assert isinstance(firms_cache, TTLCache)
|
|
assert firms_cache.maxsize == 100
|
|
assert firms_cache.ttl == 300
|
|
assert isinstance(rss_cache, TTLCache)
|
|
assert rss_cache.maxsize == 100
|
|
assert 60 <= rss_cache.ttl <= 300
|
|
|
|
|
|
def test_ingest_fires_hits_http_once_within_ttl(monkeypatch):
|
|
firms_cache.clear()
|
|
monkeypatch.setenv("FIRMS_MAP_KEY", "k" * 32)
|
|
monkeypatch.setenv("FIRMS_DATASETS", "VIIRS_NOAA20_NRT")
|
|
# fire_sources already imported FIRMS_DATASETS — patch the module attr
|
|
monkeypatch.setattr("fire_sources.FIRMS_DATASETS", ["VIIRS_NOAA20_NRT"])
|
|
|
|
hits = {"n": 0}
|
|
|
|
class FakeResp:
|
|
text = SAMPLE_CSV
|
|
|
|
def raise_for_status(self):
|
|
pass
|
|
|
|
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):
|
|
hits["n"] += 1
|
|
return FakeResp()
|
|
|
|
monkeypatch.setattr("fire_sources.httpx.AsyncClient", FakeClient)
|
|
|
|
async def fake_publish(points):
|
|
return len(points)
|
|
|
|
monkeypatch.setattr("fire_sources.publish_fire_batch", fake_publish)
|
|
|
|
assert asyncio.run(ingest_fires()) == 5
|
|
assert asyncio.run(ingest_fires()) == 5
|
|
assert hits["n"] == 1
|
|
|
|
|
|
def test_ingest_rss_hits_http_once_within_ttl(monkeypatch):
|
|
rss_cache.clear()
|
|
hits = {"n": 0}
|
|
|
|
class FakeResp:
|
|
text = """<?xml version="1.0"?>
|
|
<rss version="2.0"><channel><title>t</title>
|
|
<item><title>hello</title><link>http://x.example/1</link>
|
|
<description>body</description></item></channel></rss>"""
|
|
|
|
def raise_for_status(self):
|
|
pass
|
|
|
|
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):
|
|
hits["n"] += 1
|
|
return FakeResp()
|
|
|
|
monkeypatch.setattr("sources.httpx.AsyncClient", FakeClient)
|
|
|
|
published = []
|
|
|
|
async def fake_publish(subject, event):
|
|
published.append((subject, event))
|
|
|
|
monkeypatch.setattr("sources.publish_event", fake_publish)
|
|
|
|
assert asyncio.run(ingest_rss_feed("http://feeds.example/rss")) == 1
|
|
assert asyncio.run(ingest_rss_feed("http://feeds.example/rss")) == 1
|
|
assert hits["n"] == 1
|
|
assert published[0][0] == "events.rss"
|