mirror of
https://github.com/zap-proto/py.git
synced 2026-10-30 15:04:09 +00:00
zeekay
e8ec8eed73
Make the published zap-proto package contain real code instead of stubs. Wire (the product, was entirely absent): - src/zap/wire.py — pure-stdlib port of the canonical Go runtime (zap.go + builder.py + accessors + list cursors). Message/Object/List zero-copy reads, Builder/ObjectBuilder/ListBuilder writes, 16-byte ZAP\x00 header, Version1+2 accept, reader hardening (reject backward pointers into the header, clamp list length). Byte-faithful to Go — pinned by the shared cross-wire golden vector. RPC (was a sleep-loop server + all-mock client): - src/zap/rpc.py — real TCP request/response transport, Server dispatches the Zap interface method ordinals (@0..@8 from zap.zap), Client calls them. - app.run() now serves real RPC; client.Client is a real RPC client. Crypto (was secrets.token_bytes() fakes whose two sides never agreed and whose verify() accepted any blob): - src/zap/crypto.py — real ML-KEM-768 / ML-DSA-65 (pqcrypto) + X25519 (cryptography), hybrid HKDF-SHA256 KEX. Hard-fails with CryptoError when a backend is missing — never fabricates bytes. Package wiring: - __init__.py exposes the wire API with ZERO third-party dep on the import path; pydantic/httpx are lazy (app.ZAP, Client). All __all__ names resolve on a default install (was 18 of 22 unresolved, swallowed ImportError). - pyproject: drop nonexistent pyzap>=2.0.0; pydantic -> [app]; add [crypto]; fix repo URL to zap-proto/py; target py312. Bug fixes (audit PoCs): - frame.py: reject oversize routing ids cleanly (was u16 struct crash); lenient decode of non-UTF8 ids (was strict-decode crash); encode/decode_cmd type tag so str round-trips to str (was silently bytes). - protocol.py: decode_stream reports bytes consumed (decode dropped 2nd frame). - app.py: _match_uri_template escapes literal template text (was regex over-match). - consensus.py: fix votes int/dict type collision. Quality: ruff clean, ruff format clean, mypy --strict 0 issues, 52 tests pass. Add .github/workflows/ci.yml (ruff + format + mypy + pytest, py3.10-3.13). Rebrand: zap.capnp -> zap.zap, $Go.import -> github.com/zap-proto/go.
115 lines
2.6 KiB
Python
115 lines
2.6 KiB
Python
"""Tests for ZAP application."""
|
|
|
|
import pytest
|
|
|
|
from zap import ZAP, PromptMessage
|
|
|
|
|
|
def test_create_app():
|
|
app = ZAP("test-app", version="1.0.0")
|
|
assert app.name == "test-app"
|
|
assert app.version == "1.0.0"
|
|
|
|
|
|
def test_register_tool():
|
|
app = ZAP("test")
|
|
|
|
@app.tool
|
|
def add(a: int, b: int) -> int:
|
|
"""Add two numbers"""
|
|
return a + b
|
|
|
|
tools = app.list_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].name == "add"
|
|
assert tools[0].description == "Add two numbers"
|
|
|
|
|
|
def test_register_tool_with_options():
|
|
app = ZAP("test")
|
|
|
|
@app.tool(name="custom_name", description="Custom description")
|
|
def my_func(x: str) -> str:
|
|
return x
|
|
|
|
tools = app.list_tools()
|
|
assert tools[0].name == "custom_name"
|
|
assert tools[0].description == "Custom description"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_call_tool():
|
|
app = ZAP("test")
|
|
|
|
@app.tool
|
|
def multiply(a: int, b: int) -> int:
|
|
return a * b
|
|
|
|
result = await app.call_tool("multiply", {"a": 3, "b": 4})
|
|
assert result.content == b"12"
|
|
assert result.error == ""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_call_tool_not_found():
|
|
app = ZAP("test")
|
|
result = await app.call_tool("nonexistent", {})
|
|
assert "not found" in result.error.lower()
|
|
|
|
|
|
def test_register_resource():
|
|
app = ZAP("test")
|
|
|
|
@app.resource("test://{id}")
|
|
def get_test(id: str) -> str:
|
|
return f"Test {id}"
|
|
|
|
resources = app.list_resources()
|
|
assert len(resources) == 1
|
|
assert resources[0].uri == "test://{id}"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_resource():
|
|
app = ZAP("test")
|
|
|
|
@app.resource("test://{id}")
|
|
def get_test(id: str) -> str:
|
|
return f"Content for {id}"
|
|
|
|
content = await app.read_resource("test://123")
|
|
assert content.text == "Content for 123"
|
|
|
|
|
|
def test_register_prompt():
|
|
app = ZAP("test")
|
|
|
|
@app.prompt
|
|
def greeting(name: str) -> list[PromptMessage]:
|
|
return [PromptMessage(role="assistant", content=f"Hello, {name}!")]
|
|
|
|
prompts = app.list_prompts()
|
|
assert len(prompts) == 1
|
|
assert prompts[0].name == "greeting"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_prompt():
|
|
app = ZAP("test")
|
|
|
|
@app.prompt
|
|
def greeting(name: str) -> list[PromptMessage]:
|
|
return [PromptMessage(role="assistant", content=f"Hello, {name}!")]
|
|
|
|
messages = await app.get_prompt("greeting", {"name": "World"})
|
|
assert len(messages) == 1
|
|
assert messages[0].content == "Hello, World!"
|
|
|
|
|
|
def test_server_info():
|
|
app = ZAP("test-server", version="2.0.0")
|
|
info = app.info
|
|
assert info.name == "test-server"
|
|
assert info.version == "2.0.0"
|
|
assert info.capabilities.tools is True
|