grabowski
12b7f9f422
- tests/conftest.py: put repo root on sys.path so `import src...` resolves under pytest regardless of invocation directory. - test_matrix_formatting.py: lock in HTML formatted_body + plain-text fallback, URL linkification, HTML escaping, and send_alert field rendering. - test_station_persistence.py: cover default-load, save/reload round-trip (incl. Thai text), runtime-file precedence, and atomic-write cleanup. These are real assert-based tests (unlike the existing print-style scripts) so CI can gate on them. 13 tests, all passing.
78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
"""Assert-based tests for station-mapping persistence.
|
|
|
|
Station CRUD must survive restarts: the scraper loads its mapping from a
|
|
runtime-writable JSON file (falling back to bundled defaults) and writes it back
|
|
atomically. These tests exercise that load/save behaviour without constructing a
|
|
full scraper (which would open network/DB connections).
|
|
"""
|
|
|
|
from src.water_scraper_v3 import EnhancedWaterMonitorScraper as Scraper
|
|
|
|
|
|
def _bare_scraper(config_path):
|
|
"""A scraper instance with only the station-config attribute set.
|
|
|
|
Bypasses __init__ so no database/HTTP connection is attempted.
|
|
"""
|
|
scraper = Scraper.__new__(Scraper)
|
|
scraper.station_config_path = config_path
|
|
return scraper
|
|
|
|
|
|
def test_loads_bundled_defaults_when_runtime_file_absent(tmp_path):
|
|
scraper = _bare_scraper(str(tmp_path / "does_not_exist.json"))
|
|
mapping = scraper._load_station_mapping()
|
|
|
|
assert len(mapping) == 16
|
|
assert mapping["8"]["code"] == "P.1"
|
|
assert mapping["8"]["english_name"] == "Nawarat Bridge"
|
|
|
|
|
|
def test_save_then_reload_roundtrips_including_thai(tmp_path):
|
|
path = str(tmp_path / "stations.json")
|
|
scraper = _bare_scraper(path)
|
|
scraper.station_mapping = {
|
|
"1": {
|
|
"code": "P.99",
|
|
"thai_name": "สถานีทดสอบ",
|
|
"english_name": "Test Station",
|
|
"latitude": 1.0,
|
|
"longitude": 2.0,
|
|
"geohash": None,
|
|
}
|
|
}
|
|
|
|
assert scraper.save_stations() is True
|
|
|
|
reloaded = _bare_scraper(path)._load_station_mapping()
|
|
assert reloaded == scraper.station_mapping
|
|
assert reloaded["1"]["thai_name"] == "สถานีทดสอบ"
|
|
|
|
|
|
def test_runtime_file_takes_precedence_over_defaults(tmp_path):
|
|
path = str(tmp_path / "stations.json")
|
|
writer = _bare_scraper(path)
|
|
writer.station_mapping = {"1": {"code": "ONLY"}}
|
|
assert writer.save_stations() is True
|
|
|
|
mapping = _bare_scraper(path)._load_station_mapping()
|
|
assert list(mapping.keys()) == ["1"]
|
|
assert mapping["1"]["code"] == "ONLY"
|
|
|
|
|
|
def test_save_returns_false_without_a_path():
|
|
scraper = _bare_scraper("")
|
|
scraper.station_mapping = {}
|
|
assert scraper.save_stations() is False
|
|
|
|
|
|
def test_save_is_atomic_no_tmp_left_behind(tmp_path):
|
|
path = tmp_path / "stations.json"
|
|
scraper = _bare_scraper(str(path))
|
|
scraper.station_mapping = {"1": {"code": "P.1"}}
|
|
|
|
assert scraper.save_stations() is True
|
|
assert path.exists()
|
|
# The temp file used during the atomic write must not remain.
|
|
assert not (tmp_path / "stations.json.tmp").exists()
|