843f72e048
All checks were successful
Deploy to Production / deploy (push) Successful in 1m17s
- Add Alarm model with tenant/part/machine relationships and 3 indexes - Add alembic migration c5d6e7f8a9b0_add_alarms - Add alarms API: list (filter by ack/severity), summary, acknowledge - Auto-generate alarms on counter update (threshold warning, critical at 100%) - Duplicate alarm prevention for same part+alarm_type - Add alarms frontend page with active/acknowledged tabs, summary badges - 9 new tests (67/67 total passing) Co-Authored-By: Claude Opus 4 <noreply@anthropic.com>
226 lines
6.7 KiB
Python
226 lines
6.7 KiB
Python
import pytest
|
|
from httpx import AsyncClient
|
|
from tests.conftest import get_auth_headers
|
|
|
|
|
|
async def _create_machine(
|
|
client: AsyncClient, headers: dict, tenant_id: str = "test-co"
|
|
) -> str:
|
|
resp = await client.post(
|
|
f"/api/{tenant_id}/machines",
|
|
json={"name": "알람 테스트 설비", "equipment_code": "A-001"},
|
|
headers=headers,
|
|
)
|
|
return resp.json()["id"]
|
|
|
|
|
|
async def _create_part(
|
|
client: AsyncClient,
|
|
headers: dict,
|
|
machine_id: str,
|
|
tenant_id: str = "test-co",
|
|
alarm_threshold: float = 80.0,
|
|
) -> str:
|
|
resp = await client.post(
|
|
f"/api/{tenant_id}/machines/{machine_id}/parts",
|
|
json={
|
|
"name": "알람 부품",
|
|
"lifecycle_type": "hours",
|
|
"lifecycle_limit": 1000,
|
|
"alarm_threshold": alarm_threshold,
|
|
"counter_source": "manual",
|
|
},
|
|
headers=headers,
|
|
)
|
|
return resp.json()["id"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_alarm_below_threshold(client: AsyncClient, seeded_db):
|
|
headers = await get_auth_headers(client)
|
|
mid = await _create_machine(client, headers)
|
|
pid = await _create_part(client, headers, mid)
|
|
|
|
await client.put(
|
|
f"/api/test-co/parts/{pid}/counter",
|
|
json={"value": 700},
|
|
headers=headers,
|
|
)
|
|
|
|
resp = await client.get("/api/test-co/alarms", headers=headers)
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()["alarms"]) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_warning_alarm_at_threshold(client: AsyncClient, seeded_db):
|
|
headers = await get_auth_headers(client)
|
|
mid = await _create_machine(client, headers)
|
|
pid = await _create_part(client, headers, mid)
|
|
|
|
await client.put(
|
|
f"/api/test-co/parts/{pid}/counter",
|
|
json={"value": 850},
|
|
headers=headers,
|
|
)
|
|
|
|
resp = await client.get("/api/test-co/alarms", headers=headers)
|
|
assert resp.status_code == 200
|
|
alarms = resp.json()["alarms"]
|
|
assert len(alarms) == 1
|
|
assert alarms[0]["severity"] == "warning"
|
|
assert alarms[0]["alarm_type"] == "threshold"
|
|
assert alarms[0]["is_acknowledged"] is False
|
|
assert "85.0%" in alarms[0]["message"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_critical_alarm_at_100pct(client: AsyncClient, seeded_db):
|
|
headers = await get_auth_headers(client)
|
|
mid = await _create_machine(client, headers)
|
|
pid = await _create_part(client, headers, mid)
|
|
|
|
await client.put(
|
|
f"/api/test-co/parts/{pid}/counter",
|
|
json={"value": 1100},
|
|
headers=headers,
|
|
)
|
|
|
|
resp = await client.get("/api/test-co/alarms", headers=headers)
|
|
assert resp.status_code == 200
|
|
alarms = resp.json()["alarms"]
|
|
assert len(alarms) == 1
|
|
assert alarms[0]["severity"] == "critical"
|
|
assert alarms[0]["alarm_type"] == "critical"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_duplicate_alarm(client: AsyncClient, seeded_db):
|
|
headers = await get_auth_headers(client)
|
|
mid = await _create_machine(client, headers)
|
|
pid = await _create_part(client, headers, mid)
|
|
|
|
await client.put(
|
|
f"/api/test-co/parts/{pid}/counter",
|
|
json={"value": 900},
|
|
headers=headers,
|
|
)
|
|
await client.put(
|
|
f"/api/test-co/parts/{pid}/counter",
|
|
json={"value": 950},
|
|
headers=headers,
|
|
)
|
|
|
|
resp = await client.get("/api/test-co/alarms", headers=headers)
|
|
alarms = resp.json()["alarms"]
|
|
assert len(alarms) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_acknowledge_alarm(client: AsyncClient, seeded_db):
|
|
headers = await get_auth_headers(client)
|
|
mid = await _create_machine(client, headers)
|
|
pid = await _create_part(client, headers, mid)
|
|
|
|
await client.put(
|
|
f"/api/test-co/parts/{pid}/counter",
|
|
json={"value": 850},
|
|
headers=headers,
|
|
)
|
|
|
|
resp = await client.get("/api/test-co/alarms", headers=headers)
|
|
alarm_id = resp.json()["alarms"][0]["id"]
|
|
|
|
ack_resp = await client.post(
|
|
f"/api/test-co/alarms/{alarm_id}/acknowledge",
|
|
headers=headers,
|
|
)
|
|
assert ack_resp.status_code == 200
|
|
assert ack_resp.json()["is_acknowledged"] is True
|
|
assert ack_resp.json()["acknowledged_by"] is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_acknowledge_already_acknowledged(client: AsyncClient, seeded_db):
|
|
headers = await get_auth_headers(client)
|
|
mid = await _create_machine(client, headers)
|
|
pid = await _create_part(client, headers, mid)
|
|
|
|
await client.put(
|
|
f"/api/test-co/parts/{pid}/counter",
|
|
json={"value": 850},
|
|
headers=headers,
|
|
)
|
|
|
|
resp = await client.get("/api/test-co/alarms", headers=headers)
|
|
alarm_id = resp.json()["alarms"][0]["id"]
|
|
|
|
await client.post(f"/api/test-co/alarms/{alarm_id}/acknowledge", headers=headers)
|
|
ack2 = await client.post(
|
|
f"/api/test-co/alarms/{alarm_id}/acknowledge", headers=headers
|
|
)
|
|
assert ack2.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filter_unacknowledged(client: AsyncClient, seeded_db):
|
|
headers = await get_auth_headers(client)
|
|
mid = await _create_machine(client, headers)
|
|
pid = await _create_part(client, headers, mid)
|
|
|
|
await client.put(
|
|
f"/api/test-co/parts/{pid}/counter",
|
|
json={"value": 850},
|
|
headers=headers,
|
|
)
|
|
|
|
resp = await client.get(
|
|
"/api/test-co/alarms?is_acknowledged=false", headers=headers
|
|
)
|
|
assert len(resp.json()["alarms"]) == 1
|
|
|
|
alarm_id = resp.json()["alarms"][0]["id"]
|
|
await client.post(f"/api/test-co/alarms/{alarm_id}/acknowledge", headers=headers)
|
|
|
|
resp2 = await client.get(
|
|
"/api/test-co/alarms?is_acknowledged=false", headers=headers
|
|
)
|
|
assert len(resp2.json()["alarms"]) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alarm_summary(client: AsyncClient, seeded_db):
|
|
headers = await get_auth_headers(client)
|
|
mid = await _create_machine(client, headers)
|
|
pid = await _create_part(client, headers, mid)
|
|
|
|
await client.put(
|
|
f"/api/test-co/parts/{pid}/counter",
|
|
json={"value": 850},
|
|
headers=headers,
|
|
)
|
|
|
|
resp = await client.get("/api/test-co/alarms/summary", headers=headers)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["active_count"] == 1
|
|
assert data["warning_count"] == 1
|
|
assert data["critical_count"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tenant_isolation(client: AsyncClient, seeded_db):
|
|
headers = await get_auth_headers(client)
|
|
mid = await _create_machine(client, headers)
|
|
pid = await _create_part(client, headers, mid)
|
|
|
|
await client.put(
|
|
f"/api/test-co/parts/{pid}/counter",
|
|
json={"value": 900},
|
|
headers=headers,
|
|
)
|
|
|
|
resp_other = await client.get("/api/other-co/alarms", headers=headers)
|
|
assert resp_other.status_code == 200
|
|
assert len(resp_other.json()["alarms"]) == 0
|