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
154 lines
4.5 KiB
Python
154 lines
4.5 KiB
Python
import unittest
|
|
from types import SimpleNamespace
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from app.routers import plants
|
|
|
|
|
|
PLANT = {
|
|
"id": "nrems-03",
|
|
"company_id": 1,
|
|
"name": "3호기",
|
|
"capacity": 99.0,
|
|
"location": None,
|
|
"alerts_enabled": True,
|
|
"created_at": "2026-01-01T00:00:00+00:00",
|
|
}
|
|
|
|
LOG = {
|
|
"id": 10,
|
|
"plant_id": "nrems-03",
|
|
"current_kw": 20.0,
|
|
"today_kwh": 100.0,
|
|
"status": "정상",
|
|
"created_at": "2026-08-07T03:00:00+00:00",
|
|
}
|
|
|
|
|
|
class FakeQuery:
|
|
def __init__(self, db, table_name):
|
|
self.db = db
|
|
self.table_name = table_name
|
|
self.columns = None
|
|
self.filters = []
|
|
self.action = "select"
|
|
self.payload = None
|
|
self.db.queries.append(self)
|
|
|
|
def select(self, columns):
|
|
self.columns = columns
|
|
return self
|
|
|
|
def eq(self, column, value):
|
|
self.filters.append((column, value))
|
|
return self
|
|
|
|
def order(self, _column, desc=False, foreign_table=None):
|
|
self.foreign_table = foreign_table
|
|
return self
|
|
|
|
def limit(self, _count, foreign_table=None):
|
|
self.limit_foreign_table = foreign_table
|
|
return self
|
|
|
|
def update(self, payload):
|
|
self.action = "update"
|
|
self.payload = payload
|
|
return self
|
|
|
|
def execute(self):
|
|
if self.db.query_failure:
|
|
raise RuntimeError("database unavailable")
|
|
if self.table_name == "plants":
|
|
if not self.db.plant_exists:
|
|
return SimpleNamespace(data=[])
|
|
if self.action == "update":
|
|
updated = dict(PLANT, **self.payload)
|
|
return SimpleNamespace(data=[updated])
|
|
if "solar_logs(" in self.columns:
|
|
return SimpleNamespace(data=[dict(PLANT, solar_logs=[LOG])])
|
|
return SimpleNamespace(data=[PLANT])
|
|
if self.table_name == "solar_logs":
|
|
return SimpleNamespace(data=[LOG])
|
|
raise AssertionError(f"unexpected table: {self.table_name}")
|
|
|
|
|
|
class FakeDb:
|
|
def __init__(self, plant_exists=True, query_failure=False):
|
|
self.plant_exists = plant_exists
|
|
self.query_failure = query_failure
|
|
self.queries = []
|
|
|
|
def table(self, table_name):
|
|
return FakeQuery(self, table_name)
|
|
|
|
|
|
class PlantsApiTest(unittest.TestCase):
|
|
def test_company_list_uses_one_nested_query(self):
|
|
db = FakeDb()
|
|
|
|
result = plants.get_plants_by_company(company_id=1, db=db)
|
|
|
|
self.assertEqual(1, len(db.queries))
|
|
self.assertEqual("plants", db.queries[0].table_name)
|
|
self.assertEqual("solar_logs", db.queries[0].foreign_table)
|
|
self.assertEqual("solar_logs", db.queries[0].limit_foreign_table)
|
|
self.assertEqual("nrems-03", result.data[0].id)
|
|
self.assertEqual(10, result.data[0].latest_log.id)
|
|
|
|
def test_company_without_plants_returns_empty_success(self):
|
|
result = plants.get_plants_by_company(
|
|
company_id=999,
|
|
db=FakeDb(plant_exists=False),
|
|
)
|
|
|
|
self.assertEqual("success", result.status)
|
|
self.assertEqual([], result.data)
|
|
self.assertEqual(0, result.total_count)
|
|
|
|
def test_company_query_failure_returns_500(self):
|
|
with self.assertRaises(HTTPException) as raised:
|
|
plants.get_plants_by_company(
|
|
company_id=1,
|
|
db=FakeDb(query_failure=True),
|
|
)
|
|
|
|
self.assertEqual(500, raised.exception.status_code)
|
|
|
|
def test_string_plant_id_is_used_for_detail_lookup(self):
|
|
db = FakeDb()
|
|
|
|
result = plants.get_plant_detail(
|
|
company_id=1, plant_id="nrems-03", db=db
|
|
)
|
|
|
|
plant_query = db.queries[0]
|
|
self.assertIn(("id", "nrems-03"), plant_query.filters)
|
|
self.assertEqual("nrems-03", result["data"]["plant"]["id"])
|
|
self.assertEqual(10, result["data"]["recent_logs"][0]["id"])
|
|
|
|
def test_missing_detail_returns_404(self):
|
|
with self.assertRaises(HTTPException) as raised:
|
|
plants.get_plant_detail(
|
|
company_id=1,
|
|
plant_id="missing",
|
|
db=FakeDb(plant_exists=False),
|
|
)
|
|
|
|
self.assertEqual(404, raised.exception.status_code)
|
|
|
|
def test_alert_update_uses_typed_response(self):
|
|
result = plants.update_plant_alerts(
|
|
company_id=1,
|
|
plant_id="nrems-03",
|
|
alert_update=plants.PlantAlertUpdateRequest(alerts_enabled=False),
|
|
db=FakeDb(),
|
|
)
|
|
|
|
self.assertFalse(result.data.alerts_enabled)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|