Add unit tests for RID API response parsing

Cover the previously-untested, highest-risk parsing in
fetch_water_data_for_date by mocking the HTTP call:
- hour 1-23 map to the same day; hour 24 rolls to next-day midnight
- qvalues "***" / None yield discharge None (no crash)
- None water level is skipped
- out-of-range (0, 25) and empty hourlytime rows are skipped
- missing "rows" key returns an empty list

This gives the scraper a safety net before its module is split.
This commit is contained in:
2026-07-22 14:24:28 +07:00
parent ad7f7b8c76
commit 08dc536c93
+125
View File
@@ -0,0 +1,125 @@
"""Assert-based tests for the RID API response parsing.
The parsing in ``fetch_water_data_for_date`` is the riskiest, previously
untested code: it maps the API's 1..24 "hourlytime" onto real timestamps
(hour 24 rolls to next-day midnight) and treats ``"***"``/``None`` discharge as
missing. These tests mock the HTTP call so no network is touched and stub the
validator so we assert on the parser's output directly.
"""
import datetime
from unittest.mock import MagicMock
import pytest
import src.water_scraper_v3 as scraper_mod
from src.water_scraper_v3 import EnhancedWaterMonitorScraper as Scraper
TARGET = datetime.datetime(2026, 7, 22)
@pytest.fixture
def make_scraper(monkeypatch):
"""Return a factory that builds a bare scraper returning the given API rows."""
def _factory(rows):
scraper = Scraper.__new__(Scraper) # bypass __init__ (no DB/network)
scraper.api_url = "https://example.invalid/api"
scraper.rate_limiter = MagicMock()
scraper.request_tracker = MagicMock()
scraper.station_config_path = "/nonexistent/stations.json"
scraper.station_mapping = scraper._load_station_mapping() # bundled defaults
response = MagicMock()
response.json.return_value = {"rows": rows}
response.raise_for_status.return_value = None
scraper.session = MagicMock()
scraper.session.post.return_value = response
# Isolate parsing from validation.
monkeypatch.setattr(
scraper_mod.DataValidator,
"validate_measurements",
staticmethod(lambda m: m),
)
return scraper
return _factory
def test_parses_water_level_and_discharge(make_scraper):
rows = [
{
"hourlytime": "9.00",
"wlvalues1": "3.50",
"qvalues1": "120.5",
"QPercent1": "45.2",
}
]
data = make_scraper(rows).fetch_water_data_for_date(TARGET)
p20 = [d for d in data if d["station_code"] == "P.20"]
assert len(p20) == 1
m = p20[0]
assert m["water_level"] == 3.5
assert m["discharge"] == 120.5
assert m["discharge_percent"] == 45.2
assert m["timestamp"] == datetime.datetime(2026, 7, 22, 9, 0)
assert m["station_name_en"] == "Ban Chiang Dao"
def test_discharge_asterisks_becomes_none(make_scraper):
rows = [{"hourlytime": "10.00", "wlvalues8": "4.20", "qvalues8": "***"}]
data = make_scraper(rows).fetch_water_data_for_date(TARGET)
p1 = [d for d in data if d["station_code"] == "P.1"][0]
assert p1["water_level"] == 4.2
assert p1["discharge"] is None
assert p1["discharge_percent"] is None
def test_hour_24_rolls_to_next_day_midnight(make_scraper):
rows = [{"hourlytime": "24.00", "wlvalues1": "3.00"}]
data = make_scraper(rows).fetch_water_data_for_date(TARGET)
assert data[0]["timestamp"] == datetime.datetime(2026, 7, 23, 0, 0)
def test_hours_1_to_23_stay_same_day(make_scraper):
rows = [
{"hourlytime": "1.00", "wlvalues1": "3.00"},
{"hourlytime": "23.00", "wlvalues1": "3.10"},
]
data = make_scraper(rows).fetch_water_data_for_date(TARGET)
times = sorted(d["timestamp"] for d in data)
assert times == [
datetime.datetime(2026, 7, 22, 1, 0),
datetime.datetime(2026, 7, 22, 23, 0),
]
def test_none_water_level_is_skipped(make_scraper):
rows = [{"hourlytime": "9.00", "wlvalues1": None, "wlvalues2": "2.5"}]
data = make_scraper(rows).fetch_water_data_for_date(TARGET)
codes = {d["station_code"] for d in data}
assert "P.20" not in codes # station 1 skipped (None water level)
assert "P.75" in codes # station 2 present
def test_out_of_range_and_empty_hours_skipped(make_scraper):
rows = [
{"hourlytime": "25.00", "wlvalues1": "3.0"},
{"hourlytime": "0.00", "wlvalues1": "3.0"},
{"hourlytime": "", "wlvalues1": "3.0"},
]
data = make_scraper(rows).fetch_water_data_for_date(TARGET)
assert data == []
def test_missing_rows_key_returns_empty(make_scraper):
scraper = make_scraper([])
scraper.session.post.return_value.json.return_value = {"unexpected": True}
assert scraper.fetch_water_data_for_date(TARGET) == []