busya
f87c873027
- app.py split from 1966-line test_app.py into 7 files (68 tests, 92% coverage): - test_app_lifespan: lifespan, ensure_initial_admin_user - test_app_handlers: exception handlers, HSTS - test_app_middleware: log_requests, middleware chain - test_app_ws_auth: _authenticate_websocket - test_app_ws_endpoint: WS main loop, 5 endpoint handlers - test_app_ws_events: task/maintenance/dataset/translate WS streams - test_app_spa: SPA serving, TestClient integration - migration_engine: extended coverage with init, edge cases, error paths
229 lines
10 KiB
Python
229 lines
10 KiB
Python
# #region Test.AppModule.Lifespan [C:3] [TYPE Module] [SEMANTICS test,app,lifespan,admin,bootstrap]
|
|
# @BRIEF Tests for app.py — ensure_initial_admin_user + lifespan startup/shutdown.
|
|
# @RELATION BINDS_TO -> [AppModule]
|
|
# @TEST_EDGE: missing_env_flag -> ensure_initial_admin_user skips when INITIAL_ADMIN_CREATE is off
|
|
# @TEST_EDGE: missing_admin_creds -> logs warning when flag on but username/password missing
|
|
# @TEST_EDGE: existing_admin_user -> skips creation when user already exists
|
|
# @TEST_EDGE: admin_role_creation -> creates Admin role if missing
|
|
# @TEST_EDGE: db_rollback_on_error -> rolls back DB transaction on exception
|
|
# @TEST_INVARIANT: admin_bootstrap_guard -> VERIFIED_BY: test_ensure_initial_admin_user_flag_off, test_ensure_initial_admin_user_missing_creds
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
import os
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from fastapi import FastAPI
|
|
|
|
|
|
# =============================================================================
|
|
# ensure_initial_admin_user
|
|
# =============================================================================
|
|
|
|
class TestEnsureInitialAdminUser:
|
|
"""ensure_initial_admin_user — environment-gated admin bootstrap."""
|
|
|
|
# #region test_flag_off [C:2] [TYPE Function]
|
|
@patch.dict(os.environ, {"INITIAL_ADMIN_CREATE": "false"}, clear=False)
|
|
def test_flag_off(self):
|
|
from src.app import ensure_initial_admin_user
|
|
result = ensure_initial_admin_user()
|
|
assert result is None
|
|
# #endregion test_flag_off
|
|
|
|
# #region test_flag_on_missing_creds [C:2] [TYPE Function]
|
|
@patch.dict(os.environ, {
|
|
"INITIAL_ADMIN_CREATE": "true", "INITIAL_ADMIN_USERNAME": "", "INITIAL_ADMIN_PASSWORD": "",
|
|
}, clear=False)
|
|
def test_flag_on_missing_creds(self):
|
|
from src.app import ensure_initial_admin_user
|
|
assert ensure_initial_admin_user() is None
|
|
# #endregion test_flag_on_missing_creds
|
|
|
|
# #region test_flag_on_only_username [C:2] [TYPE Function]
|
|
@patch.dict(os.environ, {
|
|
"INITIAL_ADMIN_CREATE": "1", "INITIAL_ADMIN_USERNAME": "admin", "INITIAL_ADMIN_PASSWORD": "",
|
|
}, clear=False)
|
|
def test_flag_on_only_username(self):
|
|
from src.app import ensure_initial_admin_user
|
|
assert ensure_initial_admin_user() is None
|
|
# #endregion test_flag_on_only_username
|
|
|
|
# #region test_creates_admin_role_and_user [C:2] [TYPE Function]
|
|
@patch.dict(os.environ, {
|
|
"INITIAL_ADMIN_CREATE": "true", "INITIAL_ADMIN_USERNAME": "bootstrap-admin", "INITIAL_ADMIN_PASSWORD": "s3cret!",
|
|
}, clear=False)
|
|
def test_creates_admin_role_and_user(self):
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value.filter.return_value.first.side_effect = [None, None]
|
|
mock_db.add = MagicMock()
|
|
with patch("src.app.AuthSessionLocal", return_value=mock_db):
|
|
from src.app import ensure_initial_admin_user
|
|
ensure_initial_admin_user()
|
|
role_calls = [c for c in mock_db.add.call_args_list if c.args[0].__class__.__name__ == "Role"]
|
|
user_calls = [c for c in mock_db.add.call_args_list if c.args[0].__class__.__name__ == "User"]
|
|
assert len(role_calls) == 1
|
|
assert role_calls[0].args[0].name == "Admin"
|
|
assert len(user_calls) == 1
|
|
assert user_calls[0].args[0].username == "bootstrap-admin"
|
|
assert mock_db.commit.call_count == 2
|
|
# #endregion test_creates_admin_role_and_user
|
|
|
|
# #region test_skips_existing_user [C:2] [TYPE Function]
|
|
@patch.dict(os.environ, {
|
|
"INITIAL_ADMIN_CREATE": "true", "INITIAL_ADMIN_USERNAME": "existing-admin", "INITIAL_ADMIN_PASSWORD": "s3cret!",
|
|
}, clear=False)
|
|
def test_skips_existing_user(self):
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value.filter.return_value.first.side_effect = [
|
|
MagicMock(name="Admin"), MagicMock(username="existing-admin"),
|
|
]
|
|
with patch("src.app.AuthSessionLocal", return_value=mock_db):
|
|
from src.app import ensure_initial_admin_user
|
|
ensure_initial_admin_user()
|
|
mock_db.add.assert_not_called()
|
|
mock_db.commit.assert_not_called()
|
|
# #endregion test_skips_existing_user
|
|
|
|
# #region test_rollback_on_error [C:2] [TYPE Function]
|
|
@patch.dict(os.environ, {
|
|
"INITIAL_ADMIN_CREATE": "true", "INITIAL_ADMIN_USERNAME": "admin", "INITIAL_ADMIN_PASSWORD": "s3cret!",
|
|
}, clear=False)
|
|
def test_rollback_on_error(self):
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value.filter.return_value.first.side_effect = Exception("DB crash")
|
|
with patch("src.app.AuthSessionLocal", return_value=mock_db):
|
|
from src.app import ensure_initial_admin_user
|
|
with pytest.raises(Exception, match="DB crash"):
|
|
ensure_initial_admin_user()
|
|
mock_db.rollback.assert_called_once()
|
|
mock_db.close.assert_called_once()
|
|
# #endregion test_rollback_on_error
|
|
|
|
# #region test_reuses_existing_admin_role [C:2] [TYPE Function]
|
|
@patch.dict(os.environ, {
|
|
"INITIAL_ADMIN_CREATE": "true", "INITIAL_ADMIN_USERNAME": "new-user", "INITIAL_ADMIN_PASSWORD": "s3cret!",
|
|
}, clear=False)
|
|
def test_reuses_existing_admin_role(self):
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value.filter.return_value.first.side_effect = [MagicMock(name="Admin"), None]
|
|
with patch("src.app.AuthSessionLocal", return_value=mock_db):
|
|
from src.app import ensure_initial_admin_user
|
|
ensure_initial_admin_user()
|
|
user_calls = [c for c in mock_db.add.call_args_list if c.args[0].__class__.__name__ == "User"]
|
|
role_calls = [c for c in mock_db.add.call_args_list if c.args[0].__class__.__name__ == "Role"]
|
|
assert len(user_calls) == 1
|
|
assert len(role_calls) == 0
|
|
mock_db.commit.assert_called_once()
|
|
# #endregion test_reuses_existing_admin_role
|
|
|
|
|
|
# =============================================================================
|
|
# lifespan
|
|
# =============================================================================
|
|
|
|
class TestLifespan:
|
|
"""Lifespan — startup/shutdown lifecycle."""
|
|
|
|
# #region test_lifespan_startup_initializes_deps [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async def test_lifespan_startup_initializes_deps(self):
|
|
from src.app import lifespan
|
|
mock_app = MagicMock(spec=FastAPI)
|
|
with (
|
|
patch("src.app.seed_trace_id") as mock_seed,
|
|
patch("src.app.ensure_encryption_key") as mock_enc_key,
|
|
patch("src.app.init_db") as mock_init_db,
|
|
patch("src.app.ensure_initial_admin_user") as mock_admin,
|
|
patch("src.app.get_scheduler_service") as mock_get_sched,
|
|
):
|
|
mock_scheduler = MagicMock()
|
|
mock_get_sched.return_value = mock_scheduler
|
|
async with lifespan(mock_app):
|
|
pass
|
|
mock_seed.assert_called_once()
|
|
mock_enc_key.assert_called_once()
|
|
mock_init_db.assert_called_once()
|
|
mock_admin.assert_called_once()
|
|
mock_scheduler.start.assert_called_once()
|
|
mock_scheduler.stop.assert_called_once()
|
|
# #endregion test_lifespan_startup_initializes_deps
|
|
|
|
# #region test_lifespan_stuck_run_cleanup [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async def test_lifespan_stuck_run_cleanup(self):
|
|
from src.app import lifespan
|
|
mock_app = MagicMock(spec=FastAPI)
|
|
mock_stuck = MagicMock()
|
|
mock_stuck.id = "run-stuck-1"
|
|
mock_stuck.status = "running"
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value.filter.return_value.all.return_value = [mock_stuck]
|
|
with (
|
|
patch("src.app.seed_trace_id"), patch("src.app.ensure_encryption_key"),
|
|
patch("src.app.init_db"), patch("src.app.ensure_initial_admin_user"),
|
|
patch("src.app.get_scheduler_service"),
|
|
patch("src.core.database.SessionLocal", return_value=mock_db),
|
|
):
|
|
async with lifespan(mock_app):
|
|
pass
|
|
assert mock_stuck.status == "FAIL"
|
|
assert "Force-stopped" in mock_stuck.summary
|
|
mock_db.commit.assert_called()
|
|
mock_db.close.assert_called()
|
|
# #endregion test_lifespan_stuck_run_cleanup
|
|
|
|
# #region test_lifespan_stuck_cleanup_handles_exception [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async def test_lifespan_stuck_cleanup_handles_exception(self):
|
|
from src.app import lifespan
|
|
mock_app = MagicMock(spec=FastAPI)
|
|
with (
|
|
patch("src.app.seed_trace_id"), patch("src.app.ensure_encryption_key"),
|
|
patch("src.app.init_db"), patch("src.app.ensure_initial_admin_user"),
|
|
patch("src.app.get_scheduler_service"),
|
|
patch("src.core.database.SessionLocal", side_effect=Exception("DB unreachable")),
|
|
):
|
|
async with lifespan(mock_app):
|
|
pass
|
|
# #endregion test_lifespan_stuck_cleanup_handles_exception
|
|
|
|
# #region test_lifespan_stuck_no_runs [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async def test_lifespan_stuck_no_runs(self):
|
|
from src.app import lifespan
|
|
mock_app = MagicMock(spec=FastAPI)
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value.filter.return_value.all.return_value = []
|
|
with (
|
|
patch("src.app.seed_trace_id"), patch("src.app.ensure_encryption_key"),
|
|
patch("src.app.init_db"), patch("src.app.ensure_initial_admin_user"),
|
|
patch("src.app.get_scheduler_service"),
|
|
patch("src.core.database.SessionLocal", return_value=mock_db),
|
|
):
|
|
async with lifespan(mock_app):
|
|
pass
|
|
mock_db.commit.assert_called_once()
|
|
mock_db.close.assert_called()
|
|
# #endregion test_lifespan_stuck_no_runs
|
|
|
|
# #region test_lifespan_shutdown_stops_scheduler [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async def test_lifespan_shutdown_stops_scheduler(self):
|
|
from src.app import lifespan
|
|
mock_app = MagicMock(spec=FastAPI)
|
|
mock_scheduler = MagicMock()
|
|
with (
|
|
patch("src.app.seed_trace_id"), patch("src.app.ensure_encryption_key"),
|
|
patch("src.app.init_db"), patch("src.app.ensure_initial_admin_user"),
|
|
patch("src.app.get_scheduler_service", return_value=mock_scheduler),
|
|
):
|
|
async with lifespan(mock_app):
|
|
pass
|
|
mock_scheduler.stop.assert_called_once()
|
|
# #endregion test_lifespan_shutdown_stops_scheduler
|
|
# #endregion Test.AppModule.Lifespan
|