"""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(): # The session takes its zone from the account on every request. s.add(User(id=uid, username=name, email=f"{name}@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=name, _csrf="test-token", login_at=time.time(), tz=tz) 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())