42 lines
1.2 KiB
Python
42 lines
1.2 KiB
Python
"""Streamable HTTP: stamp over /mcp."""
|
|
|
|
import socket
|
|
|
|
import anyio
|
|
import pytest
|
|
import uvicorn
|
|
from mcp import Client
|
|
|
|
from verae_mcp_timestamping_server.ledger import SHA256_EMPTY, MemoryLedger
|
|
from verae_mcp_timestamping_server.server import build_mcp
|
|
|
|
pytestmark = pytest.mark.anyio
|
|
|
|
|
|
def _free_port() -> int:
|
|
sock = socket.socket()
|
|
sock.bind(("127.0.0.1", 0))
|
|
port = sock.getsockname()[1]
|
|
sock.close()
|
|
return port
|
|
|
|
|
|
async def test_stamp_over_streamable_http():
|
|
mcp = build_mcp(MemoryLedger())
|
|
app = mcp.streamable_http_app()
|
|
port = _free_port()
|
|
server = uvicorn.Server(
|
|
uvicorn.Config(app, host="127.0.0.1", port=port, log_level="error")
|
|
)
|
|
async with anyio.create_task_group() as tg:
|
|
tg.start_soon(server.serve)
|
|
with anyio.fail_after(10):
|
|
while not server.started:
|
|
await anyio.sleep(0.05)
|
|
async with Client(f"http://127.0.0.1:{port}/mcp") as client:
|
|
result = await client.call_tool(
|
|
"stamp", {"sha256": SHA256_EMPTY, "scope": "local"}
|
|
)
|
|
assert result.is_error is not True
|
|
assert result.structured_content["receipts"][0]["hash"] == SHA256_EMPTY
|
|
server.should_exit = True
|