knafeh 1.1.0

QUIC-based RPC library with Python bindings
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249

"""E2E tests for Python async and sync APIs against a real QUIC server.

Tests use the default Protobuf codec. Two Python servers are started
(ports 14433 and 14434) — both use the Rust default codec (Protobuf)
with Python echo handlers.

Follows SOP categories A-G.
"""

import asyncio
import json
import time
import threading

import anyio

from knafeh._native import RpcServer, ServiceHandler, TlsConfig
from knafeh.client import Client, SyncClient


# ---------------------------------------------------------------------------
# Test server setup
# ---------------------------------------------------------------------------

_proto_server_addr = None
_proto_server_addr_2 = None
_servers_ready = threading.Event()


def _make_certs():
    import tempfile, subprocess, os
    d = tempfile.mkdtemp()
    cert = os.path.join(d, "cert.pem")
    key = os.path.join(d, "key.pem")
    subprocess.run([
        "openssl", "req", "-x509", "-newkey", "ec",
        "-pkeyopt", "ec_paramgen_curve:prime256v1",
        "-keyout", key, "-out", cert,
        "-days", "1", "-nodes", "-subj", "/CN=localhost",
    ], check=True, capture_output=True)
    return cert, key, d


_cert_path, _key_path, _cert_dir = _make_certs()


def _run_servers():
    """Start two Protobuf-default echo servers (ports 14433, 14434)."""
    global _proto_server_addr, _proto_server_addr_2
    tls = TlsConfig.server(_cert_path, _key_path)

    server1 = RpcServer("127.0.0.1:14433", tls)
    handler1 = ServiceHandler("echo")
    handler1.add_unary_handler("echo", lambda req: req)
    server1.add_service(handler1)

    server2 = RpcServer("127.0.0.1:14434", tls)
    handler2 = ServiceHandler("echo")
    handler2.add_unary_handler("echo", lambda req: req)
    server2.add_service(handler2)

    _proto_server_addr = "127.0.0.1:14433"
    _proto_server_addr_2 = "127.0.0.1:14434"
    _servers_ready.set()

    async def serve_both():
        async with anyio.create_task_group() as tg:
            tg.start_soon(_start, server1)
            tg.start_soon(_start, server2)

    asyncio.run(serve_both())


async def _start(server):
    await server.start()


_server_thread = threading.Thread(target=_run_servers, daemon=True)
_server_thread.start()
_servers_ready.wait(timeout=10)
time.sleep(1)


# ---------------------------------------------------------------------------
# SOP Category A: Unary — Sync API (against Protobuf server, default codec)
# ---------------------------------------------------------------------------

def test_sync_unary_echo():
    """A1: Basic echo with protobuf codec (default)."""
    with SyncClient(_proto_server_addr, tls=TlsConfig.client_insecure()) as client:
        body = b"hello protobuf"
        resp = client.call("echo/echo", body)
        assert resp == body


def test_sync_unary_multiple():
    """A1: Multiple sequential calls."""
    with SyncClient(_proto_server_addr, tls=TlsConfig.client_insecure()) as client:
        for i in range(10):
            body = f"msg-{i}".encode()
            resp = client.call("echo/echo", body)
            assert resp == body


def test_sync_unary_large_payload():
    """A4: ~100KB payload."""
    with SyncClient(_proto_server_addr, tls=TlsConfig.client_insecure()) as client:
        body = b"x" * 100_000
        resp = client.call("echo/echo", body)
        assert len(resp) == 100_000


# ---------------------------------------------------------------------------
# SOP Category A: Unary — Async API
# ---------------------------------------------------------------------------

def test_async_unary_echo():
    """A1: Basic echo async."""
    async def run():
        async with Client(_proto_server_addr, tls=TlsConfig.client_insecure()) as client:
            resp = await client.call("echo/echo", b"hello async proto")
            assert resp == b"hello async proto"
    anyio.run(run)


def test_async_unary_multiple():
    async def run():
        async with Client(_proto_server_addr, tls=TlsConfig.client_insecure()) as client:
            for i in range(10):
                body = f"msg-{i}".encode()
                resp = await client.call("echo/echo", body)
                assert resp == body
    anyio.run(run)


# ---------------------------------------------------------------------------
# SOP Category C: Connection reuse + concurrency
# ---------------------------------------------------------------------------

def test_async_concurrent_task_group():
    """C2: 50 concurrent calls via anyio TaskGroup."""
    async def run():
        async with Client(_proto_server_addr, tls=TlsConfig.client_insecure()) as client:
            results = []
            async def one():
                resp = await client.call("echo/echo", b"conc")
                results.append(resp)
            async with anyio.create_task_group() as tg:
                for _ in range(50):
                    tg.start_soon(one)
            assert len(results) == 50
    anyio.run(run)


def test_async_call_many():
    """G2: call_many — batched concurrent calls."""
    async def run():
        async with Client(_proto_server_addr, tls=TlsConfig.client_insecure()) as client:
            bodies = [f"item-{i}".encode() for i in range(50)]
            results = await client.call_many("echo/echo", bodies)
            assert len(results) == 50
            for i, resp in enumerate(results):
                assert resp == f"item-{i}".encode()
    anyio.run(run)


# ---------------------------------------------------------------------------
# SOP Category G: Load + Performance benchmarks
# ---------------------------------------------------------------------------

def test_sync_throughput():
    """G1: Sync sequential throughput (protobuf)."""
    with SyncClient(_proto_server_addr, tls=TlsConfig.client_insecure()) as client:
        body = b"bench"
        start = time.perf_counter()
        for _ in range(50):
            client.call("echo/echo", body)
        elapsed = time.perf_counter() - start
        print(f"\n  [sync protobuf] 50 sequential: {elapsed:.3f}s ({50/elapsed:.0f} rpc/s)")


def test_async_throughput_sequential():
    """G1: Async sequential throughput (protobuf)."""
    async def run():
        async with Client(_proto_server_addr, tls=TlsConfig.client_insecure()) as client:
            body = b"bench"
            start = time.perf_counter()
            for _ in range(50):
                await client.call("echo/echo", body)
            elapsed = time.perf_counter() - start
            print(f"\n  [async seq protobuf] 50 sequential: {elapsed:.3f}s ({50/elapsed:.0f} rpc/s)")
    anyio.run(run)


def test_async_throughput_concurrent():
    """G2: Async concurrent throughput via call_many (protobuf)."""
    async def run():
        async with Client(_proto_server_addr, tls=TlsConfig.client_insecure()) as client:
            body = b"bench"
            bodies = [body] * 50
            start = time.perf_counter()
            await client.call_many("echo/echo", bodies)
            elapsed = time.perf_counter() - start
            print(f"\n  [async call_many protobuf] 50 concurrent: {elapsed:.3f}s ({50/elapsed:.0f} rpc/s)")
    anyio.run(run)


def test_benchmark_comparison():
    """Full benchmark: Protobuf vs JSON, sync vs async."""
    N = 100

    proto_body = b"bench payload"
    json_body = json.dumps({"bench": True}).encode()

    # --- Protobuf codec (default) ---

    with SyncClient(_proto_server_addr, tls=TlsConfig.client_insecure()) as client:
        start = time.perf_counter()
        for _ in range(N):
            client.call("echo/echo", proto_body)
        sync_proto_time = time.perf_counter() - start

    async def async_seq_proto():
        async with Client(_proto_server_addr, tls=TlsConfig.client_insecure()) as client:
            start = time.perf_counter()
            for _ in range(N):
                await client.call("echo/echo", proto_body)
            return time.perf_counter() - start

    async_seq_proto_time = anyio.run(async_seq_proto)

    async def async_conc_proto():
        async with Client(_proto_server_addr, tls=TlsConfig.client_insecure()) as client:
            start = time.perf_counter()
            await client.call_many("echo/echo", [proto_body] * N)
            return time.perf_counter() - start

    async_conc_proto_time = anyio.run(async_conc_proto)

    # --- JSON codec (via JSON server) ---
    # Note: Python client uses default (protobuf) codec, so JSON server
    # comparison uses the protobuf client against JSON server — this tests
    # codec mismatch handling. Instead, we benchmark protobuf only for Python
    # since the Python client doesn't expose codec selection.

    print(f"\n  === Python E2E Benchmark — Protobuf codec ({N} calls) ===")
    print(f"  Sync sequential:    {sync_proto_time:.3f}s  ({N/sync_proto_time:.0f} rpc/s)")
    print(f"  Async sequential:   {async_seq_proto_time:.3f}s  ({N/async_seq_proto_time:.0f} rpc/s)")
    print(f"  Async call_many:    {async_conc_proto_time:.3f}s  ({N/async_conc_proto_time:.0f} rpc/s)")