haneulai
a716dbef96
Some checks are pending
CI / Crawler (Python ${{ matrix.python-version }}) (3.10) (push) Waiting to run
CI / Crawler (Python ${{ matrix.python-version }}) (3.11) (push) Waiting to run
CI / API (Python 3.11) (push) Waiting to run
CI / Database migration (push) Waiting to run
CI / App web build (Node 20) (push) Waiting to run
129 lines
3.9 KiB
Python
129 lines
3.9 KiB
Python
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
|
|
CRAWLER_DIR = Path(__file__).resolve().parents[1]
|
|
if str(CRAWLER_DIR) not in sys.path:
|
|
sys.path.insert(0, str(CRAWLER_DIR))
|
|
|
|
import daily_summary
|
|
|
|
|
|
class FakeRpc:
|
|
def __init__(self, db, params):
|
|
self.db = db
|
|
self.params = params
|
|
|
|
def execute(self):
|
|
self.db.saved = self.params["p_records"]
|
|
self.db.rpc_params = self.params
|
|
return SimpleNamespace(data=self.db.saved)
|
|
|
|
|
|
class FakeQuery:
|
|
def __init__(self, db, table_name):
|
|
self.db = db
|
|
self.table_name = table_name
|
|
self.filters = []
|
|
self.db.queries.append(self)
|
|
|
|
def select(self, _columns):
|
|
return self
|
|
|
|
def gte(self, column, value):
|
|
self.filters.append(("gte", column, value))
|
|
return self
|
|
|
|
def lt(self, column, value):
|
|
self.filters.append(("lt", column, value))
|
|
return self
|
|
|
|
def order(self, _column, desc=False):
|
|
return self
|
|
|
|
def upsert(self, payload, on_conflict=None):
|
|
self.payload = payload
|
|
return self
|
|
|
|
def execute(self):
|
|
if self.table_name == "plants":
|
|
return SimpleNamespace(data=[{"id": "plant-a", "capacity": 100}])
|
|
if self.table_name == "solar_logs":
|
|
return SimpleNamespace(data=[{
|
|
"plant_id": "plant-a",
|
|
"current_kw": 25,
|
|
"today_kwh": 120,
|
|
"created_at": "2026-08-05T06:00:00+00:00",
|
|
}])
|
|
if self.table_name == "daily_stats":
|
|
self.db.saved = self.payload
|
|
return SimpleNamespace(data=self.payload)
|
|
raise AssertionError(f"unexpected table: {self.table_name}")
|
|
|
|
|
|
class FakeDb:
|
|
def __init__(self):
|
|
self.queries = []
|
|
self.saved = None
|
|
self.rpc_params = None
|
|
|
|
def table(self, table_name):
|
|
return FakeQuery(self, table_name)
|
|
|
|
def rpc(self, function_name, params):
|
|
if function_name != "upsert_daily_stats":
|
|
raise AssertionError(function_name)
|
|
return FakeRpc(self, params)
|
|
|
|
|
|
class DailySummaryTimezoneTest(unittest.TestCase):
|
|
def test_daily_close_uses_kst_utc_half_open_bounds(self):
|
|
db = FakeDb()
|
|
|
|
with (
|
|
patch.object(daily_summary, "get_supabase_client", return_value=db),
|
|
patch.object(daily_summary, "get_all_plants", return_value=[]),
|
|
):
|
|
result = daily_summary.calculate_daily_stats("2026-08-05")
|
|
|
|
solar_query = next(q for q in db.queries if q.table_name == "solar_logs")
|
|
self.assertIn(("gte", "created_at", "2026-08-04T15:00:00+00:00"), solar_query.filters)
|
|
self.assertIn(("lt", "created_at", "2026-08-05T15:00:00+00:00"), solar_query.filters)
|
|
self.assertTrue(result)
|
|
self.assertEqual(120, db.saved[0]["total_generation"])
|
|
self.assertEqual("daily_summary", db.rpc_params["p_source"])
|
|
self.assertFalse(db.rpc_params["p_allow_decrease"])
|
|
|
|
def test_lower_original_site_value_does_not_replace_log_maximum(self):
|
|
db = FakeDb()
|
|
plant = {
|
|
"id": "plant-a",
|
|
"type": "test",
|
|
"options": {},
|
|
}
|
|
|
|
with (
|
|
patch.object(daily_summary, "get_supabase_client", return_value=db),
|
|
patch.object(daily_summary, "get_all_plants", return_value=[plant]),
|
|
patch.object(
|
|
daily_summary,
|
|
"get_history_crawler",
|
|
return_value=lambda *_args: [{
|
|
"plant_id": "plant-a",
|
|
"date": "2026-08-05",
|
|
"generation_kwh": 100,
|
|
}],
|
|
),
|
|
):
|
|
result = daily_summary.calculate_daily_stats("2026-08-05")
|
|
|
|
self.assertTrue(result)
|
|
self.assertEqual(120, db.saved[0]["total_generation"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|