50 lines
1.8 KiB
Python
50 lines
1.8 KiB
Python
import datetime
|
|
|
|
from sqlalchemy import create_engine, text
|
|
|
|
from src.postgres_history import PostgresHistory
|
|
|
|
|
|
def test_history_returns_station_series_in_chronological_order(tmp_path):
|
|
engine = create_engine(f"sqlite:///{tmp_path / 'history.db'}")
|
|
with engine.begin() as connection:
|
|
connection.execute(text("CREATE TABLE stations (id INTEGER PRIMARY KEY, station_code TEXT)"))
|
|
connection.execute(
|
|
text(
|
|
"CREATE TABLE water_measurements ("
|
|
"timestamp DATETIME, station_id INTEGER, water_level REAL, "
|
|
"discharge REAL, discharge_percent REAL)"
|
|
)
|
|
)
|
|
connection.execute(text("INSERT INTO stations VALUES (1, 'P.1'), (2, 'P.20')"))
|
|
connection.execute(
|
|
text(
|
|
"INSERT INTO water_measurements VALUES "
|
|
"('2026-08-09 13:00:00', 1, 3.2, 110.0, 40.0),"
|
|
"('2026-08-09 14:00:00', 1, 3.4, 120.0, 42.0),"
|
|
"('2026-08-09 14:00:00', 2, 2.1, 30.0, 15.0)"
|
|
)
|
|
)
|
|
|
|
history = PostgresHistory(engine=engine).station_history(
|
|
"P.1",
|
|
start=datetime.datetime(2026, 8, 9, 12),
|
|
end=datetime.datetime(2026, 8, 9, 15),
|
|
limit=100,
|
|
)
|
|
|
|
assert [row["timestamp"].hour for row in history] == [13, 14]
|
|
assert [row["discharge"] for row in history] == [110.0, 120.0]
|
|
assert all(row["station_code"] == "P.1" for row in history)
|
|
|
|
|
|
def test_history_rejects_excessive_limit():
|
|
history = PostgresHistory.__new__(PostgresHistory)
|
|
|
|
try:
|
|
history.station_history("P.1", datetime.datetime.now(), datetime.datetime.now(), 100001)
|
|
except ValueError as error:
|
|
assert "limit" in str(error)
|
|
else:
|
|
raise AssertionError("Expected excessive history limit to be rejected")
|