78 lines
2.5 KiB
Python
Executable file
78 lines
2.5 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Minimal llama-server health double for lifecycle integration tests."""
|
|
|
|
import argparse
|
|
import http.server
|
|
import json
|
|
import os
|
|
import time
|
|
|
|
|
|
class Handler(http.server.BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
if self.path == "/health":
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(b'{"status":"ok"}')
|
|
else:
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
|
|
def do_POST(self):
|
|
if self.path != "/v1/chat/completions":
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
return
|
|
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
try:
|
|
request = json.loads(self.rfile.read(length))
|
|
assert request["stream"] is True
|
|
assert request["messages"][0]["content"] == 'quote " slash \\\nline'
|
|
except (AssertionError, KeyError, json.JSONDecodeError):
|
|
self.send_response(400)
|
|
self.end_headers()
|
|
return
|
|
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/event-stream")
|
|
self.send_header("Cache-Control", "no-cache")
|
|
self.end_headers()
|
|
self.wfile.flush()
|
|
|
|
time.sleep(0.08)
|
|
self.wfile.write(
|
|
b'data: {"choices":[{"delta":{"role":"assistant","content":""}}]}\n\n'
|
|
)
|
|
self.wfile.flush()
|
|
time.sleep(0.12)
|
|
self.wfile.write(
|
|
b'data: {"choices":[{"delta":{"reasoning_content":"think"}}]}\n\n'
|
|
)
|
|
self.wfile.flush()
|
|
time.sleep(0.08)
|
|
self.wfile.write(
|
|
b'data: {"choices":[{"delta":{"content":"answer"},"finish_reason":null}]}\n\n'
|
|
)
|
|
self.wfile.write(
|
|
b'data: {"choices":[{"delta":{},"finish_reason":"stop"}],'
|
|
b'"usage":{"prompt_tokens":3,"completion_tokens":2,"total_tokens":5}}\n\n'
|
|
)
|
|
self.wfile.write(b"data: [DONE]\n\n")
|
|
self.wfile.flush()
|
|
|
|
def log_message(self, _format, *_args):
|
|
pass
|
|
|
|
|
|
parser = argparse.ArgumentParser(add_help=False)
|
|
parser.add_argument("--port", type=int, required=True)
|
|
args, _ = parser.parse_known_args()
|
|
|
|
pid_file = os.environ.get("GEM_FAKE_SERVER_PID_FILE")
|
|
if pid_file:
|
|
with open(pid_file, "w", encoding="utf-8") as handle:
|
|
handle.write(str(os.getpid()))
|
|
|
|
http.server.ThreadingHTTPServer(("127.0.0.1", args.port), Handler).serve_forever()
|