From c77ed2865f29e709f778f1ef18d9dbe5ffbfadfc Mon Sep 17 00:00:00 2001 From: Yossi Ovadia Date: Tue, 28 Jul 2026 13:41:26 -0700 Subject: [PATCH 1/2] test(sdk): add Anthropic Messages integration tests against llm-katan MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Exercise the full Anthropic filter chain (anthropic_messages_format, anthropic_validate, anthropic_messages_protocol, token_count) through Praxis against llm-katan's Anthropic Messages endpoint. Covers non-streaming, streaming SSE, multi-turn, system messages, and token usage extraction — none of which had SDK-level integration coverage against a live Anthropic-format backend. Skips gracefully when llm-katan is unreachable. Relates to #600 Signed-off-by: Yossi Ovadia --- .../test_anthropic_messages_llmkatan.py | 311 ++++++++++++++++++ 1 file changed, 311 insertions(+) create mode 100644 tests/integration/sdk/anthropic/test_anthropic_messages_llmkatan.py diff --git a/tests/integration/sdk/anthropic/test_anthropic_messages_llmkatan.py b/tests/integration/sdk/anthropic/test_anthropic_messages_llmkatan.py new file mode 100644 index 0000000000..acccd86128 --- /dev/null +++ b/tests/integration/sdk/anthropic/test_anthropic_messages_llmkatan.py @@ -0,0 +1,311 @@ +#!/usr/bin/env python3 +# /// script +# requires-python = ">=3.11" +# dependencies = [ +# "anthropic>=0.40", +# "pytest>=8.0", +# ] +# /// +""" +Anthropic Messages API integration tests against llm-katan. + +Starts a Praxis proxy with the Anthropic passthrough pipeline backed +by llm-katan's echo backend, then exercises non-streaming, streaming, +multi-turn, and usage extraction using the official Anthropic Python SDK. + +Usage: + cargo build -p praxis-ai-proxy + uv run tests/integration/sdk/anthropic/test_anthropic_messages_llmkatan.py -s -v + +Environment variables: + LLM_KATAN_BASE_URL llm-katan base URL (default: https://3-147-232-199.sslip.io) + LLM_KATAN_API_KEY llm-katan Anthropic API key (default: llm-katan-anthropic-key) + LLM_KATAN_MODEL model name (default: llm-katan-echo) + PRAXIS_AI_BIN path to praxis-ai binary (auto-detected if unset) +""" + +import os +import signal +import socket +import subprocess +import sys +import tempfile +import time +from urllib.parse import urlparse + +import pytest +from anthropic import Anthropic + +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- + +LLM_KATAN_BASE_URL = os.environ.get( + "LLM_KATAN_BASE_URL", "https://3-147-232-199.sslip.io" +) +LLM_KATAN_API_KEY = os.environ.get( + "LLM_KATAN_API_KEY", "llm-katan-anthropic-key" +) +LLM_KATAN_MODEL = os.environ.get("LLM_KATAN_MODEL", "llm-katan-echo") +PRAXIS_AI_BIN = os.environ.get("PRAXIS_AI_BIN") + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _find_binary() -> str: + if PRAXIS_AI_BIN: + if os.path.isfile(PRAXIS_AI_BIN): + return PRAXIS_AI_BIN + raise FileNotFoundError( + f"PRAXIS_AI_BIN={PRAXIS_AI_BIN!r} not found" + ) + for candidate in ["target/debug/praxis-ai", "target/release/praxis-ai"]: + if os.path.isfile(candidate): + return candidate + raise FileNotFoundError( + "praxis-ai binary not found — run `cargo build -p praxis-ai-proxy` first" + ) + + +def _llm_katan_endpoint() -> str: + parsed = urlparse(LLM_KATAN_BASE_URL) + host = parsed.hostname or "3-147-232-199.sslip.io" + port = parsed.port or (443 if parsed.scheme == "https" else 80) + return f"{host}:{port}" + + +def _llm_katan_host() -> str: + parsed = urlparse(LLM_KATAN_BASE_URL) + return parsed.hostname or "3-147-232-199.sslip.io" + + +def _llm_katan_reachable() -> bool: + try: + parsed = urlparse(LLM_KATAN_BASE_URL) + host = parsed.hostname or "3-147-232-199.sslip.io" + port = parsed.port or (443 if parsed.scheme == "https" else 80) + with socket.create_connection((host, port), timeout=5): + return True + except OSError: + return False + + +def _write_config(proxy_port: int) -> str: + config = f"""\ +listeners: + - name: test + address: "127.0.0.1:{proxy_port}" + filter_chains: [anthropic] + +filter_chains: + - name: anthropic + filters: + - filter: anthropic_messages_format + on_invalid: continue + - filter: anthropic_validate + - filter: anthropic_messages_protocol + default_version: "2023-06-01" + - filter: token_count + provider: anthropic + - filter: token_usage_headers + - filter: router + routes: + - path_prefix: "/" + cluster: llm-katan + - filter: load_balancer + clusters: + - name: llm-katan + endpoints: + - "{_llm_katan_endpoint()}" + tls: + sni: "{_llm_katan_host()}" +""" + fd, path = tempfile.mkstemp(suffix=".yaml") + with os.fdopen(fd, "w") as f: + f.write(config) + return path + + +def _wait_for_proxy(port: int, timeout: float = 30.0) -> None: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + try: + with socket.create_connection(("127.0.0.1", port), timeout=0.5): + return + except OSError: + time.sleep(0.2) + raise TimeoutError(f"Praxis did not start within {timeout}s") + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="session") +def praxis_proxy(tmp_path_factory, request): + """Start a Praxis proxy backed by llm-katan for the test session.""" + if not _llm_katan_reachable(): + pytest.skip( + f"llm-katan not reachable at {LLM_KATAN_BASE_URL} — skipping" + ) + + port = _free_port() + config_path = _write_config(port) + binary = _find_binary() + + log_dir = tmp_path_factory.mktemp("anthropic-logs") + log_path = str(log_dir / "praxis.log") + log_file = open(log_path, "w") + started = False + + proc = subprocess.Popen( + [binary, "-c", config_path], + stdout=log_file, + stderr=subprocess.STDOUT, + ) + try: + _wait_for_proxy(port) + started = True + yield port + finally: + proc.send_signal(signal.SIGINT) + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait() + log_file.close() + if not started or request.session.testsfailed > 0: + with open(log_path) as f: + print( + f"\n=== Praxis logs ===\n{f.read()}", + file=sys.stderr, + ) + os.unlink(config_path) + + +@pytest.fixture(scope="session") +def anthropic_client(praxis_proxy): + """Return an Anthropic client pointed at the local Praxis proxy.""" + return Anthropic( + base_url=f"http://127.0.0.1:{praxis_proxy}", + api_key=LLM_KATAN_API_KEY, + max_retries=0, + timeout=180, + ) + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +class TestAnthropicMessagesLLMKatan: + """Integration tests for Anthropic Messages API through Praxis.""" + + def test_non_streaming_basic(self, anthropic_client): + response = anthropic_client.messages.create( + model=LLM_KATAN_MODEL, + max_tokens=128, + messages=[{"role": "user", "content": "Hello from praxis-ai"}], + ) + + assert response.type == "message" + assert response.role == "assistant" + assert len(response.content) > 0 + assert response.content[0].type == "text" + assert len(response.content[0].text) > 0 + assert response.stop_reason == "end_turn" + + def test_non_streaming_with_system(self, anthropic_client): + response = anthropic_client.messages.create( + model=LLM_KATAN_MODEL, + max_tokens=128, + system="You are a helpful assistant.", + messages=[{"role": "user", "content": "What are you?"}], + ) + + assert response.type == "message" + assert response.role == "assistant" + assert len(response.content) > 0 + assert response.content[0].type == "text" + assert len(response.content[0].text) > 0 + + def test_streaming_basic(self, anthropic_client): + event_types = set() + + with anthropic_client.messages.stream( + model=LLM_KATAN_MODEL, + max_tokens=128, + messages=[{"role": "user", "content": "Stream test"}], + ) as stream: + for event in stream: + event_types.add(event.type) + + assert "message_start" in event_types, ( + f"should see message_start; saw: {event_types}" + ) + assert "content_block_delta" in event_types, ( + f"should see content_block_delta; saw: {event_types}" + ) + assert "message_stop" in event_types, ( + f"should see message_stop; saw: {event_types}" + ) + + def test_streaming_collects_full_text(self, anthropic_client): + collected = "" + + with anthropic_client.messages.stream( + model=LLM_KATAN_MODEL, + max_tokens=128, + messages=[{"role": "user", "content": "Streaming text collection"}], + ) as stream: + for text in stream.text_stream: + collected += text + + assert len(collected) > 0, "streamed text should not be empty" + + def test_multi_turn(self, anthropic_client): + response = anthropic_client.messages.create( + model=LLM_KATAN_MODEL, + max_tokens=128, + messages=[ + {"role": "user", "content": "My name is Alice."}, + {"role": "assistant", "content": "Hello Alice!"}, + {"role": "user", "content": "What is my name?"}, + ], + ) + + assert response.type == "message" + assert response.role == "assistant" + assert len(response.content) > 0 + assert response.content[0].type == "text" + assert len(response.content[0].text) > 0 + + def test_usage_present(self, anthropic_client): + response = anthropic_client.messages.create( + model=LLM_KATAN_MODEL, + max_tokens=128, + messages=[{"role": "user", "content": "Usage test"}], + ) + + assert response.usage is not None, "usage should be present" + assert response.usage.input_tokens > 0, ( + f"input_tokens should be > 0; got {response.usage.input_tokens}" + ) + assert response.usage.output_tokens > 0, ( + f"output_tokens should be > 0; got {response.usage.output_tokens}" + ) + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-v"] + sys.argv[1:])) From 158f8f555e6bff2b8ba9398ea2aba7efa335b932 Mon Sep 17 00:00:00 2001 From: Yossi Ovadia Date: Wed, 29 Jul 2026 08:38:10 -0700 Subject: [PATCH 2/2] fix: address praxis-bot review findings - Remove hardcoded llm-katan IP default; require LLM_KATAN_BASE_URL env var and skip when unset - Parse URL once at module level instead of redundantly in each helper - Add error-path test (malformed JSON rejected by anthropic_validate) - Add inline config comment explaining why no example config is loaded Relates to #600 Signed-off-by: Yossi Ovadia --- .../test_anthropic_messages_llmkatan.py | 64 +++++++++++++------ 1 file changed, 45 insertions(+), 19 deletions(-) diff --git a/tests/integration/sdk/anthropic/test_anthropic_messages_llmkatan.py b/tests/integration/sdk/anthropic/test_anthropic_messages_llmkatan.py index acccd86128..35537fafc0 100644 --- a/tests/integration/sdk/anthropic/test_anthropic_messages_llmkatan.py +++ b/tests/integration/sdk/anthropic/test_anthropic_messages_llmkatan.py @@ -3,6 +3,7 @@ # requires-python = ">=3.11" # dependencies = [ # "anthropic>=0.40", +# "httpx>=0.27", # "pytest>=8.0", # ] # /// @@ -18,7 +19,7 @@ uv run tests/integration/sdk/anthropic/test_anthropic_messages_llmkatan.py -s -v Environment variables: - LLM_KATAN_BASE_URL llm-katan base URL (default: https://3-147-232-199.sslip.io) + LLM_KATAN_BASE_URL llm-katan base URL (required; skips when unset) LLM_KATAN_API_KEY llm-katan Anthropic API key (default: llm-katan-anthropic-key) LLM_KATAN_MODEL model name (default: llm-katan-echo) PRAXIS_AI_BIN path to praxis-ai binary (auto-detected if unset) @@ -40,9 +41,7 @@ # Configuration # --------------------------------------------------------------------------- -LLM_KATAN_BASE_URL = os.environ.get( - "LLM_KATAN_BASE_URL", "https://3-147-232-199.sslip.io" -) +LLM_KATAN_BASE_URL = os.environ.get("LLM_KATAN_BASE_URL") LLM_KATAN_API_KEY = os.environ.get( "LLM_KATAN_API_KEY", "llm-katan-anthropic-key" ) @@ -75,23 +74,20 @@ def _find_binary() -> str: ) -def _llm_katan_endpoint() -> str: - parsed = urlparse(LLM_KATAN_BASE_URL) - host = parsed.hostname or "3-147-232-199.sslip.io" - port = parsed.port or (443 if parsed.scheme == "https" else 80) - return f"{host}:{port}" - - -def _llm_katan_host() -> str: +def _parse_llm_katan_url() -> tuple[str, int, bool]: + """Parse LLM_KATAN_BASE_URL into (host, port, uses_tls).""" parsed = urlparse(LLM_KATAN_BASE_URL) - return parsed.hostname or "3-147-232-199.sslip.io" + tls = parsed.scheme == "https" + host = parsed.hostname or "127.0.0.1" + port = parsed.port or (443 if tls else 80) + return host, port, tls def _llm_katan_reachable() -> bool: + if not LLM_KATAN_BASE_URL: + return False try: - parsed = urlparse(LLM_KATAN_BASE_URL) - host = parsed.hostname or "3-147-232-199.sslip.io" - port = parsed.port or (443 if parsed.scheme == "https" else 80) + host, port, _ = _parse_llm_katan_url() with socket.create_connection((host, port), timeout=5): return True except OSError: @@ -99,6 +95,12 @@ def _llm_katan_reachable() -> bool: def _write_config(proxy_port: int) -> str: + host, port, tls = _parse_llm_katan_url() + tls_block = f""" + tls: + sni: "{host}" """ if tls else "" + # Inline config: no matching example config exists for the + # format+validate+protocol+token_count pipeline used here. config = f"""\ listeners: - name: test @@ -124,9 +126,7 @@ def _write_config(proxy_port: int) -> str: clusters: - name: llm-katan endpoints: - - "{_llm_katan_endpoint()}" - tls: - sni: "{_llm_katan_host()}" + - "{host}:{port}"{tls_block} """ fd, path = tempfile.mkstemp(suffix=".yaml") with os.fdopen(fd, "w") as f: @@ -153,6 +153,8 @@ def _wait_for_proxy(port: int, timeout: float = 30.0) -> None: @pytest.fixture(scope="session") def praxis_proxy(tmp_path_factory, request): """Start a Praxis proxy backed by llm-katan for the test session.""" + if not LLM_KATAN_BASE_URL: + pytest.skip("LLM_KATAN_BASE_URL not set — skipping") if not _llm_katan_reachable(): pytest.skip( f"llm-katan not reachable at {LLM_KATAN_BASE_URL} — skipping" @@ -307,5 +309,29 @@ def test_usage_present(self, anthropic_client): ) + def test_malformed_json_rejected(self, anthropic_client, praxis_proxy): + """Verify anthropic_validate rejects malformed JSON end-to-end.""" + import httpx + + resp = httpx.post( + f"http://127.0.0.1:{praxis_proxy}/v1/messages", + content=b"not json {{{", + headers={ + "content-type": "application/json", + "anthropic-version": "2023-06-01", + "x-api-key": LLM_KATAN_API_KEY, + }, + timeout=30, + ) + + assert resp.status_code == 400, ( + f"malformed JSON should be rejected with 400; got {resp.status_code}" + ) + body = resp.json() + assert body["error"]["type"] == "invalid_request_error", ( + f"error type should be invalid_request_error; got {body}" + ) + + if __name__ == "__main__": sys.exit(pytest.main([__file__, "-v"] + sys.argv[1:]))