grabowski
ae5d0a13d7
Add mapped river and ThaiWater sensor layers, PostgreSQL history charts, API endpoints, and dashboard tests.
76 lines
2.6 KiB
Python
76 lines
2.6 KiB
Python
from unittest.mock import MagicMock
|
|
|
|
from src.thaiwater import ThaiWaterClient
|
|
|
|
|
|
SAMPLE_RESPONSE = {
|
|
"data": {
|
|
"50": {
|
|
"type": "FeatureCollection",
|
|
"features": [
|
|
{
|
|
"geometry": {"type": "Point", "coordinates": [98.635262, 19.638411]},
|
|
"properties": {
|
|
"id": "123",
|
|
"waterlevelDatetime": "2026-08-09T15:00:00+07:00",
|
|
"waterlevelMsl": 742.34,
|
|
"storagePercent": 38.57,
|
|
"diffWlBank": 1.87,
|
|
"riverName": "Ping River",
|
|
"station": {
|
|
"stationCode": "G07003-P.65",
|
|
"station": "Ban Muang Pok",
|
|
},
|
|
"agency": {"agencyShort": "RID"},
|
|
"basin": {"basin": "Ping"},
|
|
},
|
|
},
|
|
{
|
|
"geometry": {"type": "Point", "coordinates": [100.1, 18.1]},
|
|
"properties": {
|
|
"id": "999",
|
|
"station": {"stationCode": "N.1", "station": "Nan station"},
|
|
"basin": {"basin": "Nan"},
|
|
},
|
|
},
|
|
],
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
def test_fetch_ping_sensors_normalizes_and_filters_basin():
|
|
session = MagicMock()
|
|
response = session.get.return_value
|
|
response.json.return_value = SAMPLE_RESPONSE
|
|
response.raise_for_status.return_value = None
|
|
|
|
sensors = ThaiWaterClient(session=session, api_key="public-key").fetch_ping_sensors()
|
|
|
|
assert sensors == [
|
|
{
|
|
"id": "thaiwater:123",
|
|
"station_code": "P.65",
|
|
"station_name": "Ban Muang Pok",
|
|
"latitude": 19.638411,
|
|
"longitude": 98.635262,
|
|
"timestamp": "2026-08-09T15:00:00+07:00",
|
|
"water_level_msl": 742.34,
|
|
"bank_percent": 38.57,
|
|
"distance_to_bank": 1.87,
|
|
"river_name": "Ping River",
|
|
"agency": "RID",
|
|
"source": "ThaiWater",
|
|
}
|
|
]
|
|
session.get.assert_called_once()
|
|
assert session.get.call_args.kwargs["headers"]["x-api-key"] == "public-key"
|
|
|
|
|
|
def test_fetch_ping_sensors_skips_features_without_coordinates():
|
|
response_data = {"data": {"50": {"features": [{"geometry": None, "properties": {"basin": {"basin": "Ping"}}}]}}}
|
|
session = MagicMock()
|
|
session.get.return_value.json.return_value = response_data
|
|
|
|
assert ThaiWaterClient(session=session, api_key="key").fetch_ping_sensors() == []
|