FabledSteward/tests/integration/test_users_page.py

"""Integration: Settings › Users (#5336) against real Postgres. An admin adds
a local account, changes its role, sets its password, disables and deletes
it; their own account and an account that made records are refused, and
every change is audited."""
from __future__ import annotations

import asyncio
import json
import os
import time
import uuid
from datetime import datetime, timedelta, timezone

import bcrypt
import pytest

pytestmark = pytest.mark.integration
CSRF = {"X-CSRF-Token": "test-token"}


@pytest.fixture
def app():
    if not os.environ.get("STEWARD_DATABASE_URL"):
        pytest.skip("needs Postgres")
    from steward.app import create_app
    return create_app(testing=False)


async def _client(app, role="admin"):
    from steward.models.users import User, UserRole
    uid = str(uuid.uuid4())
    async with app.db_sessionmaker() as s:
        async with s.begin():
            s.add(User(id=uid, username=f"adm-{uid[:8]}", email=f"{uid}@test",
                       password_hash="x", role=UserRole(role), is_active=True))
    client = app.test_client()
    async with client.session_transaction() as sess:
        sess.update(user_id=uid, user_role=role, username=f"adm-{uid[:8]}",
                    _csrf="test-token", login_at=time.time())
    return client, uid


async def _user(app, user_id):
    from steward.models.users import User
    async with app.db_sessionmaker() as s:
        return await s.get(User, user_id)


async def _audits(app, uid):
    from sqlalchemy import select
    from steward.models.audit import AuditEvent
    async with app.db_sessionmaker() as s:
        rows = (await s.execute(select(AuditEvent).where(AuditEvent.user_id == uid)
                                .order_by(AuditEvent.timestamp))).scalars().all()
    return [(r.action, json.loads(r.detail_json or "{}")) for r in rows]


def test_an_admin_adds_changes_and_deletes_a_local_account(app):
    async def _go():
        client, me = await _client(app)
        name = f"new-{uuid.uuid4().hex[:8]}"
        resp = await client.post("/settings/users/new", headers=CSRF, form={
            "username": name, "email": f"{name}@test", "role": "viewer",
            "password": "short", "confirm": "short"})
        assert resp.status_code == 400 and "at least 8 characters" in await resp.get_data(as_text=True)

        resp = await client.post("/settings/users/new", headers=CSRF, form={
            "username": name, "email": f"{name}@test", "role": "viewer",
            "password": "correct horse", "confirm": "correct horse"})
        assert resp.status_code == 302
        uid = resp.headers["Location"].split("/settings/users/")[1].split("?")[0]
        user = await _user(app, uid)
        assert (user.username, user.role.value, user.auth_source) == (name, "viewer", "local")
        assert bcrypt.checkpw(b"correct horse", user.password_hash.encode())

        page = await (await client.get("/settings/users/")).get_data(as_text=True)
        assert f'href="/settings/users/{uid}"' in page and 'aria-label="Settings"' in page

        assert (await client.post(f"/settings/users/{uid}/role", headers=CSRF,
                                  form={"role": "operator"})).status_code == 302
        assert (await _user(app, uid)).role.value == "operator"

        resp = await client.post(f"/settings/users/{uid}/password", headers=CSRF,
                                 form={"password": "battery staple", "confirm": "battery stapler"})
        assert resp.status_code == 400 and "match" in await resp.get_data(as_text=True)
        assert (await client.post(f"/settings/users/{uid}/password", headers=CSRF,
                                  form={"password": "battery staple", "confirm": "battery staple"})).status_code == 302
        assert bcrypt.checkpw(b"battery staple", (await _user(app, uid)).password_hash.encode())

        assert (await client.post(f"/settings/users/{uid}/active", headers=CSRF)).status_code == 302
        assert not (await _user(app, uid)).is_active
        assert (await client.post(f"/settings/users/{uid}/active", headers=CSRF)).status_code == 302
        assert (await _user(app, uid)).is_active

        assert (await client.post(f"/settings/users/{uid}", headers=CSRF,
                                  form={"username": name + "-x", "email": f"{name}@test"})).status_code == 302
        assert (await _user(app, uid)).username == name + "-x"

        assert (await client.post(f"/settings/users/{uid}/delete", headers=CSRF)).status_code == 302
        assert await _user(app, uid) is None

        assert [a for a, _ in await _audits(app, me)] == [
            "user.created", "user.role_changed", "user.password_reset", "user.disabled",
            "user.enabled", "user.renamed", "user.deleted"]

    asyncio.run(_go())


def test_an_admin_cannot_disable_demote_or_delete_themselves(app):
    async def _go():
        client, me = await _client(app)
        for path, form, says in ((f"/settings/users/{me}/active", {}, "your own account"),
                                 (f"/settings/users/{me}/role", {"role": "viewer"}, "your own role"),
                                 (f"/settings/users/{me}/delete", {}, "your own account")):
            resp = await client.post(path, headers=CSRF, form=form)
            assert resp.status_code == 400 and says in await resp.get_data(as_text=True), path
        me_now = await _user(app, me)
        assert me_now.is_active and me_now.role.value == "admin"

    asyncio.run(_go())


def test_an_account_that_made_records_is_disabled_not_deleted(app):
    from steward.models.alerts import MaintenanceWindow

    async def _go():
        client, _ = await _client(app)
        _, other = await _client(app, role="operator")
        now = datetime.now(timezone.utc)
        async with app.db_sessionmaker() as s:
            async with s.begin():
                # Over already, so it silences no other test's alerts.
                s.add(MaintenanceWindow(name="patching", start_at=now - timedelta(days=2),
                                        end_at=now - timedelta(days=1), created_by=other))
        page = await (await client.get(f"/settings/users/{other}")).get_data(as_text=True)
        assert "1 maintenance window" in page and "Delete account</button>" not in page
        resp = await client.post(f"/settings/users/{other}/delete", headers=CSRF)
        assert resp.status_code == 400 and "Disable the account instead" in await resp.get_data(as_text=True)
        assert await _user(app, other) is not None

    asyncio.run(_go())


def test_a_provider_account_s_role_is_not_edited_here(app):
    from steward.models.users import User, UserRole

    async def _go():
        client, _ = await _client(app)
        uid = str(uuid.uuid4())
        async with app.db_sessionmaker() as s:
            async with s.begin():
                s.add(User(id=uid, username=f"sso-{uid[:8]}", email=f"{uid}@oidc", password_hash="x",
                           role=UserRole.viewer, auth_source="oidc", external_id=uid))
        resp = await client.post(f"/settings/users/{uid}/role", headers=CSRF, form={"role": "admin"})
        assert resp.status_code == 400 and "at each sign-in" in await resp.get_data(as_text=True)
        resp = await client.post(f"/settings/users/{uid}/password", headers=CSRF,
                                 form={"password": "long enough", "confirm": "long enough"})
        assert resp.status_code == 400
        assert (await _user(app, uid)).role.value == "viewer"

    asyncio.run(_go())


def test_only_an_admin_reaches_users(app):
    async def _go():
        client, _ = await _client(app, role="operator")
        assert (await client.get("/settings/users/")).status_code == 403
        assert (await client.post("/settings/users/new", headers=CSRF, form={})).status_code == 403

    asyncio.run(_go())