|
|
@@ -0,0 +1,131 @@
|
|
|
+"""Unit tests for the ``backend.app.cli`` kiosk-bootstrap subcommand."""
|
|
|
+
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+from collections.abc import AsyncGenerator
|
|
|
+
|
|
|
+import pytest
|
|
|
+import pytest_asyncio
|
|
|
+from sqlalchemy import select
|
|
|
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
|
+
|
|
|
+from backend.app.cli import DEFAULT_KIOSK_KEY_NAME, KioskBootstrapError, kiosk_bootstrap
|
|
|
+from backend.app.core.auth import _validate_api_key
|
|
|
+from backend.app.core.database import Base
|
|
|
+from backend.app.models.api_key import APIKey
|
|
|
+
|
|
|
+
|
|
|
+@pytest_asyncio.fixture
|
|
|
+async def session_maker() -> AsyncGenerator[async_sessionmaker, None]:
|
|
|
+ engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
|
+ async with engine.begin() as conn:
|
|
|
+ await conn.run_sync(Base.metadata.create_all)
|
|
|
+ maker = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
|
|
+ try:
|
|
|
+ yield maker
|
|
|
+ finally:
|
|
|
+ await engine.dispose()
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+@pytest.mark.unit
|
|
|
+async def test_bootstrap_creates_key_when_none_exists(session_maker):
|
|
|
+ key = await kiosk_bootstrap(
|
|
|
+ DEFAULT_KIOSK_KEY_NAME,
|
|
|
+ force=False,
|
|
|
+ session_maker=session_maker,
|
|
|
+ ensure_schema=False,
|
|
|
+ )
|
|
|
+
|
|
|
+ assert key.startswith("bb_")
|
|
|
+ assert len(key) > 20
|
|
|
+
|
|
|
+ async with session_maker() as db:
|
|
|
+ rows = (await db.execute(select(APIKey))).scalars().all()
|
|
|
+ assert len(rows) == 1
|
|
|
+ row = rows[0]
|
|
|
+ assert row.name == DEFAULT_KIOSK_KEY_NAME
|
|
|
+ assert row.enabled is True
|
|
|
+ assert row.can_queue is False
|
|
|
+ assert row.can_control_printer is False
|
|
|
+ assert row.can_read_status is True
|
|
|
+ assert row.printer_ids is None
|
|
|
+ assert row.expires_at is None
|
|
|
+ assert row.key_prefix.startswith("bb_")
|
|
|
+ assert row.key_hash != key # stored value is a hash, not the plaintext
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+@pytest.mark.unit
|
|
|
+async def test_bootstrap_refuses_to_overwrite_without_force(session_maker):
|
|
|
+ first = await kiosk_bootstrap(
|
|
|
+ DEFAULT_KIOSK_KEY_NAME,
|
|
|
+ force=False,
|
|
|
+ session_maker=session_maker,
|
|
|
+ ensure_schema=False,
|
|
|
+ )
|
|
|
+
|
|
|
+ with pytest.raises(KioskBootstrapError) as exc_info:
|
|
|
+ await kiosk_bootstrap(
|
|
|
+ DEFAULT_KIOSK_KEY_NAME,
|
|
|
+ force=False,
|
|
|
+ session_maker=session_maker,
|
|
|
+ ensure_schema=False,
|
|
|
+ )
|
|
|
+
|
|
|
+ assert "already exists" in str(exc_info.value)
|
|
|
+ assert "--force" in str(exc_info.value)
|
|
|
+
|
|
|
+ # First key survives unchanged and still validates
|
|
|
+ async with session_maker() as db:
|
|
|
+ row = (await db.execute(select(APIKey))).scalar_one()
|
|
|
+ validated = await _validate_api_key(db, first)
|
|
|
+ assert validated is not None
|
|
|
+ assert validated.id == row.id
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+@pytest.mark.unit
|
|
|
+async def test_bootstrap_force_rotates_existing_key(session_maker):
|
|
|
+ first = await kiosk_bootstrap(
|
|
|
+ DEFAULT_KIOSK_KEY_NAME,
|
|
|
+ force=False,
|
|
|
+ session_maker=session_maker,
|
|
|
+ ensure_schema=False,
|
|
|
+ )
|
|
|
+ second = await kiosk_bootstrap(
|
|
|
+ DEFAULT_KIOSK_KEY_NAME,
|
|
|
+ force=True,
|
|
|
+ session_maker=session_maker,
|
|
|
+ ensure_schema=False,
|
|
|
+ )
|
|
|
+
|
|
|
+ assert first != second
|
|
|
+
|
|
|
+ async with session_maker() as db:
|
|
|
+ rows = (await db.execute(select(APIKey))).scalars().all()
|
|
|
+ assert len(rows) == 1 # old row was deleted, not duplicated
|
|
|
+
|
|
|
+ # Old key no longer validates, new key does
|
|
|
+ assert await _validate_api_key(db, first) is None
|
|
|
+ validated = await _validate_api_key(db, second)
|
|
|
+ assert validated is not None
|
|
|
+ assert validated.name == DEFAULT_KIOSK_KEY_NAME
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+@pytest.mark.unit
|
|
|
+async def test_bootstrap_custom_name(session_maker):
|
|
|
+ key = await kiosk_bootstrap(
|
|
|
+ "custom-kiosk-name",
|
|
|
+ force=False,
|
|
|
+ session_maker=session_maker,
|
|
|
+ ensure_schema=False,
|
|
|
+ )
|
|
|
+
|
|
|
+ async with session_maker() as db:
|
|
|
+ row = (await db.execute(select(APIKey))).scalar_one()
|
|
|
+ assert row.name == "custom-kiosk-name"
|
|
|
+ validated = await _validate_api_key(db, key)
|
|
|
+ assert validated is not None
|
|
|
+ assert validated.name == "custom-kiosk-name"
|