test: unify auth headers into a single conftest fixture
Replace 9 duplicated per-file 'headers' fixtures and the awkward 'auth_headers' coroutine factory with one canonical 'headers' fixture in conftest.py. Migrate liveview/media tests off the await-based factory. No behavior change; 627 passed. ha-relevant: no
This commit is contained in:
@@ -44,10 +44,7 @@ async def client(db_session: AsyncSession):
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_headers(client):
|
||||
"""Returns a coroutine that logs in and returns auth headers."""
|
||||
async def _get():
|
||||
res = await client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"})
|
||||
token = res.json()["access_token"]
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
return _get
|
||||
async def headers(client: AsyncClient):
|
||||
"""Authenticated Bearer headers for the default admin test user."""
|
||||
res = await client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"})
|
||||
return {"Authorization": f"Bearer {res.json()['access_token']}"}
|
||||
|
||||
@@ -1,15 +1,8 @@
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def headers(client: AsyncClient):
|
||||
res = await client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"})
|
||||
return {"Authorization": f"Bearer {res.json()['access_token']}"}
|
||||
|
||||
|
||||
def node_payload(**kwargs):
|
||||
return {"id": str(uuid.uuid4()), "type": "server", "label": "N", "status": "unknown", "pos_x": 0, "pos_y": 0, **kwargs}
|
||||
|
||||
|
||||
@@ -1,15 +1,8 @@
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def headers(client: AsyncClient):
|
||||
res = await client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"})
|
||||
return {"Authorization": f"Bearer {res.json()['access_token']}"}
|
||||
|
||||
|
||||
def node_payload(**kwargs):
|
||||
return {"id": str(uuid.uuid4()), "type": "server", "label": "N", "status": "unknown", "pos_x": 0, "pos_y": 0, **kwargs}
|
||||
|
||||
|
||||
@@ -2,13 +2,6 @@ import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def headers(client: AsyncClient):
|
||||
res = await client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"})
|
||||
token = res.json()["access_token"]
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def two_nodes(client: AsyncClient, headers: dict):
|
||||
n1 = (await client.post("/api/v1/nodes", json={"type": "router", "label": "R1", "status": "online"}, headers=headers)).json()
|
||||
|
||||
@@ -82,10 +82,9 @@ async def test_liveview_does_not_require_jwt(client: AsyncClient):
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_liveview_returns_saved_canvas(client: AsyncClient, auth_headers):
|
||||
async def test_liveview_returns_saved_canvas(client: AsyncClient, headers):
|
||||
"""Canvas saved via POST /canvas/save appears in liveview response."""
|
||||
settings.liveview_key = "test-key"
|
||||
headers = await auth_headers()
|
||||
|
||||
# Save a canvas with one node
|
||||
payload = {
|
||||
@@ -115,10 +114,9 @@ async def test_liveview_returns_saved_canvas(client: AsyncClient, auth_headers):
|
||||
# ── custom_style + theme propagation ─────────────────────────────────────────
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_liveview_returns_custom_style_and_theme(client: AsyncClient, auth_headers):
|
||||
async def test_liveview_returns_custom_style_and_theme(client: AsyncClient, headers):
|
||||
"""custom_style and viewport.theme_id from a saved canvas surface in liveview."""
|
||||
settings.liveview_key = "test-key"
|
||||
headers = await auth_headers()
|
||||
payload = {
|
||||
"nodes": [],
|
||||
"edges": [],
|
||||
@@ -159,9 +157,8 @@ async def test_liveview_config_requires_auth(client: AsyncClient):
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_liveview_config_returns_key_when_enabled(client: AsyncClient, auth_headers):
|
||||
async def test_liveview_config_returns_key_when_enabled(client: AsyncClient, headers):
|
||||
settings.liveview_key = "share-me"
|
||||
headers = await auth_headers()
|
||||
res = await client.get("/api/v1/liveview/config", headers=headers)
|
||||
assert res.status_code == 200
|
||||
body = res.json()
|
||||
@@ -169,18 +166,16 @@ async def test_liveview_config_returns_key_when_enabled(client: AsyncClient, aut
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_liveview_config_disabled_hides_key(client: AsyncClient, auth_headers):
|
||||
async def test_liveview_config_disabled_hides_key(client: AsyncClient, headers):
|
||||
settings.liveview_key = None
|
||||
headers = await auth_headers()
|
||||
res = await client.get("/api/v1/liveview/config", headers=headers)
|
||||
assert res.status_code == 200
|
||||
assert res.json() == {"enabled": False, "key": None}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_liveview_config_empty_key_disabled(client: AsyncClient, auth_headers):
|
||||
async def test_liveview_config_empty_key_disabled(client: AsyncClient, headers):
|
||||
settings.liveview_key = ""
|
||||
headers = await auth_headers()
|
||||
res = await client.get("/api/v1/liveview/config", headers=headers)
|
||||
assert res.status_code == 200
|
||||
assert res.json() == {"enabled": False, "key": None}
|
||||
@@ -189,10 +184,9 @@ async def test_liveview_config_empty_key_disabled(client: AsyncClient, auth_head
|
||||
# ── design_id selects which canvas is rendered ───────────────────────────────
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_liveview_design_id_selects_canvas(client: AsyncClient, auth_headers):
|
||||
async def test_liveview_design_id_selects_canvas(client: AsyncClient, headers):
|
||||
"""?design_id=<id> renders that design's canvas, not the first one."""
|
||||
settings.liveview_key = "test-key"
|
||||
headers = await auth_headers()
|
||||
|
||||
# Create two designs
|
||||
d1 = (await client.post("/api/v1/designs", json={"name": "Network"}, headers=headers)).json()
|
||||
|
||||
@@ -30,8 +30,7 @@ async def test_upload_requires_auth(client, media_dir):
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_stores_file_and_returns_url(client, auth_headers, media_dir):
|
||||
headers = await auth_headers()
|
||||
async def test_upload_stores_file_and_returns_url(client, headers, media_dir):
|
||||
res = await _upload(client, headers)
|
||||
assert res.status_code == 200
|
||||
body = res.json()
|
||||
@@ -42,31 +41,27 @@ async def test_upload_stores_file_and_returns_url(client, auth_headers, media_di
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_rejects_unsupported_type(client, auth_headers, media_dir):
|
||||
headers = await auth_headers()
|
||||
async def test_upload_rejects_unsupported_type(client, headers, media_dir):
|
||||
res = await _upload(client, headers, name="a.txt", data=b"hello", content_type="text/plain")
|
||||
assert res.status_code == 415
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_rejects_content_type_magic_mismatch(client, auth_headers, media_dir):
|
||||
headers = await auth_headers()
|
||||
async def test_upload_rejects_content_type_magic_mismatch(client, headers, media_dir):
|
||||
# Claims PNG but bytes are not a PNG.
|
||||
res = await _upload(client, headers, data=b"not-a-real-png", content_type="image/png")
|
||||
assert res.status_code == 415
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_rejects_oversize(client, auth_headers, media_dir, monkeypatch):
|
||||
async def test_upload_rejects_oversize(client, headers, media_dir, monkeypatch):
|
||||
monkeypatch.setattr(media, "MAX_BYTES", 8)
|
||||
headers = await auth_headers()
|
||||
res = await _upload(client, headers, data=PNG_BYTES) # > 8 bytes
|
||||
assert res.status_code == 413
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_serves_uploaded_file(client, auth_headers, media_dir):
|
||||
headers = await auth_headers()
|
||||
async def test_get_serves_uploaded_file(client, headers, media_dir):
|
||||
up = await _upload(client, headers)
|
||||
res = await client.get(up.json()["url"]) # public, no auth
|
||||
assert res.status_code == 200
|
||||
@@ -80,15 +75,13 @@ async def test_get_rejects_bad_filename(client, media_dir):
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_rejects_bad_filename(client, auth_headers, media_dir):
|
||||
headers = await auth_headers()
|
||||
async def test_delete_rejects_bad_filename(client, headers, media_dir):
|
||||
res = await client.delete("/api/v1/media/..%2f..%2fetc%2fpasswd", headers=headers)
|
||||
assert res.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_requires_auth_and_removes_file(client, auth_headers, media_dir):
|
||||
headers = await auth_headers()
|
||||
async def test_delete_requires_auth_and_removes_file(client, headers, media_dir):
|
||||
up = await _upload(client, headers)
|
||||
filename = up.json()["filename"]
|
||||
|
||||
|
||||
@@ -1,14 +1,6 @@
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def headers(client: AsyncClient):
|
||||
res = await client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"})
|
||||
token = res.json()["access_token"]
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
async def test_list_nodes_empty(client: AsyncClient, headers: dict):
|
||||
res = await client.get("/api/v1/nodes", headers=headers)
|
||||
assert res.status_code == 200
|
||||
|
||||
@@ -20,13 +20,6 @@ from app.core.config import settings
|
||||
from app.db.models import Design, Edge, Node, PendingDevice, PendingDeviceLink
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def headers(client: AsyncClient):
|
||||
res = await client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"})
|
||||
token = res.json()["access_token"]
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clear_env_token():
|
||||
"""Ensure a clean token state per test; restore afterwards."""
|
||||
|
||||
@@ -12,13 +12,6 @@ from app.db.models import Design, Node, PendingDevice, ScanRun
|
||||
from app.services.scanner import _cancelled_runs, request_cancel, run_scan
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def headers(client: AsyncClient):
|
||||
res = await client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"})
|
||||
token = res.json()["access_token"]
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def pending_device(db_session):
|
||||
import uuid
|
||||
|
||||
@@ -8,13 +8,6 @@ from httpx import AsyncClient
|
||||
from app.core.config import Settings
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def headers(client: AsyncClient):
|
||||
res = await client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"})
|
||||
token = res.json()["access_token"]
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_settings_requires_auth(client: AsyncClient):
|
||||
res = await client.get("/api/v1/settings")
|
||||
|
||||
@@ -11,13 +11,6 @@ from httpx import AsyncClient
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.fixture
|
||||
async def headers(client: AsyncClient):
|
||||
res = await client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"})
|
||||
token = res.json()["access_token"]
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# /api/v1/zigbee/test-connection
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -7,14 +7,6 @@ from unittest.mock import patch
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def headers(client: AsyncClient):
|
||||
res = await client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"})
|
||||
token = res.json()["access_token"]
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# /api/v1/zwave/test-connection
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user