220 lines
10 KiB
Python
220 lines
10 KiB
Python
"""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", tz=None):
|
||
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, time_zone=tz))
|
||
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())
|
||
|
||
|
||
def test_an_account_sets_its_own_address_and_switches_and_an_admin_can_too(app):
|
||
async def _go():
|
||
viewer, vid = await _client(app, role="viewer")
|
||
page = await (await viewer.get("/account/")).get_data(as_text=True)
|
||
assert 'action="/account/email"' in page and "As for viewers (on)" in page
|
||
|
||
resp = await viewer.post("/account/email", headers=CSRF, form={
|
||
"email": "nope@oidc", "notify_alerts": "off", "notify_report": ""})
|
||
assert resp.status_code == 400 and "mail can reach" in await resp.get_data(as_text=True)
|
||
address = f"v-{vid[:8]}@example.com"
|
||
assert (await viewer.post("/account/email", headers=CSRF, form={
|
||
"email": address, "notify_alerts": "off", "notify_report": ""})).status_code == 302
|
||
me = await _user(app, vid)
|
||
assert (me.email, me.notify_alerts, me.notify_report) == (address, False, None)
|
||
assert [a for a, _ in await _audits(app, vid)] == ["account.email"]
|
||
|
||
admin, aid = await _client(app)
|
||
resp = await admin.post(f"/settings/users/{vid}/email", headers=CSRF,
|
||
form={"email": "ignored@example.com", "notify_alerts": "", "notify_report": "off"})
|
||
assert resp.status_code == 302
|
||
me = await _user(app, vid)
|
||
assert (me.email, me.notify_alerts, me.notify_report) == (address, None, False) # address kept
|
||
assert [a for a, _ in await _audits(app, aid)] == ["user.email_settings"]
|
||
|
||
page = await (await admin.get("/settings/notifications/")).get_data(as_text=True)
|
||
assert "Who gets email" in page and 'name="notify.alerts.viewer"' in page and "Also send to" in page
|
||
|
||
asyncio.run(_go())
|
||
|
||
|
||
def test_the_first_admin_to_open_notifications_sets_the_also_send_to_zone(app):
|
||
async def _go():
|
||
from steward.core.settings import get_setting, set_setting
|
||
|
||
async def _tz():
|
||
async with app.db_sessionmaker() as s:
|
||
return await get_setting(s, "smtp.tz")
|
||
|
||
async with app.db_sessionmaker() as s:
|
||
async with s.begin():
|
||
await set_setting(s, "smtp.tz", "")
|
||
for zone in ("America/Chicago", "Asia/Tokyo"):
|
||
admin, _ = await _client(app, tz=zone) # each request reads the zone from the account
|
||
assert (await admin.get("/settings/notifications/")).status_code == 200
|
||
assert await _tz() == "America/Chicago" # the second admin's zone changes nothing
|
||
|
||
asyncio.run(_go())
|