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
128 lines
4.5 KiB
Python
128 lines
4.5 KiB
Python
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import closing
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import Mock
|
|
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))
|
|
|
|
from alert_manager import AlertManager
|
|
from crawlers.base import is_data_valid
|
|
|
|
|
|
class AlertManagerTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
db_path = Path(self.temp_dir.name) / "alert_test.db"
|
|
self.manager = AlertManager(
|
|
db_path=db_path,
|
|
now_provider=lambda: datetime(2026, 8, 5, 12, 0, 0)
|
|
)
|
|
self.real_is_alert_enabled = self.manager._is_alert_enabled
|
|
self.manager._is_alert_enabled = Mock(return_value=True)
|
|
self.manager.send_telegram_message = Mock(return_value=True)
|
|
self.plant = {
|
|
"id": "nrems-03",
|
|
"name": "3호기",
|
|
"display_name": "3호기",
|
|
"telegram_chat_id": 1234,
|
|
}
|
|
|
|
def tearDown(self):
|
|
self.temp_dir.cleanup()
|
|
|
|
def get_state(self):
|
|
with closing(sqlite3.connect(self.manager.db_path)) as conn:
|
|
return conn.execute("""
|
|
SELECT alert_status, zero_count, first_zero_today_kwh
|
|
FROM alert_history
|
|
WHERE site_id = ?
|
|
""", (self.plant["id"],)).fetchone()
|
|
|
|
def test_collection_error_breaks_zero_sequence(self):
|
|
self.manager.check_and_alert(self.plant, 0, 100)
|
|
self.manager.check_and_alert(self.plant, 0, 100)
|
|
self.assertEqual(("NORMAL", 2, 100.0), self.get_state())
|
|
|
|
self.manager.check_and_alert(
|
|
self.plant,
|
|
0,
|
|
0,
|
|
data_valid=False
|
|
)
|
|
self.assertEqual(("NORMAL", 0, None), self.get_state())
|
|
|
|
self.manager.check_and_alert(self.plant, 0, 100)
|
|
self.manager.check_and_alert(self.plant, 0, 100)
|
|
self.manager.send_telegram_message.assert_not_called()
|
|
|
|
self.manager.check_and_alert(self.plant, 0, 100)
|
|
self.manager.send_telegram_message.assert_called_once()
|
|
self.assertEqual(("ALERT", 3, 100.0), self.get_state())
|
|
|
|
def test_cumulative_generation_growth_clears_zero_suspicion(self):
|
|
self.manager.check_and_alert(self.plant, 0, 100)
|
|
self.manager.check_and_alert(self.plant, 0, 101)
|
|
|
|
self.manager.send_telegram_message.assert_not_called()
|
|
self.assertEqual(("NORMAL", 0, None), self.get_state())
|
|
|
|
def test_disabled_alert_does_not_create_alert_state(self):
|
|
self.manager._is_alert_enabled.return_value = False
|
|
|
|
for _ in range(3):
|
|
self.manager.check_and_alert(self.plant, 0, 100)
|
|
|
|
self.manager.send_telegram_message.assert_not_called()
|
|
self.assertIsNone(self.get_state())
|
|
|
|
def test_positive_generation_recovers_alert_state(self):
|
|
for _ in range(3):
|
|
self.manager.check_and_alert(self.plant, 0, 100)
|
|
|
|
self.assertEqual("ALERT", self.get_state()[0])
|
|
|
|
self.manager.check_and_alert(self.plant, 10, 101)
|
|
|
|
self.assertEqual(("NORMAL", 0, None), self.get_state())
|
|
|
|
def test_alert_setting_uses_globally_unique_plant_id(self):
|
|
query = Mock()
|
|
query.select.return_value = query
|
|
query.eq.return_value = query
|
|
query.limit.return_value = query
|
|
query.execute.return_value = SimpleNamespace(
|
|
data=[{"alerts_enabled": False}]
|
|
)
|
|
client = Mock()
|
|
client.table.return_value = query
|
|
database_module = SimpleNamespace(
|
|
get_supabase_client=Mock(return_value=client)
|
|
)
|
|
|
|
with patch.dict(sys.modules, {"database": database_module}):
|
|
enabled = self.real_is_alert_enabled("nrems-03", "3호기")
|
|
|
|
self.assertFalse(enabled)
|
|
client.table.assert_called_once_with("plants")
|
|
query.eq.assert_called_once_with("id", "nrems-03")
|
|
|
|
def test_data_validity_distinguishes_collection_error_from_plant_fault(self):
|
|
self.assertTrue(is_data_valid({"status": "🟢 정상"}))
|
|
self.assertTrue(is_data_valid({"status": "🔴 점검/고장", "kw": 0}))
|
|
self.assertFalse(is_data_valid({"status": "🔴 오류"}))
|
|
self.assertFalse(is_data_valid({"status": "ERROR"}))
|
|
self.assertFalse(is_data_valid({"status": "🟢 정상", "data_valid": False}))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|