bvandeusenandClaude Opus 5.5
22090b02d8
Milestone 495 "Email goes to each account, in its own time zone", on the operator's ruling (System 156): notifications go to each user, times in their zone, defaults on, toggled per user or per role. - users.notify_alerts / notify_report (migration 0036): None follows the role's default, notify.<kind>.<role> in settings, all on to start. - mail.pick_recipients: active accounts that want the email and have an address mail can reach (an SSO placeholder like name@oidc is skipped and flagged), each in its own zone with no UTC beside it, then the Also send to addresses in their chosen zone with UTC beside. One copy per address, all over one SMTP connection. - Alerts, the weekly report and the test email (to you, in your zone, plus the typed addresses) send that way. The save-time smtp.tz capture is gone. - Notifications: a role × email table of defaults, who each email reaches and who can't be reached, and Also send to with its own zone, which starts as the first admin's to open the page. - Account page: your address and your two switches. Users: an Emailed column, and an admin sets an account's switches on its page. - The Alerts page counts the accounts it emails. zone_select macro and formatting.zones_by_region shared by Account and Notifications. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
201 lines
9.3 KiB
Python
201 lines
9.3 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"):
|
||
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())
|
||
|
||
|
||
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())
|