"""WP-270 operational UI inputs and bounded read-only invocation evidence.""" from __future__ import annotations import json import os from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any import httpx import pytest from fastapi import HTTPException from sqlalchemy import select from scripts.write_backup_status import publish from services.control_api.auth.models import AuthenticatedPrincipal, ManagementRole, PrincipalKind from services.control_api.persistence import Base, create_engine_from_url, session_factory from services.control_api.persistence.models import AuditEventRow from services.runtime import control_api from services.runtime.control_api import ( ToolInvokeRequest, invoke_read_only_tool, read_backup_status, ) from services.runtime.state_sync import sync_runtime_state ROOT = Path(__file__).resolve().parents[1] def test_backup_status_is_healthy_only_after_fresh_restore_verified_projection( tmp_path: Path, ) -> None: artifact = tmp_path / "hub.sql" artifact.write_bytes(b"bounded backup fixture") status = tmp_path / "status.json" document = publish(status, [artifact], restore_verified=True) result = read_backup_status( status, now=datetime.fromisoformat(str(document["completedAt"])) + timedelta(minutes=1) ) assert result == { "state": "healthy", "lastSuccessfulAt": document["completedAt"], "detail": "Latest backup and isolated restore verification succeeded.", "restoreVerified": True, "artifactCount": 1, } if os.name != "nt": assert status.stat().st_mode & 0o777 == 0o600 assert json.loads(status.read_text(encoding="utf-8"))["artifacts"][0]["sha256"] def test_backup_status_fails_closed_for_missing_stale_unverified_and_symlinked_input( tmp_path: Path, ) -> None: missing = read_backup_status(tmp_path / "missing.json") assert missing["state"] == "unknown" status = tmp_path / "status.json" status.write_text( json.dumps( { "schemaVersion": "1", "completedAt": (datetime.now(UTC) - timedelta(days=3)).isoformat(), "restoreVerified": True, "artifactCount": 1, } ), encoding="utf-8", ) assert read_backup_status(status)["state"] == "degraded" value = json.loads(status.read_text(encoding="utf-8")) value["completedAt"] = datetime.now(UTC).isoformat() value["restoreVerified"] = False status.write_text(json.dumps(value), encoding="utf-8") assert read_backup_status(status)["state"] == "degraded" link = tmp_path / "linked.json" try: link.symlink_to(status) except OSError: return assert read_backup_status(link)["state"] == "unknown" def test_backup_publisher_rejects_empty_and_symlinked_artifacts(tmp_path: Path) -> None: empty = tmp_path / "empty.sql" empty.touch() try: publish(tmp_path / "status.json", [empty], restore_verified=True) except ValueError as exc: assert "must not be empty" in str(exc) else: raise AssertionError("empty backup artifact was accepted") target = tmp_path / "target.sql" target.write_text("data", encoding="utf-8") link = tmp_path / "link.sql" try: link.symlink_to(target) except OSError: return try: publish(tmp_path / "status.json", [link], restore_verified=True) except ValueError as exc: assert "regular file" in str(exc) else: raise AssertionError("symlinked backup artifact was accepted") def test_management_console_invokes_only_the_cataloged_connector_and_audits( tmp_path: Path, monkeypatch: Any ) -> None: seen: list[httpx.Request] = [] def handler(request: httpx.Request) -> httpx.Response: seen.append(request) data: dict[str, Any] = ( {"status": "not-in-the-registered-output-contract"} if len(seen) == 2 else { "tenant": "fleet-1", "metrics": { "available": 4, "rented": 2, "cleaning": 1, "maintenance": 0, "blocked": 0, "openQualityIssues": 0, "pendingOrFailedWorkflows": 0, }, } ) structured = { "data": {**data, "unregisteredField": "must-not-cross-the-boundary"}, "meta": { "project": "mobilityops", "environment": "prod", "source": "mobilityops", "truncated": False, "nextCursor": None, "generatedAt": "2026-08-21T00:00:00+00:00", }, "warnings": ["x" * 501] if len(seen) == 3 else [], } return httpx.Response( 200, json={ "jsonrpc": "2.0", "id": "ignored", "result": { "content": [{"type": "text", "text": "healthy"}], "structuredContent": structured, "isError": False, }, }, ) monkeypatch.setattr( control_api, "evaluate_tool_readiness", lambda *_args, **_kwargs: {"ready": True}, ) monkeypatch.setenv("MCP_CLIENT_BINDINGS_JSON", "[]") engine = create_engine_from_url(f"sqlite:///{tmp_path / 'operations.db'}") Base.metadata.create_all(engine) sessions = session_factory(engine) issuer = "https://auth.example.test/application/o/hub/" with sessions() as session, session.begin(): sync_runtime_state(session, ROOT / "catalog", issuer) principal = AuthenticatedPrincipal( subject="operator@example.test", issuer=issuer, audience=("itworx-mcp-hub",), kind=PrincipalKind.MANAGEMENT_USER, expires_at=datetime.now(UTC) + timedelta(hours=1), issued_at=datetime.now(UTC), roles=frozenset({ManagementRole.OPERATOR}), ) request = ToolInvokeRequest( project="mobilityops", environment="prod", tool="mobilityops.operations.summary", arguments={}, ) result = invoke_read_only_tool( ROOT / "catalog", object(), # type: ignore[arg-type] object(), # type: ignore[arg-type] sessions, request, principal, b"a" * 32, transport=httpx.MockTransport(handler), ) assert result["ok"] is True assert result["structuredContent"]["tenant"] == "fleet-1" assert result["warnings"] == [] assert "content" not in result assert "unregisteredField" not in result["structuredContent"] assert len(seen) == 1 sent = seen[0] assert str(sent.url) == "http://connector-mobilityops:8080/mcp" assert sent.headers["x-itworx-project"] == "mobilityops" assert sent.headers["x-itworx-environment"] == "prod" assert sent.headers["x-itworx-identity-assertion"].startswith("v1:") assert "operator@example.test" not in sent.content.decode() with pytest.raises(HTTPException) as rejected: invoke_read_only_tool( ROOT / "catalog", object(), # type: ignore[arg-type] object(), # type: ignore[arg-type] sessions, request, principal, b"a" * 32, transport=httpx.MockTransport(handler), ) assert rejected.value.status_code == 502 assert len(seen) == 2 with pytest.raises(HTTPException) as malformed: invoke_read_only_tool( ROOT / "catalog", object(), # type: ignore[arg-type] object(), # type: ignore[arg-type] sessions, request, principal, b"a" * 32, transport=httpx.MockTransport(handler), ) assert malformed.value.status_code == 502 assert len(seen) == 3 with sessions() as session: audit = session.scalars(select(AuditEventRow).order_by(AuditEventRow.occurred_at)).all() assert [event.outcome for event in audit] == ["success", "error", "error"] assert {event.tool_name for event in audit} == {"mobilityops.operations.summary"} engine.dispose()