75 lines
3.1 KiB
Python
75 lines
3.1 KiB
Python
"""Tests für den Home-Assistant-Status-Endpunkt (router/ha.py)."""
|
|
import pytest
|
|
import pytest_asyncio
|
|
from httpx import AsyncClient
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="session", loop_scope="session")
|
|
async def ha_headers(client: AsyncClient):
|
|
resp = await client.post("/api/v1/auth/register", json={
|
|
"company_name": "HA GmbH",
|
|
"first_name": "Home",
|
|
"last_name": "Assistant",
|
|
"email": "admin@hagmbh.de",
|
|
"password": "Secret123",
|
|
})
|
|
assert resp.status_code == 201, resp.text
|
|
return {"Authorization": f"Bearer {resp.json()['access_token']}"}
|
|
|
|
|
|
@pytest.mark.asyncio(loop_scope="session")
|
|
async def test_ha_token_disabled_by_default(client: AsyncClient, ha_headers):
|
|
r = await client.get("/api/v1/users/me/ha-token", headers=ha_headers)
|
|
assert r.status_code == 200
|
|
assert r.json()["enabled"] is False
|
|
|
|
|
|
@pytest.mark.asyncio(loop_scope="session")
|
|
async def test_ha_status_lifecycle(client: AsyncClient, ha_headers):
|
|
r = await client.post("/api/v1/users/me/ha-token", json={}, headers=ha_headers)
|
|
assert r.status_code == 200, r.text
|
|
path = r.json()["url"].split("/api/v1", 1)[1]
|
|
|
|
out = await client.get("/api/v1" + path)
|
|
assert out.status_code == 200
|
|
body = out.json()
|
|
assert body["status"] == "out"
|
|
assert body["worked_today_min"] == 0
|
|
assert body["stamped_in_since"] is None
|
|
|
|
s = await client.post("/api/v1/time/stamp-in", json={}, headers=ha_headers)
|
|
assert s.status_code == 201, s.text
|
|
body = (await client.get("/api/v1" + path)).json()
|
|
assert body["status"] == "working"
|
|
assert body["stamped_in_since"] is not None
|
|
|
|
b = await client.post("/api/v1/time/break-start", json={}, headers=ha_headers)
|
|
assert b.status_code == 200, b.text
|
|
assert (await client.get("/api/v1" + path)).json()["status"] == "break"
|
|
|
|
await client.post("/api/v1/time/break-end", json={}, headers=ha_headers)
|
|
await client.post("/api/v1/time/stamp-out", json={}, headers=ha_headers)
|
|
assert (await client.get("/api/v1" + path)).json()["status"] == "out"
|
|
|
|
r2 = await client.post("/api/v1/users/me/ha-token", json={}, headers=ha_headers)
|
|
new_path = r2.json()["url"].split("/api/v1", 1)[1]
|
|
assert (await client.get("/api/v1" + path)).status_code == 404
|
|
assert (await client.get("/api/v1" + new_path)).status_code == 200
|
|
|
|
d = await client.delete("/api/v1/users/me/ha-token", headers=ha_headers)
|
|
assert d.status_code == 204
|
|
assert (await client.get("/api/v1" + new_path)).status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio(loop_scope="session")
|
|
async def test_ha_token_rejected_for_inactive_company(client: AsyncClient, ha_headers, db_session):
|
|
from sqlalchemy import text
|
|
r = await client.post("/api/v1/users/me/ha-token", json={}, headers=ha_headers)
|
|
path = r.json()["url"].split("/api/v1", 1)[1]
|
|
assert (await client.get("/api/v1" + path)).status_code == 200
|
|
|
|
await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
|
|
await db_session.execute(text("UPDATE companies SET is_active = false WHERE name = 'HA GmbH'"))
|
|
await db_session.commit()
|
|
assert (await client.get("/api/v1" + path)).status_code == 404
|