FabledSteward/tests/test_auth.py

import pytest
import bcrypt
from unittest.mock import AsyncMock, patch
from steward.models.users import User, UserRole


def make_user(role=UserRole.admin):
    return User(
        id="test-id",
        username="admin",
        email="admin@example.com",
        password_hash=bcrypt.hashpw(b"password", bcrypt.gensalt()).decode(),
        role=role,
        is_active=True,
        auth_source="local",
    )


async def with_csrf(client) -> dict:
    """Seed the session's forged-request token; returns the header to send."""
    async with client.session_transaction() as sess:
        sess["_csrf"] = "test-token"
    return {"X-CSRF-Token": "test-token"}


@pytest.mark.asyncio
async def test_login_page_returns_200(client):
    response = await client.get("/login")
    assert response.status_code == 200


@pytest.mark.asyncio
async def test_login_redirects_to_setup_when_no_users(client, app):
    with patch("steward.auth.routes.get_user_count", new=AsyncMock(return_value=0)):
        response = await client.get("/login")
    assert response.status_code == 302
    assert "/setup" in response.headers["Location"]


@pytest.mark.asyncio
async def test_login_success_redirects_to_dashboard(client, app):
    user = make_user()
    headers = await with_csrf(client)
    with patch("steward.auth.routes.get_user_count", new=AsyncMock(return_value=1)), \
         patch("steward.auth.routes.get_user_by_username", new=AsyncMock(return_value=user)):
        response = await client.post("/login", form={"username": "admin", "password": "password"},
                                     headers=headers)
    assert response.status_code == 302
    assert "/" in response.headers["Location"]


@pytest.mark.asyncio
async def test_login_failure_returns_400(client, app):
    user = make_user()
    headers = await with_csrf(client)
    with patch("steward.auth.routes.get_user_count", new=AsyncMock(return_value=1)), \
         patch("steward.auth.routes.get_user_by_username", new=AsyncMock(return_value=user)):
        response = await client.post("/login", form={"username": "admin", "password": "wrong"},
                                     headers=headers)
    assert response.status_code == 400


@pytest.mark.asyncio
async def test_require_role_redirects_unauthenticated(client):
    response = await client.get("/")
    assert response.status_code == 302
    assert "/login" in response.headers["Location"]


def _setup_restore_stubs(monkeypatch, tmp_path, schema="same", unopened=0):
    """First-run setup's restore (#5818) with the check and the restore
    stubbed: restoring.py and check.py have their own tests."""
    from steward.backup import job, restoring
    from steward.backup.check import AreaCheck, CheckReport, TableCheck
    seen = []

    async def check_file(app, path, name=""):
        return CheckReport(name=name, manifest={}, schema=schema, backup_heads=[], here_heads=[],
                           key_same=False, sealed=1, unopened=unopened,
                           areas=[AreaCheck("Settings", [TableCheck("app_settings", same=0)])])

    def start_restore(app, path, name, **kw):
        seen.append({"name": name, "data": path.read_bytes()} | kw)
        return True

    monkeypatch.setattr(restoring, "SAFETY_DIR", tmp_path)
    monkeypatch.setattr(job, "check_file", check_file)
    monkeypatch.setattr(restoring, "start_restore", start_restore)
    return seen


async def _send_backup(client, name="steward-20261015T030000Z.tar"):
    import io

    from quart.datastructures import FileStorage
    headers = await with_csrf(client)
    return await client.post("/setup/restore", headers=headers, files={
        "file": FileStorage(io.BytesIO(b"a backup"), filename=name)})


@pytest.mark.asyncio
async def test_setup_restores_a_backup_without_a_safety_backup(client, monkeypatch, tmp_path):
    seen = _setup_restore_stubs(monkeypatch, tmp_path, schema="older")
    with patch("steward.auth.routes.get_user_count", new=AsyncMock(return_value=0)):
        response = await _send_backup(client)
    assert response.status_code == 302 and response.headers["Location"].endswith("/setup")
    [started] = seen
    assert started["name"] == "steward-20261015T030000Z.tar" and started["data"] == b"a backup"
    assert started["safety"] is False and started["replace_all"] is True
    assert started["username"] == "first-run setup"


@pytest.mark.asyncio
async def test_setup_refuses_a_backup_it_cant_restore(client, monkeypatch, tmp_path):
    seen = _setup_restore_stubs(monkeypatch, tmp_path, schema="newer")
    with patch("steward.auth.routes.get_user_count", new=AsyncMock(return_value=0)):
        response = await _send_backup(client)
    assert response.status_code == 400 and not seen
    assert "update Steward, then restore it" in await response.get_data(as_text=True)
    assert not list((tmp_path / "incoming").iterdir())          # nothing left held


@pytest.mark.asyncio
async def test_setup_restore_is_gone_once_a_user_exists(client, monkeypatch, tmp_path):
    seen = _setup_restore_stubs(monkeypatch, tmp_path)
    with patch("steward.auth.routes.get_user_count", new=AsyncMock(return_value=1)):
        response = await _send_backup(client)
    assert response.status_code == 302 and "/login" in response.headers["Location"] and not seen


@pytest.mark.asyncio
async def test_setup_offers_a_restore(client):
    with patch("steward.auth.routes.get_user_count", new=AsyncMock(return_value=0)):
        page = await (await client.get("/setup")).get_data(as_text=True)
    assert 'action="/setup/restore"' in page and 'enctype="multipart/form-data"' in page