grabowski
4e619aa2ae
CI / Format & lint (push) Successful in 19s
CI / Test suite (push) Successful in 47s
Docs / Validate documentation (push) Successful in 19s
Rust / fmt, clippy, test, build (push) Successful in 7m35s
Security / Dependency vulnerabilities (push) Successful in 1m37s
Security / Static analysis (push) Successful in 17s
Security / License report (push) Successful in 1m30s
rust/collector replaces the collection half of the Python web app: RID hourly gauges (with the late-report semantics: store everything, retry every 10 min while RID is behind, re-fetch yesterday's holes, startup gap fill), HII rain + water level, RID dams (today + yesterday) and Open-Meteo rain. Same tables and upsert semantics, one UNNEST statement per table, Bangkok wall-clock time regardless of the server time zone. Python side: COLLECTION_ENABLED=false makes the API leader stop collecting and instead re-forecast + notify when the newest RID reading in the database advances (polled every FORECAST_POLL_SECONDS) and once per clock hour; /scraping/status then serves the collector's status file; /scrape/trigger answers 409. Verified: 20 Rust tests (parsers, gap/schedule logic, database tests against PostgreSQL); live parity run of both collectors into two empty databases: all 9 tables row-for-row identical (scripts/collector_parity.py); the Python API served health, measurements, stats, status and model forecasts from a database filled only by ping-collector. 177 Python tests pass. reqwest uses native-tls: app.rid.go.th offers only TLS 1.2 CBC suites, which rustls refuses. Deploy/cutover/rollback: scripts/install_collector.sh, scripts/ping-collector.service; CI: .gitea/workflows/rust.yml (fmt, clippy, tests with a PostgreSQL service, release artifact).
97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
"""COLLECTION_ENABLED=false: the Rust ping-collector collects, this app only
|
|
re-forecasts and notifies when the database's newest RID reading advances
|
|
(and once per clock hour), and /scraping/status reports the collector's file."""
|
|
|
|
import asyncio
|
|
import datetime
|
|
import json
|
|
|
|
from src import web_api
|
|
from src.config import Config
|
|
|
|
|
|
def test_forecasts_follow_new_readings_and_the_clock_hour(monkeypatch):
|
|
clock = {"now": datetime.datetime(2026, 10, 17, 21, 30)}
|
|
|
|
class FixedClock(datetime.datetime):
|
|
@classmethod
|
|
def now(cls, tz=None):
|
|
return clock["now"]
|
|
|
|
# newest reading per poll: unchanged, unchanged, advances, next hour
|
|
latest = iter(
|
|
[
|
|
datetime.datetime(2026, 10, 17, 20),
|
|
datetime.datetime(2026, 10, 17, 20),
|
|
datetime.datetime(2026, 10, 17, 21),
|
|
datetime.datetime(2026, 10, 17, 21),
|
|
]
|
|
)
|
|
calls = {"precompute": 0, "notify": 0}
|
|
sleeps = []
|
|
|
|
async def precompute():
|
|
calls["precompute"] += 1
|
|
|
|
async def notify():
|
|
calls["notify"] += 1
|
|
|
|
async def fake_sleep(seconds):
|
|
sleeps.append(seconds)
|
|
if len(sleeps) == 3: # the last poll happens in the next clock hour
|
|
clock["now"] = datetime.datetime(2026, 10, 17, 22, 1)
|
|
if len(sleeps) == 4:
|
|
raise asyncio.CancelledError
|
|
|
|
monkeypatch.setattr(web_api, "datetime", FixedClock)
|
|
monkeypatch.setattr(web_api.asyncio, "sleep", fake_sleep)
|
|
monkeypatch.setattr(web_api, "_latest_rid_reading", lambda: next(latest))
|
|
monkeypatch.setattr(web_api, "_precompute_forecasts", precompute)
|
|
monkeypatch.setattr(web_api, "_notify_transitions", notify)
|
|
monkeypatch.setattr(Config, "FORECAST_POLL_SECONDS", 60)
|
|
|
|
asyncio.run(web_api.forecast_notify_task())
|
|
|
|
# first poll (new + hourly), third (new reading), fourth (new clock hour)
|
|
assert calls == {"precompute": 3, "notify": 3}
|
|
assert sleeps == [60, 60, 60, 60]
|
|
|
|
|
|
def test_status_comes_from_the_collector_file(tmp_path, monkeypatch):
|
|
status = tmp_path / "collector-status.json"
|
|
status.write_text(
|
|
json.dumps(
|
|
{
|
|
"is_running": False,
|
|
"total_runs": 4,
|
|
"successful_runs": 3,
|
|
"failed_runs": 1,
|
|
"last_run": "2026-10-17T22:05:01",
|
|
"next_run": "2026-10-17T23:05:00",
|
|
"feed_current": True,
|
|
"latest_reading": "2026-10-17T22:00:00",
|
|
"readings_saved": 352,
|
|
"collector": "ping-collector 0.1.0",
|
|
}
|
|
)
|
|
)
|
|
monkeypatch.setattr(Config, "COLLECTION_ENABLED", False)
|
|
monkeypatch.setattr(Config, "SCRAPING_STATUS_FILE", str(status))
|
|
monkeypatch.setitem(web_api.app_state, "leader_lock", object())
|
|
|
|
resp = asyncio.run(web_api.get_scraping_status())
|
|
|
|
assert resp.total_runs == 4
|
|
assert resp.feed_current is True
|
|
assert resp.latest_reading == datetime.datetime(2026, 10, 17, 22)
|
|
|
|
|
|
def test_manual_trigger_is_refused_when_collection_runs_elsewhere(monkeypatch):
|
|
import pytest
|
|
from fastapi import BackgroundTasks, HTTPException
|
|
|
|
monkeypatch.setattr(Config, "COLLECTION_ENABLED", False)
|
|
with pytest.raises(HTTPException) as err:
|
|
asyncio.run(web_api.trigger_scraping(BackgroundTasks()))
|
|
assert err.value.status_code == 409
|