bvandeusenandClaude Opus 5.5
329aa15e3d
A change Steward can't see happen (a config edited, a disk swapped) is recorded by whoever made it: what was done, its area, when (in the reader's own zone), the hosts, what it was done to and a detail. Its author or an admin edits or deletes it; recording, editing and deleting are audited and those audit rows stay off the timeline. log_change records one as "<user> via MCP", editable by the token's user. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
166 lines
8.1 KiB
Python
166 lines
8.1 KiB
Python
"""Integration: a change recorded by hand (#6350), from the page and through
|
|
MCP log_change. The form reads its time in the reader's own zone; only the
|
|
author or an admin edits or deletes one; each step is audited and the audit
|
|
rows stay off the timeline. Requires STEWARD_DATABASE_URL.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import time
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
@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 _user(app, name: str, role: str, tz: str | None = 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=name, email=f"{name}@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=name, _csrf="test-token", login_at=time.time(),
|
|
**({"tz": tz} if tz else {}))
|
|
return client, uid
|
|
|
|
|
|
async def _host(app, name: str):
|
|
from steward.models.hosts import Host
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
host = Host(name=name, address="10.8.8.8")
|
|
s.add(host)
|
|
return host
|
|
|
|
|
|
async def _changes(app, host: str):
|
|
from sqlalchemy import select
|
|
from steward.models.timeline import TimelineEntry
|
|
async with app.db_sessionmaker() as s:
|
|
return (await s.execute(select(TimelineEntry).where(TimelineEntry.hosts.contains([host]))
|
|
.order_by(TimelineEntry.at))).scalars().all()
|
|
|
|
|
|
async def _audited(app, entry_id: str) -> list[str]:
|
|
from sqlalchemy import select
|
|
from steward.models.audit import AuditEvent
|
|
async with app.db_sessionmaker() as s:
|
|
return list((await s.execute(select(AuditEvent.action).where(AuditEvent.entity_id == entry_id)
|
|
.order_by(AuditEvent.timestamp))).scalars())
|
|
|
|
|
|
def test_a_change_recorded_on_the_page_is_edited_by_its_author_and_deleted_by_an_admin(app):
|
|
async def _go():
|
|
tag = uuid.uuid4().hex[:8]
|
|
name = f"tl-ch-{tag}"
|
|
host = await _host(app, name)
|
|
author, author_id = await _user(app, f"ch-a-{tag}", "operator", tz="America/New_York")
|
|
other, _ = await _user(app, f"ch-o-{tag}", "operator")
|
|
admin, _ = await _user(app, f"ch-ad-{tag}", "admin")
|
|
viewer, _ = await _user(app, f"ch-v-{tag}", "viewer")
|
|
form = {"csrf_token": "test-token", "summary": "Moved Postgres data to the new SSD", "area": "storage",
|
|
"hosts": name, "subject": "postgres", "note": "Old disk kept as a spare.",
|
|
"at": "2026-10-20T08:00", "back": f"/hosts/{host.id}"}
|
|
|
|
# The form opens from the host's card with the host ticked; a viewer can't open it.
|
|
page = await (await author.get(f"/timeline/changes/new?host={name}&back=/hosts/{host.id}")).get_data(as_text=True)
|
|
assert f'value="{name}" checked' in page and f'value="/hosts/{host.id}"' in page
|
|
assert (await viewer.get("/timeline/changes/new")).status_code == 403
|
|
|
|
# Refused, with the form kept: an unknown host writes nothing.
|
|
bad = await author.post("/timeline/changes", form={**form, "hosts": "no-such-host"})
|
|
assert bad.status_code == 400
|
|
text = await bad.get_data(as_text=True)
|
|
assert "No host is named no-such-host" in text and "Moved Postgres data" in text
|
|
|
|
res = await author.post("/timeline/changes", form=form)
|
|
assert res.status_code == 302 and res.headers["Location"] == f"/hosts/{host.id}"
|
|
[entry] = await _changes(app, name)
|
|
assert (entry.type, entry.area, entry.subject, entry.actor) == (
|
|
"change.recorded", "storage", "postgres", f"ch-a-{tag}")
|
|
assert entry.at == datetime(2026, 10, 20, 12, 0, tzinfo=timezone.utc) # 08:00 in New York
|
|
assert entry.detail == {"note": "Old disk kept as a spare.", "author_id": author_id}
|
|
|
|
# Its author sees Edit on the timeline; another operator does not, and can't edit it.
|
|
mine = await (await author.get(f"/timeline/?host={name}")).get_data(as_text=True)
|
|
assert f"/timeline/changes/{entry.id}/edit" in mine
|
|
theirs = await (await other.get(f"/timeline/?host={name}")).get_data(as_text=True)
|
|
assert f"/timeline/changes/{entry.id}/edit" not in theirs
|
|
assert (await other.get(f"/timeline/changes/{entry.id}/edit")).status_code == 403
|
|
assert (await other.post(f"/timeline/changes/{entry.id}", form=form)).status_code == 403
|
|
|
|
edit = await (await author.get(f"/timeline/changes/{entry.id}/edit")).get_data(as_text=True)
|
|
assert 'value="2026-10-20T08:00"' in edit # shown in their zone
|
|
await author.post(f"/timeline/changes/{entry.id}",
|
|
form={**form, "summary": "Moved Postgres data to the SSD", "area": "hardware"})
|
|
[entry] = await _changes(app, name)
|
|
assert (entry.summary, entry.area) == ("Moved Postgres data to the SSD", "hardware")
|
|
assert entry.detail["author_id"] == author_id # still theirs
|
|
|
|
res = await admin.post(f"/timeline/changes/{entry.id}/delete",
|
|
form={"csrf_token": "test-token", "back": "/timeline/"})
|
|
assert res.status_code == 302 and await _changes(app, name) == []
|
|
assert await _audited(app, entry.id) == [
|
|
"timeline.change_recorded", "timeline.change_edited", "timeline.change_deleted"]
|
|
# The audit rows are not on the timeline: the change was, as itself.
|
|
from sqlalchemy import select
|
|
from steward.models.timeline import TimelineEntry
|
|
async with app.db_sessionmaker() as s:
|
|
leaked = (await s.execute(select(TimelineEntry).where(
|
|
TimelineEntry.summary.contains("Moved Postgres data"),
|
|
TimelineEntry.actor.in_([f"ch-a-{tag}", f"ch-ad-{tag}"])))).scalars().all()
|
|
assert leaked == []
|
|
|
|
asyncio.run(_go())
|
|
|
|
|
|
def test_log_change_records_a_change_through_mcp(app):
|
|
from steward.core.api_tokens import Caller
|
|
from steward.mcp import _context
|
|
from steward.mcp.tools.timeline import get_timeline, log_change
|
|
from steward.models.users import UserRole
|
|
|
|
async def _go():
|
|
tag = uuid.uuid4().hex[:8]
|
|
name = f"tl-mc-{tag}"
|
|
await _host(app, name)
|
|
_, uid = await _user(app, f"mc-{tag}", "operator")
|
|
_context._app_ctx.set(app)
|
|
_context._caller_ctx.set(Caller(user_id=uid, username=f"mc-{tag}", role=UserRole.operator, token_id=7))
|
|
|
|
when = (datetime.now(timezone.utc) - timedelta(days=1)).replace(microsecond=0)
|
|
out = await log_change("Raised the swap to 8 GiB", "Host OS", hosts=name, subject="/swapfile",
|
|
detail="It ran out during the backup.",
|
|
at=when.isoformat().replace("+00:00", "Z"))
|
|
assert out["type"] == "change.recorded" and out["kind"] == "change" and out["area"] == "host_os"
|
|
assert out["by"] == f"mc-{tag} via MCP" and out["hosts"] == [name]
|
|
assert out["at"] == when.isoformat().replace("+00:00", "Z") and out["note"] == "It ran out during the backup."
|
|
[entry] = await _changes(app, name)
|
|
assert entry.detail["author_id"] == uid # the token's user can edit it
|
|
assert await _audited(app, entry.id) == ["timeline.change_recorded"]
|
|
seen = await get_timeline(host=name, range="30d")
|
|
assert [e["entry_id"] for e in seen["entries"]] == [entry.id]
|
|
|
|
with pytest.raises(ValueError, match="list_hosts"):
|
|
await log_change("x", "storage", hosts=["nope"])
|
|
with pytest.raises(ValueError, match="area 'plumbing'"):
|
|
await log_change("x", "plumbing")
|
|
with pytest.raises(ValueError, match="ISO-8601"):
|
|
await log_change("x", "storage", at="yesterday")
|
|
|
|
asyncio.run(_go())
|