From 24309449545e4f4588b2f97fe91d1b8643a9bfd6 Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <1051233107@qq.com> Date: Wed, 28 Oct 2026 11:49:27 +0800 Subject: [PATCH] test(mock): add scoped model faults and embedding recovery controls (#2790) --- tools/mock-backends.md | 102 +++++++++++++ tools/mock_llm.py | 268 ++++++++++++++++++++++++++-------- tools/mock_llm_scenarios.json | 130 +++++++++++++++++ tools/test_mock_llm.py | 243 +++++++++++++++++++++++++++--- 4 files changed, 666 insertions(+), 77 deletions(-) create mode 100644 tools/mock-backends.md diff --git a/tools/mock-backends.md b/tools/mock-backends.md new file mode 100644 index 000000000..b62ee1f70 --- /dev/null +++ b/tools/mock-backends.md @@ -0,0 +1,102 @@ +# Deterministic model faults + +Run the test backend with Python 3.11 or newer: + +```sh +python tools/mock_llm.py --port 18080 --record /tmp/mock-requests.jsonl +``` + +Point the LangBot OpenAI-compatible provider at `http://127.0.0.1:18080/v1` +and choose a model below. The default `mock-chat`, `mock-embedding`, tool and +smoke models retain their existing behavior. `GET /v1/models` lists all models. + +| Model | Injected behavior | +| --- | --- | +| `mock-stream-disconnect` | Send two SSE chunks, then close an incomplete HTTP chunked response. Requires streaming. | +| `mock-stream-stall` | Pause for two seconds after the second SSE chunk, then finish. Requires streaming. | +| `mock-recover-503` | Return 503 on the first request in a scope, then `LANGBOT_RECOVERY_OK`. | +| `mock-recover-stream-disconnect` | Disconnect the first stream in a scope; later streams finish with `LANGBOT_STREAM_RECOVERY_OK`. | +| `mock-embedding-429`, `mock-embedding-503` | Return the named HTTP error. | +| `mock-embedding-timeout` | Wait two seconds before responding. Set the client timeout below two seconds to observe a timeout. | +| `mock-embedding-wrong-count` | Omit the final input's vector, including an empty result for one input. | +| `mock-embedding-wrong-dimension` | Return one fewer dimension, or two dimensions when one was requested. Supports float and base64 encoding. | +| `mock-embedding-recover-503` | Return 503 on the first embedding request in a scope, then normal vectors. | + +The daily tool recovery models validate returned tool results before advancing. +`mock-daily-empty-rag-chat` rejects either configured knowledge sentinel anywhere +in the model messages. Token-hash embeddings exercise protocol and retrieval +plumbing; they do not evaluate semantic model quality. + +## Isolate recovery attempts + +Assign each run/case/conversation a unique scope, such as `run17.rag.group3`. +Set this in the LangBot model's advanced parameters: + +```json +{"extra_body": {"mock_scope": "run17.rag.group3"}} +``` + +The Core LiteLLM requester forwards this as top-level `mock_scope` in both chat +and embedding requests. A model-level value is shared by every conversation +using that model: use a separate model configuration per scope or use a marker +`LB_MOCK_SCOPE=run17.rag.group3` in the current user message. Markers in older +chat history are ignored. Embedding input strings also support markers, but +they change the hashed vectors; prefer `extra_body` for embedding tests. +An explicit `mock_scope` overrides markers. IDs accept 1–128 ASCII letters, +digits and `_.:-`. Scopes are test correlation IDs, not tenant authorization. + +To exercise the same contract directly, run this request twice; the first +returns 503 and the second returns the recovery message: + +```sh +curl -i http://127.0.0.1:18080/v1/chat/completions \ + -H 'Content-Type: application/json' \ + -d '{"model":"mock-recover-503","mock_scope":"run17.direct","messages":[{"role":"user","content":"hello"}]}' +``` + +Counters are isolated by endpoint, model name and scope. Each validated request, +including an SDK retry, consumes an attempt; tool steps still follow the actual +tool history. SDK retries can consume the failure and recovery within one +product call: the current Core embedding path retries a 503 even when model +advanced parameters set `max_retries: 0` and `num_retries: 0`. Use the permanent +error models to assert a visible error, and inspect recorded attempts for +recovery scenarios. A request rejected by prompt/tool/input validation does not +consume an attempt. + +After draining a scenario's requests, release all its counters: + +```sh +curl -X DELETE http://127.0.0.1:18080/_mock/scopes/run17.rag.group3 +``` + +The response reports the number released. Reusing a released scope restarts its +fault sequence. Idle counters expire after one hour; at 2,048 retained counters, +new scopes receive 400 until counters are released or expire. Existing scopes +continue working. The backend is unauthenticated and defaults to loopback; +use it only in the test environment. + +## Configure and verify a fault + +Custom `--scenarios` JSON can set `status`, `delay_seconds`, +`chunk_delay_seconds`, `disconnect_after_chunks`, or the pair +`stall_after_chunks` / `stall_seconds`. Delays are bounded to 0–60 seconds; +the first role chunk counts toward stream positions. Add `recover_after: N` +to apply the controls only to the first N requests per scope (1–100). +Embedding scenarios use `kind: "embedding"` and optionally +`embedding_fault: "count"` or `"dimension"`. + +Each completed handler appends the request, scope, attempt (for recovery), HTTP +status, actual `faults`, and `response_complete` to the JSONL record. Delays +include measured elapsed time. A deliberate stream cut records its chunk count +and `injected_disconnect`; socket errors caused by the client record +`disconnected`. `response` contains the planned chat message or embedding +payload; `response_complete` reports whether the server finished writing it, +not whether the client received it. Check the client result too. A configured +stream position that was never reached produces no injection event and cannot +count as a successful fault test. Records contain request bodies; use test data. + +Run the HTTP tests without installing LangBot dependencies: + +```sh +python -m unittest discover -s tools -p test_mock_llm.py -v +``` diff --git a/tools/mock_llm.py b/tools/mock_llm.py index 6a97cb3fa..29eba3668 100644 --- a/tools/mock_llm.py +++ b/tools/mock_llm.py @@ -9,6 +9,7 @@ import json import math from pathlib import Path import re +import socket import struct import threading import time @@ -18,6 +19,36 @@ import uuid MAX_BODY_BYTES = 1024 * 1024 USAGE = {'prompt_tokens': 10, 'completion_tokens': 5, 'total_tokens': 15} GROUP_TURN = re.compile(r'LB_GROUP=(\S+) LB_TURN=(\d+) LB_TOKEN=([a-f0-9]+) LB_PREVIOUS=(none|[a-f0-9]+)') +SCOPE_MARKER = re.compile(r'\bLB_MOCK_SCOPE=([^\s]+)') +SCOPE_ID = re.compile(r'[A-Za-z0-9_.:-]{1,128}') +MAX_FAULT_SCOPES = 2048 +FAULT_SCOPE_TTL_SECONDS = 3600 + + +class StreamDisconnected(Exception): + """The mock deliberately closed an unfinished HTTP chunked response.""" + + +def fault_scope(body, embedding): + if 'mock_scope' in body: + scope = body['mock_scope'] + else: + if not embedding: + text = next( + (message.get('content', '') for message in reversed(body['messages']) if message.get('role') == 'user'), + '', + ) + else: + text = body.get('input', '') + if isinstance(text, list): + text = '\n'.join(str(item.get('text', '') if isinstance(item, dict) else item) for item in text) + scopes = set(SCOPE_MARKER.findall(str(text))) + if len(scopes) > 1: + raise ValueError('request contains conflicting LB_MOCK_SCOPE markers') + scope = next(iter(scopes), None) + if scope is not None and (not isinstance(scope, str) or not SCOPE_ID.fullmatch(scope)): + raise ValueError('mock_scope must contain 1..128 letters, digits, or _.:-') + return scope def group_ack(messages, start): @@ -116,6 +147,8 @@ def select_message(body, scenario): if expected_prompt := step.get('expect_prompt'): if expected_prompt not in json.dumps(messages, ensure_ascii=False): raise ValueError(f'model input must contain {expected_prompt!r}') + if any(value in json.dumps(messages, ensure_ascii=False) for value in step.get('forbid_prompt', [])): + raise ValueError('model input contains forbidden knowledge content') expected = step.get('expect_tool_output') if expected is not None: results = turn[calls[-1] + 1 :] if calls else [] @@ -151,10 +184,34 @@ class MockServer(ThreadingHTTPServer): daemon_threads = True def __init__(self, address, scenarios, record_path=None): + validate_scenarios(scenarios) super().__init__(address, Handler) - self.scenarios = scenarios + self.scenarios = {'mock-embedding': {'kind': 'embedding'}, **scenarios} self.record_path = record_path self.record_lock = threading.Lock() + self.fault_lock = threading.Lock() + self.fault_scopes = {} + + def fault_controls(self, body, scenario, entry): + scope = fault_scope(body, entry['path'] == '/v1/embeddings') + entry['scope'] = scope + if 'recover_after' not in scenario: + return scenario + if scope is None: + raise ValueError('recovery scenarios require mock_scope or LB_MOCK_SCOPE') + key = (entry['path'], body['model'], scope) + now = time.monotonic() + with self.fault_lock: + for expired in [ + key for key, (_, touched) in self.fault_scopes.items() if now - touched >= FAULT_SCOPE_TTL_SECONDS + ]: + del self.fault_scopes[expired] + if key not in self.fault_scopes and len(self.fault_scopes) >= MAX_FAULT_SCOPES: + raise ValueError('mock fault scope limit reached; release completed scopes first') + attempt = self.fault_scopes.get(key, (0, now))[0] + 1 + self.fault_scopes[key] = (attempt, now) + entry.update(attempt=attempt, fault_active=attempt <= scenario['recover_after']) + return scenario if entry['fault_active'] else {} def record(self, entry): if self.record_path: @@ -192,7 +249,7 @@ class Handler(BaseHTTPRequestHandler): 'created': 0, 'owned_by': 'mock', } - for name in [*self.server.scenarios, 'mock-embedding'] + for name in self.server.scenarios ], }, ) @@ -207,8 +264,33 @@ class Handler(BaseHTTPRequestHandler): }, ) + def do_DELETE(self): + prefix = '/_mock/scopes/' + scope = self.path.removeprefix(prefix) + if not self.path.startswith(prefix) or not SCOPE_ID.fullmatch(scope): + self.json_response(400, {'error': {'message': 'use DELETE /_mock/scopes/'}}) + return + with self.server.fault_lock: + keys = [key for key in self.server.fault_scopes if key[2] == scope] + for key in keys: + del self.server.fault_scopes[key] + self.json_response(200, {'scope': scope, 'released': len(keys)}) + + def delay(self, seconds, kind, entry): + if seconds: + start = time.monotonic() + time.sleep(seconds) + entry['faults'].append({'kind': kind, 'seconds': seconds, 'elapsed_seconds': time.monotonic() - start}) + def do_POST(self): - entry = {'request_id': uuid.uuid4().hex, 'time': time.time(), 'path': self.path, 'status': 400} + entry = { + 'request_id': uuid.uuid4().hex, + 'time': time.time(), + 'path': self.path, + 'status': 400, + 'faults': [], + 'response_complete': False, + } try: if self.path not in ('/v1/chat/completions', '/v1/embeddings'): entry['status'] = 404 @@ -221,6 +303,7 @@ class Handler(BaseHTTPRequestHandler): } }, ) + entry['response_complete'] = True return self.connection.settimeout(10) length = int(self.headers.get('Content-Length', 0)) @@ -230,31 +313,43 @@ class Handler(BaseHTTPRequestHandler): if not isinstance(body, dict) or not isinstance(body.get('model'), str): raise ValueError('request must be an object with a model name') entry['request'] = body - if self.path == '/v1/embeddings': - if body['model'] != 'mock-embedding': - raise ValueError('embedding model must be mock-embedding') - payload = mock_embeddings(body) - entry.update(status=200, response=payload) - self.json_response(200, payload) - return scenario = self.server.scenarios.get(body['model']) if scenario is None: raise ValueError('unknown model; use GET /v1/models') - message = select_message(body, scenario) - time.sleep(scenario.get('delay_seconds', 0)) - status = scenario.get('status', 200) + embedding = self.path == '/v1/embeddings' + if embedding != (scenario.get('kind') == 'embedding'): + raise ValueError('model does not support this endpoint') + if embedding: + payload = mock_embeddings(body) + else: + message = select_message(body, scenario) + if not body.get('stream') and ( + 'disconnect_after_chunks' in scenario or 'stall_after_chunks' in scenario + ): + raise ValueError('this fault scenario requires stream=true') + controls = self.server.fault_controls(body, scenario, entry) + self.delay(controls.get('delay_seconds', 0), 'delay', entry) + status = controls.get('status', 200) entry['status'] = status if status != 200: - self.json_response( - status, - { - 'error': { - 'message': 'injected mock failure', - 'type': 'api_error', - 'code': str(status), - } - }, - ) + entry['faults'].append({'kind': 'http_error', 'status': status}) + entry['response'] = { + 'error': {'message': 'injected mock failure', 'type': 'api_error', 'code': str(status)} + } + self.json_response(status, entry['response']) + entry['response_complete'] = True + return + if embedding: + if fault := controls.get('embedding_fault'): + if fault == 'count': + payload['data'].pop() + else: + dimensions = body.get('dimensions', 384) + payload = mock_embeddings({**body, 'dimensions': dimensions - 1 if dimensions > 1 else 2}) + entry['faults'].append({'kind': 'embedding_' + fault}) + entry['response'] = payload + self.json_response(200, payload) + entry['response_complete'] = True return entry['response'] = message base = { @@ -264,7 +359,7 @@ class Handler(BaseHTTPRequestHandler): } finish = 'tool_calls' if message.get('tool_calls') else 'stop' if body.get('stream'): - self.stream_response(body, scenario, message, base, finish) + self.stream_response(body, controls, message, base, finish, entry) else: self.json_response( 200, @@ -275,29 +370,51 @@ class Handler(BaseHTTPRequestHandler): 'usage': USAGE, }, ) + entry['response_complete'] = True + except StreamDisconnected: + entry['injected_disconnect'] = True except (ValueError, KeyError, TypeError, TimeoutError) as exc: entry.update(status=400, error=str(exc)) self.json_response(400, {'error': {'message': str(exc), 'type': 'invalid_request_error'}}) + entry['response_complete'] = True except (BrokenPipeError, ConnectionResetError): entry['disconnected'] = True finally: self.server.record(entry) - def stream_response(self, body, scenario, message, base, finish): + def stream_response(self, body, scenario, message, base, finish, entry): + chunked = 'disconnect_after_chunks' in scenario + if chunked: + self.protocol_version = 'HTTP/1.1' self.send_response(200) self.send_header('Content-Type', 'text/event-stream') self.send_header('Cache-Control', 'no-cache') + if chunked: + self.send_header('Transfer-Encoding', 'chunked') self.end_headers() + def write(data): + if chunked: + data = f'{len(data):x}\r\n'.encode() + data + b'\r\n' + self.wfile.write(data) + self.wfile.flush() + def send(delta, reason=None): payload = { **base, 'object': 'chat.completion.chunk', 'choices': [{'index': 0, 'delta': delta, 'finish_reason': reason}], } - self.wfile.write(('data: ' + json.dumps(payload, ensure_ascii=False) + '\n\n').encode()) - self.wfile.flush() - time.sleep(scenario.get('chunk_delay_seconds', 0)) + write(('data: ' + json.dumps(payload, ensure_ascii=False) + '\n\n').encode()) + entry['stream_chunks'] = entry.get('stream_chunks', 0) + 1 + if entry['stream_chunks'] == scenario.get('disconnect_after_chunks'): + entry['faults'].append({'kind': 'stream_disconnect', 'after_chunks': entry['stream_chunks']}) + self.close_connection = True + self.connection.shutdown(socket.SHUT_RDWR) + raise StreamDisconnected + if entry['stream_chunks'] == scenario.get('stall_after_chunks'): + self.delay(scenario['stall_seconds'], 'stream_stall', entry) + self.delay(scenario.get('chunk_delay_seconds', 0), 'chunk_delay', entry) send({'role': 'assistant'}) content = message.get('content') or '' @@ -339,9 +456,65 @@ class Handler(BaseHTTPRequestHandler): 'choices': [], 'usage': USAGE, } - self.wfile.write(('data: ' + json.dumps(payload) + '\n\n').encode()) - self.wfile.write(b'data: [DONE]\n\n') - self.wfile.flush() + write(('data: ' + json.dumps(payload) + '\n\n').encode()) + write(b'data: [DONE]\n\n') + if chunked: + self.wfile.write(b'0\r\n\r\n') + self.wfile.flush() + + +def validate_scenarios(scenarios): + if not isinstance(scenarios, dict) or not scenarios: + raise ValueError('scenarios must be a non-empty object keyed by model name') + for name, scenario in scenarios.items(): + if not isinstance(scenario, dict): + raise ValueError(f'{name}: scenario must be an object') + if scenario.get('kind', 'chat') not in ('chat', 'embedding'): + raise ValueError(f'{name}: kind must be chat or embedding') + steps = scenario.get('steps', []) + if not isinstance(steps, list) or (not steps and scenario.get('kind') != 'embedding'): + raise ValueError(f'{name}: steps must be a non-empty array') + if any(not isinstance(step, dict) for step in steps): + raise ValueError(f'{name}: each step must be an object') + for step in steps: + if not isinstance(step.get('content'), str) and not step.get('tool_calls'): + raise ValueError(f'{name}: each step needs content or tool_calls') + if 'expect_tool_output' in step and not isinstance(step['expect_tool_output'], str): + raise ValueError(f'{name}: expect_tool_output must be a string') + if 'expect_prompt' in step and not isinstance(step['expect_prompt'], str): + raise ValueError(f'{name}: expect_prompt must be a string') + if 'forbid_prompt' in step and ( + not isinstance(step['forbid_prompt'], list) + or any(not isinstance(value, str) or not value for value in step['forbid_prompt']) + ): + raise ValueError(f'{name}: forbid_prompt must contain non-empty strings') + if 'tool_calls' in step: + calls = step['tool_calls'] + if not isinstance(calls, list) or any( + not isinstance(call, dict) + or not isinstance(call.get('name'), str) + or not isinstance(call.get('arguments'), dict) + for call in calls + ): + raise ValueError(f'{name}: tool_calls need a name and arguments object') + if type(scenario.get('status', 200)) is not int or scenario.get('status', 200) not in (200, 401, 429, 500, 503): + raise ValueError(f'{name}: unsupported injected HTTP status') + for key in ('delay_seconds', 'chunk_delay_seconds', 'stall_seconds'): + value = scenario.get(key, 0) + if isinstance(value, bool) or not isinstance(value, (int, float)) or not 0 <= value <= 60: + raise ValueError(f'{name}: {key} must be between 0 and 60') + for key in ('recover_after', 'disconnect_after_chunks', 'stall_after_chunks'): + if key in scenario and (type(scenario[key]) is not int or not 1 <= scenario[key] <= 100): + raise ValueError(f'{name}: {key} must be an integer in 1..100') + if ('stall_after_chunks' in scenario) != ('stall_seconds' in scenario): + raise ValueError(f'{name}: stall_after_chunks and stall_seconds must be set together') + if scenario.get('embedding_fault') not in (None, 'count', 'dimension'): + raise ValueError(f'{name}: embedding_fault must be count or dimension') + if scenario.get('kind') == 'embedding': + if any(key in scenario for key in ('disconnect_after_chunks', 'stall_after_chunks', 'chunk_delay_seconds')): + raise ValueError(f'{name}: embedding scenarios cannot set stream controls') + elif 'embedding_fault' in scenario: + raise ValueError(f'{name}: embedding_fault requires kind=embedding') def main(): @@ -360,35 +533,10 @@ def main(): ) args = parser.parse_args() scenarios = json.loads(args.scenarios.read_text(encoding='utf-8')) - if not isinstance(scenarios, dict) or not scenarios: - parser.error('scenarios must be a non-empty object keyed by model name') - for name, scenario in scenarios.items(): - if not isinstance(scenario, dict) or not isinstance(scenario.get('steps'), list) or not scenario['steps']: - parser.error(f'{name}: steps must be a non-empty array') - if any(not isinstance(step, dict) for step in scenario['steps']): - parser.error(f'{name}: each step must be an object') - for step in scenario['steps']: - if not isinstance(step.get('content'), str) and not step.get('tool_calls'): - parser.error(f'{name}: each step needs content or tool_calls') - if 'expect_tool_output' in step and not isinstance(step['expect_tool_output'], str): - parser.error(f'{name}: expect_tool_output must be a string') - if 'expect_prompt' in step and not isinstance(step['expect_prompt'], str): - parser.error(f'{name}: expect_prompt must be a string') - if 'tool_calls' in step: - calls = step['tool_calls'] - if not isinstance(calls, list) or any( - not isinstance(call, dict) - or not isinstance(call.get('name'), str) - or not isinstance(call.get('arguments'), dict) - for call in calls - ): - parser.error(f'{name}: tool_calls need a name and arguments object') - if scenario.get('status', 200) not in (200, 401, 429, 500, 503): - parser.error(f'{name}: unsupported injected HTTP status') - for key in ('delay_seconds', 'chunk_delay_seconds'): - value = scenario.get(key, 0) - if not isinstance(value, (int, float)) or not 0 <= value <= 60: - parser.error(f'{name}: {key} must be between 0 and 60') + try: + validate_scenarios(scenarios) + except ValueError as exc: + parser.error(str(exc)) with MockServer((args.host, args.port), scenarios, args.record) as server: print(f'Mock LLM: http://{args.host}:{server.server_port}/v1', flush=True) try: diff --git a/tools/mock_llm_scenarios.json b/tools/mock_llm_scenarios.json index 0753c5b6a..f870a43b6 100644 --- a/tools/mock_llm_scenarios.json +++ b/tools/mock_llm_scenarios.json @@ -165,5 +165,135 @@ "content": "LANGBOT_OTHER_RAG_OK" } ] + }, + "mock-daily-tool-recovery": { + "steps": [ + { + "tool_calls": [ + { + "name": "exec", + "arguments": { + "command": "printf EXPECTED_RECOVERY_ERROR >&2; exit 7" + } + } + ] + }, + { + "expect_tool_output": "EXPECTED_RECOVERY_ERROR", + "tool_calls": [ + { + "name": "exec", + "arguments": { + "command": "printf RECOVERY_OK" + } + } + ] + }, + { + "expect_tool_output": "RECOVERY_OK", + "content": "LANGBOT_TOOL_RECOVERY_OK" + } + ] + }, + "mock-daily-tool-timeout-recovery": { + "steps": [ + { + "tool_calls": [ + { + "name": "exec", + "arguments": { + "command": "rm -f /tmp/langbot-timeout-leak; sleep 3; printf LEAK > /tmp/langbot-timeout-leak", + "timeout_sec": 1 + } + } + ] + }, + { + "expect_tool_output": "timed_out", + "tool_calls": [ + { + "name": "exec", + "arguments": { + "command": "sleep 3; test ! -e /tmp/langbot-timeout-leak && printf TIMEOUT_RECOVERY_OK" + } + } + ] + }, + { + "expect_tool_output": "TIMEOUT_RECOVERY_OK", + "content": "LANGBOT_TOOL_TIMEOUT_RECOVERY_OK" + } + ] + }, + "mock-daily-empty-rag-chat": { + "steps": [ + { + "forbid_prompt": [ + "aurora-parser-rag-9137", + "cedar-isolation-4826" + ], + "content": "LANGBOT_EMPTY_RAG_CHAT_OK" + } + ] + }, + "mock-stream-disconnect": { + "disconnect_after_chunks": 2, + "steps": [ + { + "content": "LANGBOT_STREAM_OK" + } + ] + }, + "mock-stream-stall": { + "stall_after_chunks": 2, + "stall_seconds": 2, + "steps": [ + { + "content": "LANGBOT_STREAM_OK" + } + ] + }, + "mock-recover-503": { + "status": 503, + "recover_after": 1, + "steps": [ + { + "content": "LANGBOT_RECOVERY_OK" + } + ] + }, + "mock-recover-stream-disconnect": { + "disconnect_after_chunks": 2, + "recover_after": 1, + "steps": [ + { + "content": "LANGBOT_STREAM_RECOVERY_OK" + } + ] + }, + "mock-embedding-429": { + "kind": "embedding", + "status": 429 + }, + "mock-embedding-503": { + "kind": "embedding", + "status": 503 + }, + "mock-embedding-timeout": { + "kind": "embedding", + "delay_seconds": 2 + }, + "mock-embedding-wrong-count": { + "kind": "embedding", + "embedding_fault": "count" + }, + "mock-embedding-wrong-dimension": { + "kind": "embedding", + "embedding_fault": "dimension" + }, + "mock-embedding-recover-503": { + "kind": "embedding", + "status": 503, + "recover_after": 1 } } diff --git a/tools/test_mock_llm.py b/tools/test_mock_llm.py index 8c8d74577..2f31f329d 100644 --- a/tools/test_mock_llm.py +++ b/tools/test_mock_llm.py @@ -2,16 +2,20 @@ from concurrent.futures import ThreadPoolExecutor import base64 +from http.client import IncompleteRead import json from pathlib import Path import tempfile import struct import threading +import time import unittest +from unittest.mock import patch from urllib.error import HTTPError from urllib.request import Request, urlopen -from mock_llm import MockServer +import mock_llm +from mock_llm import MockServer, validate_scenarios class MockLLMTest(unittest.TestCase): @@ -19,7 +23,18 @@ class MockLLMTest(unittest.TestCase): def setUpClass(cls): cls.tmp = tempfile.TemporaryDirectory() cls.record = Path(cls.tmp.name) / 'requests.jsonl' + cls.record.touch() scenarios = json.loads(Path(__file__).with_name('mock_llm_scenarios.json').read_text()) + scenarios['test-stall-recovery'] = { + **scenarios['mock-stream-stall'], + 'stall_seconds': 0.2, + 'recover_after': 1, + } + scenarios['test-delay-recovery'] = { + 'delay_seconds': 0.2, + 'recover_after': 1, + 'steps': [{'content': 'RECOVERED'}], + } cls.server = MockServer(('127.0.0.1', 0), scenarios, cls.record) cls.thread = threading.Thread(target=cls.server.serve_forever) cls.thread.start() @@ -32,7 +47,7 @@ class MockLLMTest(unittest.TestCase): cls.thread.join() cls.tmp.cleanup() - def post(self, model='mock-chat', **kwargs): + def post(self, model='mock-chat', timeout=5, **kwargs): body = { 'model': model, 'messages': [{'role': 'user', 'content': 'test'}], @@ -46,9 +61,34 @@ class MockLLMTest(unittest.TestCase): 'Authorization': 'Bearer must-not-be-recorded', }, ) - with urlopen(request, timeout=5) as response: + with urlopen(request, timeout=timeout) as response: return response.read().decode() + def embed(self, timeout=5, **patch): + body = {'model': 'mock-embedding', 'input': ['repeatable fixture', 'another fixture'], **patch} + request = Request( + self.url + '/v1/embeddings', + data=json.dumps(body).encode(), + headers={'Content-Type': 'application/json'}, + ) + with urlopen(request, timeout=timeout) as response: + return json.load(response) + + def records(self, scope, count): + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + with self.server.record_lock: + rows = [json.loads(line) for line in self.record.read_text().splitlines()] + rows = [row for row in rows if row.get('scope') == scope] + if len(rows) >= count: + return sorted(rows, key=lambda row: row.get('attempt', 0)) + time.sleep(0.01) + self.fail(f'missing completed request records for {scope}') + + def release(self, scope): + with urlopen(Request(self.url + '/_mock/scopes/' + scope, method='DELETE'), timeout=5) as response: + return json.load(response) + def test_chat_and_model_discovery(self): with urlopen(self.url + '/v1/models') as response: models = json.load(response) @@ -80,28 +120,18 @@ class MockLLMTest(unittest.TestCase): self.assertEqual(failure.exception.code, 400) def test_embedding_http_float_base64_and_validation(self): - def embed(**patch): - body = {'model': 'mock-embedding', 'input': ['repeatable fixture', 'another fixture'], **patch} - request = Request( - self.url + '/v1/embeddings', - data=json.dumps(body).encode(), - headers={'Content-Type': 'application/json'}, - ) - with urlopen(request, timeout=5) as response: - return json.load(response) - - first = embed()['data'] - self.assertEqual(first, embed()['data']) + first = self.embed()['data'] + self.assertEqual(first, self.embed()['data']) self.assertEqual([item['index'] for item in first], [0, 1]) self.assertEqual(len(first[0]['embedding']), 384) self.assertAlmostEqual(sum(value * value for value in first[0]['embedding']), 1) - encoded = embed(encoding_format='base64')['data'][0]['embedding'] + encoded = self.embed(encoding_format='base64')['data'][0]['embedding'] decoded = struct.unpack('<384f', base64.b64decode(encoded)) for actual, expected in zip(decoded, first[0]['embedding']): self.assertAlmostEqual(actual, expected, places=6) for patch in ({'input': []}, {'dimensions': 0}, {'encoding_format': 'bad'}, {'model': 'unknown'}): with self.subTest(patch=patch), self.assertRaises(HTTPError) as caught: - embed(**patch) + self.embed(**patch) self.assertEqual(caught.exception.code, 400) def box_turn(self, streaming=False): @@ -171,6 +201,185 @@ class MockLLMTest(unittest.TestCase): ) self.assertEqual(response['choices'][0]['message']['content'], 'LANGBOT_RAG_OK') + def test_empty_rag_rejects_context_from_another_knowledge_base(self): + self.assertIn('LANGBOT_EMPTY_RAG_CHAT_OK', self.post('mock-daily-empty-rag-chat')) + for leaked in ('aurora-parser-rag-9137', 'cedar-isolation-4826'): + with self.subTest(leaked=leaked), self.assertRaises(HTTPError) as caught: + self.post('mock-daily-empty-rag-chat', messages=[{'role': 'user', 'content': leaked}]) + self.assertEqual(caught.exception.code, 400) + + def test_concurrent_scopes_fail_independently_then_recover(self): + def conversation(index): + scope = f'concurrent-{index}' + body = {'messages': [{'role': 'user', 'content': f'hello LB_MOCK_SCOPE={scope}'}]} + with self.assertRaises(HTTPError) as caught: + self.post('mock-recover-503', **body) + self.assertEqual(caught.exception.code, 503) + self.assertIn('LANGBOT_CHAT_OK', self.post()) + self.assertIn('LANGBOT_RECOVERY_OK', self.post('mock-recover-503', **body)) + rows = self.records(scope, 2) + self.assertEqual([row['attempt'] for row in rows], [1, 2]) + self.assertEqual(rows[0]['faults'], [{'kind': 'http_error', 'status': 503}]) + self.assertEqual(rows[1]['faults'], []) + self.assertTrue(rows[1]['response_complete']) + self.assertEqual(self.release(scope)['released'], 1) + + with ThreadPoolExecutor(max_workers=4) as pool: + list(pool.map(conversation, range(8))) + + def test_concurrent_retries_consume_only_one_failure(self): + def request(_): + try: + self.post('mock-recover-503', mock_scope='same-scope') + return 200 + except HTTPError as exc: + return exc.code + + with ThreadPoolExecutor(max_workers=4) as pool: + statuses = list(pool.map(request, range(8))) + self.assertEqual(sorted(statuses), [200] * 7 + [503]) + rows = self.records('same-scope', 8) + self.assertEqual(sorted(row['attempt'] for row in rows), list(range(1, 9))) + self.assertEqual(self.release('same-scope')['released'], 1) + + def test_stream_disconnect_is_a_transport_failure_and_recovers(self): + with self.assertRaises(IncompleteRead) as caught: + self.post('mock-recover-stream-disconnect', stream=True, mock_scope='disconnect') + self.assertIn(b'LANGBOT_', caught.exception.partial) + self.assertNotIn(b'[DONE]', caught.exception.partial) + result = self.post('mock-recover-stream-disconnect', stream=True, mock_scope='disconnect') + self.assertIn('data: [DONE]', result) + rows = self.records('disconnect', 2) + self.assertEqual(rows[0]['faults'], [{'kind': 'stream_disconnect', 'after_chunks': 2}]) + self.assertFalse(rows[0]['response_complete']) + self.assertTrue(rows[0]['injected_disconnect']) + self.assertEqual(rows[1]['faults'], []) + self.assertTrue(rows[1]['response_complete']) + self.release('disconnect') + + def test_delay_and_midstream_stall_timeout_then_recover(self): + for model, stream, kind in ( + ('test-delay-recovery', False, 'delay'), + ('test-stall-recovery', True, 'stream_stall'), + ): + with self.subTest(model=model): + with self.assertRaises(TimeoutError): + self.post(model, stream=stream, timeout=0.05, mock_scope=model) + self.assertIn('LANGBOT_CHAT_OK', self.post()) + result = self.post(model, stream=stream, mock_scope=model) + self.assertIn('data: [DONE]' if stream else 'RECOVERED', result) + rows = sorted(self.records(model, 2), key=lambda row: row['attempt']) + self.assertEqual(rows[0]['faults'][0]['kind'], kind) + self.assertGreaterEqual(rows[0]['faults'][0]['elapsed_seconds'], 0.2) + self.assertEqual(rows[1]['faults'], []) + self.release(model) + + def test_embedding_errors_timeout_count_and_dimension(self): + for status in (429, 503): + scope = f'embed-{status}' + with self.assertRaises(HTTPError) as caught: + self.embed(model=f'mock-embedding-{status}', mock_scope=scope) + self.assertEqual(caught.exception.code, status) + self.assertEqual(self.records(scope, 1)[0]['faults'], [{'kind': 'http_error', 'status': status}]) + with self.assertRaises(TimeoutError): + self.embed(model='mock-embedding-timeout', mock_scope='embed-timeout', timeout=0.05) + self.assertEqual(len(self.embed()['data']), 2) + for encoding in ('float', 'base64'): + wrong_count = self.embed(model='mock-embedding-wrong-count', encoding_format=encoding) + self.assertEqual(len(wrong_count['data']), 1) + wrong_dimension = self.embed( + model='mock-embedding-wrong-dimension', + dimensions=4, + encoding_format=encoding, + ) + self.assertEqual([item['index'] for item in wrong_dimension['data']], [0, 1]) + vector = wrong_dimension['data'][0]['embedding'] + self.assertEqual(len(vector) if encoding == 'float' else len(base64.b64decode(vector)) // 4, 3) + delay = self.records('embed-timeout', 1)[0]['faults'][0] + self.assertEqual(delay['kind'], 'delay') + self.assertGreaterEqual(delay['elapsed_seconds'], 2) + + def test_embedding_recovery_scope_does_not_modify_vectors(self): + expected = self.embed()['data'] + + def conversation(scope): + with self.assertRaises(HTTPError) as caught: + self.embed(model='mock-embedding-recover-503', mock_scope=scope) + self.assertEqual(caught.exception.code, 503) + result = self.embed(model='mock-embedding-recover-503', mock_scope=scope) + self.assertEqual(result['data'], expected) + self.assertEqual(self.records(scope, 2)[1]['faults'], []) + self.release(scope) + + with ThreadPoolExecutor(max_workers=2) as pool: + list(pool.map(conversation, ('embed-a', 'embed-b'))) + + def test_markers_select_current_scope_and_invalid_input_does_not_advance(self): + with self.assertRaises(HTTPError) as caught: + self.post('mock-recover-503', messages=[], mock_scope='marker-current') + self.assertEqual(caught.exception.code, 400) + with self.assertRaises(HTTPError) as caught: + self.post( + 'mock-recover-503', + messages=[ + {'role': 'user', 'content': 'LB_MOCK_SCOPE=marker-old'}, + {'role': 'assistant', 'content': 'prior reply'}, + {'role': 'user', 'content': [{'type': 'text', 'text': 'LB_MOCK_SCOPE=marker-current'}]}, + ], + ) + self.assertEqual(caught.exception.code, 503) + self.assertEqual(self.records('marker-current', 1)[0]['attempt'], 1) + self.assertIn('LANGBOT_RECOVERY_OK', self.post('mock-recover-503', mock_scope='marker-current')) + self.assertEqual(self.release('marker-current')['released'], 1) + self.assertEqual(self.release('marker-old')['released'], 0) + with self.assertRaises(HTTPError) as caught: + self.embed(model='mock-embedding-recover-503', input=['LB_MOCK_SCOPE=marker-embed']) + self.assertEqual(caught.exception.code, 503) + self.assertEqual(self.records('marker-embed', 1)[0]['attempt'], 1) + self.release('marker-embed') + + def test_scope_validation_capacity_expiry_and_cleanup(self): + for scope in (None, '', [], 'x' * 129, 'bad/scope'): + with self.subTest(scope=scope), self.assertRaises(HTTPError) as caught: + self.post('mock-recover-503', mock_scope=scope) + self.assertEqual(caught.exception.code, 400) + with patch.object(mock_llm, 'MAX_FAULT_SCOPES', len(self.server.fault_scopes) + 1): + with self.assertRaises(HTTPError) as caught: + self.post('mock-recover-503', mock_scope='capacity-a') + self.assertEqual(caught.exception.code, 503) + with self.assertRaises(HTTPError) as caught: + self.post('mock-recover-503', mock_scope='capacity-b') + self.assertEqual(caught.exception.code, 400) + self.assertIn('scope limit', caught.exception.read().decode()) + self.assertIn('LANGBOT_RECOVERY_OK', self.post('mock-recover-503', mock_scope='capacity-a')) + self.assertEqual(self.release('capacity-a')['released'], 1) + with self.assertRaises(HTTPError) as caught: + self.post('mock-recover-503', mock_scope='capacity-b') + self.assertEqual(caught.exception.code, 503) + with patch.object(mock_llm, 'FAULT_SCOPE_TTL_SECONDS', 0), self.assertRaises(HTTPError) as caught: + self.post('mock-recover-503', mock_scope='capacity-b') + self.assertEqual(caught.exception.code, 503) + self.assertEqual(self.release('capacity-b')['released'], 1) + self.assertEqual(self.release('capacity-b')['released'], 0) + + def test_scenario_controls_reject_invalid_values(self): + for control in ( + {'delay_seconds': 61}, + {'delay_seconds': float('nan')}, + {'delay_seconds': True}, + {'recover_after': 0}, + {'disconnect_after_chunks': -1}, + {'stall_after_chunks': 2}, + {'stall_seconds': 1}, + {'embedding_fault': 'count'}, + {'forbid_prompt': 'not-an-array'}, + ): + scenario = {'steps': [{'content': 'OK'}], **control} + if 'forbid_prompt' in control: + scenario = {'steps': [{'content': 'OK', **control}]} + with self.subTest(control=control), self.assertRaises(ValueError): + validate_scenarios({'invalid': scenario}) + def test_errors_and_invalid_requests(self): for status in (401, 429, 503): with self.subTest(status=status), self.assertRaises(HTTPError) as caught: