96 lines
3.5 KiB
Python
96 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from pydantic import ValidationError
|
|
|
|
from lumaops_backend.config import Settings
|
|
from lumaops_backend.database import Database
|
|
from lumaops_backend.logging_config import JsonFormatter
|
|
from lumaops_backend.main import create_app
|
|
from lumaops_backend.secrets import SecretStore
|
|
|
|
|
|
def test_production_lan_bind_requires_authentication() -> None:
|
|
with pytest.raises(ValidationError, match="niet-lokale productiebind"):
|
|
Settings(auth_enabled=False)
|
|
|
|
|
|
def test_authentication_rejects_placeholder_token() -> None:
|
|
with pytest.raises(ValidationError, match="minstens 32 tekens"):
|
|
Settings(LUMAOPS_ADMIN_TOKEN="replace-with-a-long-random-token") # noqa: S106
|
|
|
|
|
|
def test_migration_is_repeatable(settings: Settings) -> None:
|
|
database = Database(settings)
|
|
database.initialize()
|
|
database.initialize()
|
|
with database.connection() as conn:
|
|
versions = conn.execute("SELECT version FROM schema_migrations").fetchall()
|
|
foreign_keys = conn.execute("PRAGMA foreign_keys").fetchone()[0]
|
|
journal_mode = conn.execute("PRAGMA journal_mode").fetchone()[0]
|
|
assert [row["version"] for row in versions] == [
|
|
"0001_initial",
|
|
"0002_automation_descriptions",
|
|
"0003_device_desired_state",
|
|
"0004_device_classification",
|
|
]
|
|
assert foreign_keys == 1
|
|
assert journal_mode == "wal"
|
|
|
|
|
|
def test_secrets_are_encrypted_and_key_is_persistent(settings: Settings) -> None:
|
|
first = SecretStore(settings)
|
|
ciphertext = first.encrypt("very-secret-token")
|
|
assert b"very-secret-token" not in ciphertext
|
|
second = SecretStore(settings)
|
|
assert second.decrypt(ciphertext) == "very-secret-token"
|
|
assert (settings.config_dir / "secret.key").exists()
|
|
|
|
|
|
def test_json_logger_redacts_secret_fields() -> None:
|
|
formatter = JsonFormatter()
|
|
record = logging.LogRecord(
|
|
"test",
|
|
logging.INFO,
|
|
__file__,
|
|
1,
|
|
"payload %s",
|
|
({"api_key": "secret", "device": "ok"},),
|
|
None,
|
|
)
|
|
rendered = formatter.format(record)
|
|
assert "[REDACTED]" in rendered
|
|
assert "secret" not in rendered
|
|
|
|
|
|
def test_auth_cookie_requires_csrf(tmp_path: Path) -> None:
|
|
settings = Settings(
|
|
LUMAOPS_ENV="test",
|
|
connector_mode="mock",
|
|
auth_enabled=True,
|
|
LUMAOPS_ADMIN_TOKEN="correct-horse-battery-staple-test-token", # noqa: S106 - test-only token
|
|
secure_cookies=False,
|
|
config_dir=tmp_path / "config",
|
|
openrgb_config_dir=tmp_path / "openrgb",
|
|
data_dir=tmp_path / "data",
|
|
logs_dir=tmp_path / "logs",
|
|
static_dir=tmp_path / "static",
|
|
database_url=f"sqlite:///{(tmp_path / 'data' / 'auth.db').as_posix()}",
|
|
log_level="ERROR",
|
|
)
|
|
with TestClient(create_app(settings)) as client:
|
|
assert client.get("/api/v1/system").status_code == 401
|
|
login = client.post(
|
|
"/api/v1/auth/login", json={"token": "correct-horse-battery-staple-test-token"}
|
|
)
|
|
assert login.status_code == 200
|
|
assert client.get("/api/v1/system").status_code == 200
|
|
without_csrf = client.post("/api/v1/discovery")
|
|
assert without_csrf.status_code == 403
|
|
csrf = login.json()["csrf_token"]
|
|
assert client.post("/api/v1/discovery", headers={"X-CSRF-Token": csrf}).status_code == 200
|