xinyin025/LangBot
1
0
Fork 0
mirror of https://github.com/langbot-app/LangBot.git synced 2026-10-30 04:15:45 +08:00
Code Issues Packages Projects Releases Wiki Activity

test(tools): add deterministic lbctl regression job

Browse Source
This commit is contained in:
huanghuoguoguo
2026-10-18 10:34:03 +08:00
parent 399685962e
commit e4043a250b
11 changed files with 2437 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/README.md
+234
View File
@@ -0,0 +1,234 @@
# Mock LLM for regression tests
Run the deterministic OpenAI Chat Completions endpoint with Python 3.11+, without
API credentials or third-party dependencies:
```sh
python3 tools/mock_llm.py --port 18080 --record /tmp/langbot-mock-requests.jsonl
```
In LangBot, create an `openai-chat-completions` Provider whose base URL is
`http://127.0.0.1:18080/v1` and whose API key is any test-only value. Use a model
from `GET /v1/models`. The loopback address works when Core and this tool run on
the same host; containers must use an address reachable from their own network.
| Model | Behavior |
| --- | --- |
| `mock-chat` | Returns `LANGBOT_CHAT_OK`, including SSE streaming |
| `mock-rag` | Requires the golden document sentinel in the model input, then returns `LANGBOT_RAG_OK` |
| `mock-embedding` | Returns deterministic normalized token-hash vectors from `/v1/embeddings` |
| `mock-box` | Calls `exec` twice to write/read `/workspace/mock-llm.txt`, then returns `LANGBOT_BOX_OK` |
| `mock-401`, `mock-429`, `mock-503` | Returns the selected HTTP failure |
| `mock-slow` | Delays the response by two seconds |
| `mock-slow-stream` | Delays each SSE chunk by 0.2 seconds |
`mock-rag` requires `aurora-parser-rag-9137` in the submitted messages. The
[golden HTML fixture](../skills/skills/langbot-testing/fixtures/rag/parser-golden.html)
contains this marker; retrieve it into the model context without including it
in the user's question.
For `mock-box`, enable Box in the Runner and expose the native `exec` tool. Each
tool result must contain `LANGBOT_BOX_SENTINEL` before the next step is returned.
Tool-call IDs and fragmented streaming arguments use the normal OpenAI protocol.
Scenario progress comes from each request's current conversation turn, so
concurrent sessions and retries do not consume a shared script counter.
Add fixed response/tool sequences with `--scenarios path/to/scenarios.json`, using
`mock_llm_scenarios.json` as the example. `expect_tool_output` checks a substring
in the previous tool results; `expect_prompt` checks the submitted messages.
The server returns tool calls; it never executes
their commands. Request recordings contain prompts, tool results, responses,
status, and disconnects, but no HTTP headers. Use test data only.
Run the focused HTTP checks on the Linux test host:
```sh
python3 -m unittest discover -s tools -p test_mock_llm.py -v
```
These checks validate the mock server. A LangBot regression must additionally
verify real tool outputs and retrieved document IDs/content. Token usage is
synthetic, and this endpoint does not evaluate answer quality or embedding models.
`mock-embedding` accepts string/batch inputs, float/base64 encoding, and optional
dimensions (default 384). It tests the embedding protocol and RAG plumbing;
token hashes do not establish semantic retrieval quality.
## Normal traffic and resource collection
`regression_traffic.py` executes a JSON list of normal `lbctl` cases at a selected
rate. Each case has a name, CLI arguments, and an expected response substring:
```json
[{"name":"chat","args":["pipeline","run","PIPELINE_UUID","--message","Hello"],"expect":"LANGBOT_CHAT_OK"}]
```
Prepare a Workspace-bound CLI context and its API key environment variable, then
run on the Linux test host:
```sh
python3 tools/regression_traffic.py \
--lbctl /path/to/lbctl --config /path/to/lbctl.yaml \
--cases /path/to/cases.json --output /path/to/new-results-directory \
--duration 600 --rps 2 --workers 4 --cooldown 30 \
--pid core=CORE_PID --pid mock=MOCK_PID \
--health core=http://127.0.0.1:5300/healthz \
--health mock=http://127.0.0.1:18080/healthz
```
Add `--container CONTAINER_ID` for Docker CPU/memory/PID statistics. The executing
account must be able to inspect the selected process trees and Docker daemon.
Resource collection reuses `scripts/cloud_runtime_soak.py`; it does not require
cgroup v2. PID resource totals include descendants, so select disjoint trees.
The result directory contains `requests.jsonl`, `resources.jsonl`, `progress.json`,
and `summary.json`. The requested rate is a target: four workers bound concurrency,
and slow requests lower the achieved rate. Requests with unknown outcomes are
recorded without retry. A command fails its business check if it returns an error
or omits the expected response. Health/resource collection failures also make the
run fail. API key values are redacted from recorded CLI output.
Latency percentiles include the entire CLI invocation and use successful
requests only. Achieved request rate is `requests / load_elapsed_seconds`.
`health_samples` counts sampling rounds; `health_failures` counts rounds with
any failed health probe or process/container resource read.
## Configured regression suites
[`regression_suite.py`](regression_suite.py) turns a JSON profile into a repeatable
resource preparation, traffic, observation, verdict, and cleanup sequence. Start
with [`regression_profiles/normal.json`](regression_profiles/normal.json), which
defines the chat/Box/RAG mix from the ten-minute trial.
Use a dedicated QA Workspace. First provision the instance with `lbctl install`
or your source deployment, enable Docker Box, start the mock and embedding
backends, and prepare a Workspace-bound `lbctl` config. Set the profile's CLI
config path, backend URLs, and three plugin archive paths. The sample embedding
endpoint must serve `all-MiniLM-L6-v2`; the mock serves generation only. Relative
paths resolve against the profile directory, so adjust them if you move it.
Run on the Linux test host:
```sh
python3 tools/regression_suite.py validate --profile tools/regression_profiles/normal.json
python3 tools/regression_suite.py run \
--profile tools/regression_profiles/normal.json --output .qa/regression/run-001
```
`validate` checks profile syntax and references without contacting the instance.
The output directory for `run` must be new. CLI options `--duration`, `--rps`, and
`--workers` override those values in every phase. No additional Python dependency
is needed.
The profile controls:
- `plugins`: local archives or GitHub/marketplace request files, installed with
`lbctl plugin install ... --wait`.
- `resources`: Providers, models, knowledge bases, document ingestion, and
Pipelines, created in dependency order. `${resource_id}` refers to a UUID
returned by an earlier creation; `${workspace_uuid}` is the current Workspace.
Pipeline config patches merge with actual server defaults. Existing resources
can instead be supplied through `vars`, leaving `resources` empty.
- `cases`: CLI argument arrays and expected business text. Integer `weight`
values set the mix; `2/1/1/1` means `40/20/20/20`. An optional `body` object
becomes the case's request file. All workers share the weighted sequence.
- `phases`: each phase has a name, duration, target RPS, worker limit, and optional
cooldown. Multiple phases provide a step load. In-flight requests drain at the
end; the concurrency limit can reduce achieved RPS.
- `observe`: health URLs, host PIDs, PID files, and Docker container names/IDs.
Set PIDs for disjoint process trees to avoid counting descendants twice.
- `thresholds`: minimum achieved/target RPS ratio and maximum P95 seconds per
case, in addition to the existing business and health/resource checks. The
sample limits are starting acceptance settings, not established product SLOs.
- `cleanup`: `on-success` (default), `always`, or `never`. Only recorded resources
created by this run are deleted, in reverse order. Plugins and the existing
instance remain installed. Failed/unknown creations are not automatically
retried; inspect the instance if a creation outcome is unknown.
Failed runs retain resources by default. After inspection, clean them with the
same profile and credentials:
```sh
python3 tools/regression_suite.py cleanup \
--profile tools/regression_profiles/normal.json --output .qa/regression/run-001
```
Cleanup checks the Workspace against `owned.json` and records remaining items
after partial failures. Keep `owned.json` until cleanup completes. Each run also
contains `profile.json`, resolved `resources.json`, and one directory per phase
with the original request/resource records. The root `summary.json` includes
target/achieved RPS, phase results, threshold failures, and the final verdict.
The `regression_suite.py` entry reuses an existing instance and backend processes.
Use the complete lifecycle Job below to own deployment, credentials and backends.
Cases still use the existing text oracle and new-session CLI entry; continuous
conversations, IM adapter acceptance, and resource-leak verdicts need separate
scenarios. Test the suite plumbing on Linux with:
```sh
python3 -m unittest discover -s tools -p test_regression_suite.py -v
```
## Complete lifecycle Job
[`regression_job.py`](regression_job.py) owns the deployment as well as the suite.
Use [`regression_profiles/job.json`](regression_profiles/job.json) to pin the Core
Release and select the suite and plugin input directory. The job runs this chain:
1. `lbctl install --profile all` creates a fresh deployment inside the run directory.
2. A Compose override starts our mock backend on the deployment's private network
and injects a generated test-only API key. The Job binds its own CLI context.
3. The suite installs plugins, creates resources and ingests documents. Each case
runs once as warmup before measured traffic starts.
4. Configured load phases run with health, process-tree and container sampling.
5. Service-log tails and raw request/resource records are retained. Successful
Jobs stop the deployment, remove only sandboxes mounted under their own data
root, and use `lbctl uninstall --purge-data` to reclaim the deployment.
Both generation and embedding use our mock backend, with no external model
service or real model credentials. Plugin archives, the pinned Core image,
`lbctl`, Docker and Compose v2 are Job inputs/prerequisites. This wrapper uses
PyYAML, already included in LangBot's dependencies, to verify the CLI's deployment
record and recover ownership after an interrupted install.
Run on a Linux host with Docker access and permission to inspect the selected
container processes. Use an idle dedicated Docker daemon: current Box runtime
startup cleanup can remove sandboxes from another runtime. The Job refuses to
start when existing Box sandboxes are present. The current `lbctl` installer
also reserves host port 5401 and 2280–2285, independently of its HTTP port.
Set the plugin directory and the `lbctl` executable in `job.json`, then run:
```sh
uv run --no-sync python tools/regression_job.py validate --profile tools/regression_profiles/job.json
uv run --no-sync python tools/regression_job.py run \
--profile tools/regression_profiles/job.json --output .qa/regression/job-001
```
`--duration`, `--rps` and `--workers` override every load phase. Each output
directory must be new. Job cleanup policy is `on-success`, `always`, or `never`;
it governs the whole deployment, whereas suite cleanup governs API resources
when a suite runs on an existing deployment. `purge_data: false` preserves the
deployment data after removing containers. Failed Jobs retain their environment
by default; after inspection use:
```sh
uv run --no-sync python tools/regression_job.py cleanup \
--output .qa/regression/job-001
```
The root `summary.json` is the final Job verdict; `suite/` contains the measured
results, `mock/requests.jsonl` contains backend recordings, and `services.log`
contains up to 10,000 log lines per service. Generated credentials and deployment
inputs stay in `.private/`; share the result files, not that directory. A Job has
its own CLI locator/config and records its deployment identity before cleanup.
Changing that identity makes cleanup fail closed. Cleanup uses the saved Job
state and does not require the original profile or plugin input files.
The Job replaces the real embedding model with `mock-embedding`. This measures
LangBot/plugin/Box/RAG overhead, not real-model inference cost or retrieval
quality. Run the Job plumbing checks on Linux with:
```sh
uv run --no-sync python -m unittest discover -s tools -p 'test_*.py' -v
```
tools/mock_llm.py
+366
View File
@@ -0,0 +1,366 @@
#!/usr/bin/env python3
"""A deterministic OpenAI Chat Completions server for local regression tests."""
import argparse
import base64
import hashlib
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import math
from pathlib import Path
import re
import struct
import threading
import time
import uuid
MAX_BODY_BYTES = 1024 * 1024
USAGE = {'prompt_tokens': 10, 'completion_tokens': 5, 'total_tokens': 15}
def mock_embeddings(body):
"""Return deterministic normalized token hashes, not semantic model inference."""
inputs = body.get('input')
inputs = [inputs] if isinstance(inputs, str) else inputs
if (
not isinstance(inputs, list)
or not 1 <= len(inputs) <= 128
or any(not isinstance(text, str) or not text for text in inputs)
):
raise ValueError('embedding input must contain 1..128 non-empty strings')
dimensions = body.get('dimensions', 384)
if isinstance(dimensions, bool) or not isinstance(dimensions, int) or not 1 <= dimensions <= 4096:
raise ValueError('embedding dimensions must be an integer in 1..4096')
encoding = body.get('encoding_format', 'float')
if encoding not in ('float', 'base64'):
raise ValueError('embedding encoding_format must be float or base64')
data = []
for index, text in enumerate(inputs):
vector = [0.0] * dimensions
for token in re.findall(r'\w+', text.casefold()) or [text]:
digest = hashlib.sha256(token.encode()).digest()
vector[int.from_bytes(digest[:4], 'little') % dimensions] += 1 if digest[4] % 2 else -1
norm = math.sqrt(sum(value * value for value in vector))
if not norm:
vector[0], norm = 1.0, 1.0
vector = [value / norm for value in vector]
embedding = (
base64.b64encode(struct.pack(f'<{dimensions}f', *vector)).decode() if encoding == 'base64' else vector
)
data.append({'object': 'embedding', 'index': index, 'embedding': embedding})
return {'object': 'list', 'model': body['model'], 'data': data, 'usage': {'prompt_tokens': 10, 'total_tokens': 10}}
def select_message(body, scenario):
messages = body.get('messages')
if not isinstance(messages, list) or not messages or any(not isinstance(m, dict) for m in messages):
raise ValueError('messages must be a non-empty array of objects')
if not isinstance(body.get('stream', False), bool):
raise ValueError('stream must be a boolean')
if body.get('stream_options') is not None and not isinstance(body['stream_options'], dict):
raise ValueError('stream_options must be an object')
# Select from the current turn's history, never a shared request counter.
start = next(
(i for i in range(len(messages) - 1, -1, -1) if messages[i].get('role') == 'user'),
None,
)
if start is None:
raise ValueError('a user message is required')
turn = messages[start:]
calls = [i for i, m in enumerate(turn) if m.get('role') == 'assistant' and m.get('tool_calls')]
for index, position in enumerate(calls):
end = calls[index + 1] if index + 1 < len(calls) else len(turn)
requested = {call['id'] for call in turn[position]['tool_calls']}
returned = [m.get('tool_call_id') for m in turn[position + 1 : end] if m.get('role') == 'tool']
if len(returned) != len(requested) or set(returned) != requested:
raise ValueError('each tool call must have exactly one matching tool result')
steps = scenario['steps']
if len(calls) >= len(steps):
raise ValueError('tool history exceeds the configured scenario')
step = steps[len(calls)]
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}')
expected = step.get('expect_tool_output')
if expected is not None:
results = turn[calls[-1] + 1 :] if calls else []
result_text = '\n'.join(str(m.get('content', '')) for m in results if m.get('role') == 'tool')
if expected not in result_text:
raise ValueError(f'tool result must contain {expected!r}')
message = {'role': 'assistant', 'content': step.get('content')}
if tool_calls := step.get('tool_calls'):
tools = body.get('tools', [])
if not isinstance(tools, list) or any(
not isinstance(t, dict) or not isinstance(t.get('function'), dict) for t in tools
):
raise ValueError('tools must be an array of objects')
available = {t.get('function', {}).get('name') for t in tools if t.get('type') == 'function'}
if body.get('tool_choice') == 'none' or any(call['name'] not in available for call in tool_calls):
raise ValueError('scenario tools must be advertised and enabled by the caller')
message['tool_calls'] = [
{
'id': f'call_mock_{len(calls)}_{i}',
'type': 'function',
'function': {
'name': call['name'],
'arguments': json.dumps(call['arguments']),
},
}
for i, call in enumerate(tool_calls)
]
return message
class MockServer(ThreadingHTTPServer):
daemon_threads = True
def __init__(self, address, scenarios, record_path=None):
super().__init__(address, Handler)
self.scenarios = scenarios
self.record_path = record_path
self.record_lock = threading.Lock()
def record(self, entry):
if self.record_path:
with (
self.record_lock,
self.record_path.open('a', encoding='utf-8') as stream,
):
stream.write(json.dumps(entry, ensure_ascii=False) + '\n')
class Handler(BaseHTTPRequestHandler):
def log_message(self, *_args):
pass
def json_response(self, status, payload):
data = json.dumps(payload, ensure_ascii=False).encode()
self.send_response(status)
self.send_header('Content-Type', 'application/json')
self.send_header('Content-Length', str(len(data)))
self.end_headers()
self.wfile.write(data)
def do_GET(self):
if self.path == '/healthz':
self.json_response(200, {'status': 'ok'})
elif self.path == '/v1/models':
self.json_response(
200,
{
'object': 'list',
'data': [
{
'id': name,
'object': 'model',
'created': 0,
'owned_by': 'mock',
}
for name in [*self.server.scenarios, 'mock-embedding']
],
},
)
else:
self.json_response(
404,
{
'error': {
'message': 'unknown endpoint',
'type': 'invalid_request_error',
}
},
)
def do_POST(self):
entry = {'request_id': uuid.uuid4().hex, 'time': time.time(), 'path': self.path, 'status': 400}
try:
if self.path not in ('/v1/chat/completions', '/v1/embeddings'):
entry['status'] = 404
self.json_response(
404,
{
'error': {
'message': 'unknown endpoint',
'type': 'invalid_request_error',
}
},
)
return
self.connection.settimeout(10)
length = int(self.headers.get('Content-Length', 0))
if self.headers.get('Transfer-Encoding') or not 0 < length <= MAX_BODY_BYTES:
raise ValueError('a Content-Length between 1 and 1048576 is required')
body = json.loads(self.rfile.read(length))
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)
entry['status'] = status
if status != 200:
self.json_response(
status,
{
'error': {
'message': 'injected mock failure',
'type': 'api_error',
'code': str(status),
}
},
)
return
entry['response'] = message
base = {
'id': 'chatcmpl-' + entry['request_id'],
'created': int(time.time()),
'model': body['model'],
}
finish = 'tool_calls' if message.get('tool_calls') else 'stop'
if body.get('stream'):
self.stream_response(body, scenario, message, base, finish)
else:
self.json_response(
200,
{
**base,
'object': 'chat.completion',
'choices': [{'index': 0, 'message': message, 'finish_reason': finish}],
'usage': USAGE,
},
)
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'}})
except (BrokenPipeError, ConnectionResetError):
entry['disconnected'] = True
finally:
self.server.record(entry)
def stream_response(self, body, scenario, message, base, finish):
self.send_response(200)
self.send_header('Content-Type', 'text/event-stream')
self.send_header('Cache-Control', 'no-cache')
self.end_headers()
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))
send({'role': 'assistant'})
content = message.get('content') or ''
for offset in range(0, len(content), 8):
send({'content': content[offset : offset + 8]})
for index, call in enumerate(message.get('tool_calls', [])):
send(
{
'tool_calls': [
{
'index': index,
'id': call['id'],
'type': 'function',
'function': {
'name': call['function']['name'],
'arguments': '',
},
}
]
}
)
arguments = call['function']['arguments']
for offset in range(0, len(arguments), 16):
send(
{
'tool_calls': [
{
'index': index,
'function': {'arguments': arguments[offset : offset + 16]},
}
]
}
)
send({}, finish)
if (body.get('stream_options') or {}).get('include_usage'):
payload = {
**base,
'object': 'chat.completion.chunk',
'choices': [],
'usage': USAGE,
}
self.wfile.write(('data: ' + json.dumps(payload) + '\n\n').encode())
self.wfile.write(b'data: [DONE]\n\n')
self.wfile.flush()
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--host', default='127.0.0.1')
parser.add_argument('--port', type=int, default=18080)
parser.add_argument(
'--scenarios',
type=Path,
default=Path(__file__).with_name('mock_llm_scenarios.json'),
)
parser.add_argument(
'--record',
type=Path,
help='Append request/response JSONL without HTTP headers; use test data only',
)
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')
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:
server.serve_forever()
except KeyboardInterrupt:
pass
if __name__ == '__main__':
main()
tools/mock_llm_scenarios.json
+20
View File
@@ -0,0 +1,20 @@
{
"mock-chat": {
"steps": [{"content": "LANGBOT_CHAT_OK"}]
},
"mock-rag": {
"steps": [{"expect_prompt": "aurora-parser-rag-9137", "content": "LANGBOT_RAG_OK"}]
},
"mock-box": {
"steps": [
{"tool_calls": [{"name": "exec", "arguments": {"command": "printf LANGBOT_BOX_SENTINEL > /workspace/mock-llm.txt; cat /workspace/mock-llm.txt"}}]},
{"expect_tool_output": "LANGBOT_BOX_SENTINEL", "tool_calls": [{"name": "exec", "arguments": {"command": "cat /workspace/mock-llm.txt"}}]},
{"expect_tool_output": "LANGBOT_BOX_SENTINEL", "content": "LANGBOT_BOX_OK"}
]
},
"mock-401": {"status": 401, "steps": [{"content": "unused"}]},
"mock-429": {"status": 429, "steps": [{"content": "unused"}]},
"mock-503": {"status": 503, "steps": [{"content": "unused"}]},
"mock-slow": {"delay_seconds": 2, "steps": [{"content": "LANGBOT_SLOW_OK"}]},
"mock-slow-stream": {"chunk_delay_seconds": 0.2, "steps": [{"content": "LANGBOT_SLOW_STREAM_OK"}]}
}
tools/regression_job.py
+464
View File
@@ -0,0 +1,464 @@
#!/usr/bin/env python3
"""Own one complete lbctl deployment, mock backend, regression, and teardown."""
import argparse
import copy
import json
import os
from pathlib import Path
import re
import secrets
import shutil
import subprocess
import time
import urllib.error
import urllib.request
import yaml
from regression_suite import Suite, number, resolve, validate
KEY_ENV = 'LANGBOT_QA_API_KEY'
def write_json(path, data):
temporary = path.with_suffix(path.suffix + '.tmp')
temporary.write_text(json.dumps(data, ensure_ascii=False, indent=2))
temporary.chmod(0o600)
temporary.replace(path)
def load_profile(path):
profile = json.loads(path.read_text())
if not isinstance(profile, dict) or profile.get('version') != 1:
raise ValueError('job profile version must be 1')
if profile.keys() - {'version', 'lbctl', 'suite', 'deployment', 'vars', 'cleanup', 'purge_data', 'mock_image'}:
raise ValueError('unknown job profile field')
deployment = profile.get('deployment', {})
if not isinstance(deployment, dict) or deployment.keys() - {'version', 'port'}:
raise ValueError('deployment accepts version and port')
version = deployment.get('version', '')
if not isinstance(version, str) or not re.fullmatch(r'v[0-9]+\.[0-9]+\.[0-9]+', version):
raise ValueError('deployment.version must pin a stable vX.Y.Z Release')
number(deployment.get('port', 5300), 'deployment port', minimum=1, integer=True)
if deployment.get('port', 5300) > 65535:
raise ValueError('deployment port must be <= 65535')
if profile.get('cleanup', 'on-success') not in ('always', 'on-success', 'never'):
raise ValueError('cleanup must be always/on-success/never')
if not isinstance(profile.get('purge_data', True), bool):
raise ValueError('purge_data must be boolean')
suite_path = Path(profile['suite']).expanduser()
suite_path = suite_path if suite_path.is_absolute() else path.parent / suite_path
suite = json.loads(suite_path.read_text())
suite.setdefault('vars', {}).update(profile.get('vars', {}))
suite['vars'].update(
mock_base_url='http://regression_mock:18080/v1', embedding_base_url='http://regression_mock:18080/v1'
)
suite['cli'] = {'config': 'job-generated.yaml'}
for item in suite.get('resources', []):
if item['kind'] == 'model' and item.get('type') == 'embedding':
item['body']['name'] = 'mock-embedding'
validate(suite)
return profile, suite, suite_path.resolve()
class Job:
def __init__(self, profile, suite_profile, suite_path, output):
self.profile = profile
self.suite_profile = copy.deepcopy(suite_profile)
self.suite_path = suite_path
self.output = output.resolve()
self.private = self.output / '.private'
self.deploy = self.private / 'deployment'
self.state_path = self.output / 'state.json'
self.cli_config = self.private / 'lbctl.yaml'
self.env = {**os.environ, 'XDG_CONFIG_HOME': str(self.private / 'config')}
self.cli = [profile.get('lbctl', 'lbctl'), '--config', str(self.cli_config), '-o', 'json']
self.redactions = []
self.state = {'job_id': secrets.token_hex(8), 'record': None, 'bound': False, 'phase': 'created'}
def redact(self, value):
for secret in self.redactions:
value = value.replace(secret, '[REDACTED]')
return value
def save(self):
write_json(self.state_path, self.state)
def recover_record(self):
if (self.private / 'job-id').read_text().strip() != self.state['job_id']:
raise RuntimeError('Job ownership marker differs from state.json')
path = self.deploy / 'deployment.yaml'
if not path.exists():
if self.state.get('record'):
raise RuntimeError('deployment record disappeared; refusing cleanup')
return
if path.is_symlink():
raise RuntimeError('deployment record must not be a symlink')
try:
record = yaml.safe_load(path.read_text())
except yaml.YAMLError as exc:
raise ValueError('invalid deployment YAML') from exc
if not isinstance(record, dict) or not record.get('deployment_id'):
raise RuntimeError('invalid deployment record')
if (
Path(record['compose_file']).resolve().parent != self.deploy
or Path(record['data_dir']).resolve() != self.deploy / 'data'
or Path(record['env_file']).resolve().parent != self.deploy
):
raise RuntimeError('deployment record is outside this Job')
if self.state.get('record') and record['deployment_id'] != self.state['record']['deployment_id']:
raise RuntimeError('deployment identity changed; refusing cleanup')
self.state['record'] = record
self.save()
def load_redactions(self):
if self.state.get('record'):
path = Path(self.state['record']['env_file'])
if path.exists():
for line in path.read_text().splitlines():
name, separator, value = line.partition('=')
if separator and any(word in name for word in ('TOKEN', 'KEY')) and len(value) >= 16:
self.redactions.append(value)
def command(self, args, timeout=900):
result = subprocess.run(args, capture_output=True, text=True, env=self.env, timeout=timeout)
return result.returncode, self.redact(result.stdout), self.redact(result.stderr)
def checked(self, args, timeout=900):
code, stdout, stderr = self.command(args, timeout)
if code:
raise RuntimeError((stderr + stdout)[-4000:])
return stdout
def cli_call(self, args):
code, stdout, stderr = self.command([*self.cli, *args])
result = json.loads(stdout) if stdout.strip() else {}
data = result.get('data', {})
if args[0] == 'install' and isinstance(data, dict) and data.get('record'):
record = data['record']
if (
Path(record['compose_file']).resolve().parent != self.deploy
or Path(record['data_dir']).resolve() != self.deploy / 'data'
):
raise RuntimeError('install returned a deployment outside this Job')
self.state['record'] = record
self.save() # Installation may have changed state even when its final health check failed.
if code:
raise RuntimeError((stderr + stdout)[-4000:])
return data
def compose(self, *args, override=False):
record = self.state['record']
command = ['docker', 'compose', '--project-name', record['compose_project'], '--file', record['compose_file']]
if override:
command += ['--file', str(self.private / 'compose.override.json')]
command += ['--env-file', record['env_file'], '--profile', 'all', *args]
return self.checked(command)
def project_containers(self):
project = self.state['record']['compose_project']
output = self.checked(
[
'docker',
'ps',
'-a',
'--filter',
f'label=com.docker.compose.project={project}',
'--format',
'{{.ID}} {{.Label "com.docker.compose.service"}}',
]
)
return {service: container for container, service in (line.split() for line in output.splitlines() if line)}
def address(self, container):
networks = json.loads(
self.checked(['docker', 'inspect', '--format', '{{json .NetworkSettings.Networks}}', container])
)
addresses = [network['IPAddress'] for network in networks.values() if network.get('IPAddress')]
if len(addresses) != 1:
raise RuntimeError('expected one private container network')
return addresses[0]
@staticmethod
def ready(url):
deadline = time.monotonic() + 120
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
while time.monotonic() < deadline:
try:
with opener.open(url, timeout=2) as response:
payload = json.load(response)
if response.status == 200 and payload.get('ready') is not False and payload.get('code', 0) == 0:
return
except (urllib.error.URLError, TimeoutError, OSError, ValueError):
pass
time.sleep(0.5)
raise RuntimeError(f'readiness deadline exceeded: {url}')
def prepare(self):
variables = {
**self.suite_profile.get('vars', {}),
'profile_dir': str(self.suite_path.parent),
'run_dir': str(self.output),
}
inputs = [plugin['file'] for plugin in self.suite_profile.get('plugins', [])]
inputs += [
document['file']
for item in self.suite_profile.get('resources', [])
for document in item.get('documents', [])
]
for name in inputs:
path = Path(resolve(name, variables)).expanduser()
path = path if path.is_absolute() else self.suite_path.parent / path
if not path.is_file():
raise ValueError(f'Job input file is missing: {path}')
self.state['phase'] = 'install'
self.save()
# Current Box runtimes reap sandboxes from previous runtime IDs on startup.
# Do not start one alongside another user's live sandbox on this daemon.
boxes = self.checked(['docker', 'ps', '-a', '--filter', 'label=langbot.box=true', '--format', '{{.ID}}'])
if boxes.strip():
raise RuntimeError('Docker daemon has existing Box sandboxes; use an idle dedicated test daemon')
key = secrets.token_urlsafe(32)
self.env[KEY_ENV] = key
self.redactions.append(key)
(self.private / 'api-key').write_text(key)
(self.private / 'api-key').chmod(0o600)
deployment = self.profile['deployment']
self.cli_call(
[
'install',
'--dir',
str(self.deploy),
'--version',
deployment['version'],
'--profile',
'all',
'--port',
str(deployment.get('port', 5300)),
]
)
self.recover_record()
self.load_redactions()
mock_results = self.output / 'mock'
mock_results.mkdir(mode=0o700)
override = {
'services': {
'langbot': {
'environment': {'API__GLOBAL_API_KEY': key},
'volumes': [f'{self.deploy / "data/box"}:{self.deploy / "data/box"}'],
},
'regression_mock': {
'image': self.profile.get('mock_image', 'python:3.12-alpine'),
'volumes': [f'{Path(__file__).parent.resolve()}:/regression:ro', f'{mock_results}:/results'],
'command': [
'python',
'/regression/mock_llm.py',
'--host',
'0.0.0.0',
'--port',
'18080',
'--record',
'/results/requests.jsonl',
],
'networks': ['langbot_network'],
'labels': {'langbot.regression_job': self.state['job_id']},
},
},
}
write_json(self.private / 'compose.override.json', override)
self.compose('up', '-d', override=True)
containers = self.project_containers()
for service in ('langbot', 'langbot_plugin_runtime', 'langbot_box', 'regression_mock'):
if service not in containers:
raise RuntimeError(f'missing deployment service: {service}')
endpoint = self.state['record']['endpoint']
health = {
'core': endpoint + '/healthz',
'mock': f'http://{self.address(containers["regression_mock"])}:18080/healthz',
'box': f'http://{self.address(containers["langbot_box"])}:5410/readyz',
}
for url in health.values():
self.ready(url)
identity = self.cli_call(['--endpoint', endpoint, 'whoami'])
self.cli_call(
[
'context',
'add',
'regression',
'--endpoint',
endpoint,
'--api-key-env',
KEY_ENV,
'--expect-workspace',
identity['workspace_uuid'],
]
)
self.cli_call(['context', 'use', 'regression'])
self.state.update(
bound=True, workspace_uuid=identity['workspace_uuid'], containers=containers, phase='resources'
)
self.save()
profile = self.suite_profile
profile['cli'] = {'executable': self.cli[0], 'config': str(self.cli_config)}
profile['observe'] = {'health': health, 'pids': {}, 'containers': list(containers.values())}
for service, container in containers.items():
pid = int(self.checked(['docker', 'inspect', '--format', '{{.State.Pid}}', container]).strip())
profile['observe']['pids'][service] = pid
destination = self.output / 'suite'
destination.mkdir(mode=0o700)
suite = Suite(profile, self.suite_path, destination, env=self.env)
suite.write('profile.json', profile)
suite.prepare()
# Prime every configured business case before measuring steady load.
for case in profile['cases']:
case = resolve(case, suite.variables)
data = suite.call(case['args'], case.get('body'))
if case['expect'] not in json.dumps(data, ensure_ascii=False):
raise RuntimeError(f'warmup oracle failed: {case["name"]}')
profile['observe']['containers'] += self.owned_sandboxes()
suite.write('profile.json', profile)
return suite
def owned_sandboxes(self):
record = self.state.get('record')
if not record:
return []
root = (Path(record['data_dir']) / 'box').resolve()
candidates = self.checked(
['docker', 'ps', '-a', '--filter', 'label=langbot.box=true', '--format', '{{.ID}}']
).splitlines()
owned = []
for container in candidates:
mounts = json.loads(self.checked(['docker', 'inspect', '--format', '{{json .Mounts}}', container]))
if any(Path(mount['Source']).resolve().is_relative_to(root) for mount in mounts if mount.get('Source')):
owned.append(container)
return owned
def collect(self):
self.recover_record()
self.load_redactions()
if self.state.get('record'):
logs = self.compose('logs', '--no-color', '--tail', '10000')
(self.output / 'services.log').write_text(logs)
def cleanup(self):
self.recover_record()
self.load_redactions()
record = self.state.get('record')
if self.state['phase'] == 'cleaned':
if record and self.project_containers():
raise RuntimeError('containers appeared after Job cleanup; inspect deployment before acting')
return
if record:
if (
Path(record['compose_file']).resolve().parent != self.deploy
or Path(record['data_dir']).resolve() != self.deploy / 'data'
):
raise RuntimeError('cleanup refused: deployment is outside this Job')
if self.state['bound']:
self.cli_call(['stop'])
else:
self.compose('stop', override=(self.private / 'compose.override.json').exists())
sandboxes = self.owned_sandboxes()
if sandboxes:
self.checked(['docker', 'rm', '-f', *sandboxes])
if self.state['bound']:
self.cli_call(
['uninstall', '--yes', *(['--purge-data'] if self.profile.get('purge_data', True) else [])]
)
else:
self.compose('down', '--remove-orphans', override=(self.private / 'compose.override.json').exists())
if self.profile.get('purge_data', True) and (self.deploy / 'data').exists():
shutil.rmtree(self.deploy / 'data')
for path in (self.private / 'api-key', self.private / 'compose.override.json'):
path.unlink(missing_ok=True)
self.state['phase'] = 'cleaned'
self.save()
def run(self):
self.output.mkdir(parents=True, exist_ok=False, mode=0o700)
self.private.mkdir(mode=0o700)
(self.private / 'job-id').write_text(self.state['job_id'])
write_json(self.output / 'job-profile.json', self.profile)
self.save()
result = {'errors': [], 'completed': False}
try:
suite = self.prepare()
self.state['phase'] = 'traffic'
self.save()
summary = suite.run()
summary['passed'] = not summary['errors']
suite.write('summary.json', summary)
result.update(completed=True, suite=summary, errors=list(summary['errors']))
self.state['phase'] = 'finished'
self.save()
except KeyboardInterrupt:
result['errors'].append('Job interrupted; in-flight outcomes may be unknown')
except (RuntimeError, ValueError, OSError, KeyError, subprocess.TimeoutExpired) as exc:
result['errors'].append(self.redact(str(exc)))
finally:
if not result['completed'] and not result['errors']:
result['errors'].append('Job aborted before completion')
try:
self.collect()
except (RuntimeError, ValueError, OSError, subprocess.TimeoutExpired) as exc:
result['errors'].append('artifact collection: ' + self.redact(str(exc)))
policy = self.profile.get('cleanup', 'on-success')
if policy == 'always' or (policy == 'on-success' and not result['errors']):
try:
self.cleanup()
except (RuntimeError, ValueError, OSError, KeyError, subprocess.TimeoutExpired) as exc:
result['errors'].append('cleanup: ' + self.redact(str(exc)))
result.update(passed=not result['errors'], phase=self.state['phase'])
write_json(self.output / 'summary.json', result)
return result
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('action', choices=('validate', 'run', 'cleanup'))
parser.add_argument('--profile', type=Path)
parser.add_argument('--output', type=Path)
parser.add_argument('--duration', type=float)
parser.add_argument('--rps', type=float)
parser.add_argument('--workers', type=int)
args = parser.parse_args()
try:
if args.action == 'cleanup':
if args.output is None:
raise ValueError('--output is required for cleanup')
profile = json.loads((args.output / 'job-profile.json').read_text())
job = Job(profile, {}, args.output / 'unused-profile.json', args.output)
job.state = json.loads(job.state_path.read_text())
if (job.private / 'api-key').exists():
key = (job.private / 'api-key').read_text()
job.env[KEY_ENV] = key
job.redactions.append(key)
job.cleanup()
return
if args.profile is None:
raise ValueError('--profile is required for run/validate')
profile, suite, path = load_profile(args.profile.resolve())
for phase in suite['phases']:
for flag, field in [('duration', 'duration_seconds'), ('rps', 'rps'), ('workers', 'workers')]:
if getattr(args, flag) is not None:
phase[field] = getattr(args, flag)
validate(suite)
if args.action == 'validate':
print('Job profile valid; no deployment or backend changed.')
return
if args.output is None:
raise ValueError('--output is required for run')
job = Job(profile, suite, path, args.output)
result = job.run()
print(json.dumps(result, ensure_ascii=False, indent=2))
if not result['passed']:
raise SystemExit(1)
except (RuntimeError, ValueError, OSError, KeyError, subprocess.TimeoutExpired) as exc:
parser.exit(1, f'{type(exc).__name__}: {exc}\n')
if __name__ == '__main__':
main()
tools/regression_profiles/job.json
+10
View File
@@ -0,0 +1,10 @@
{
"version": 1,
"lbctl": "lbctl",
"suite": "normal.json",
"deployment": {"version": "v4.11.0", "port": 15351},
"vars": {"plugin_dir": "/path/to/plugin-zips"},
"mock_image": "python:3.12-alpine",
"cleanup": "on-success",
"purge_data": true
}
tools/regression_profiles/normal.json
+111
View File
@@ -0,0 +1,111 @@
{
"version": 1,
"cli": {"executable": "lbctl", "config": "./lbctl.yaml"},
"vars": {
"plugin_dir": "/path/to/plugin-zips",
"mock_base_url": "http://127.0.0.1:18080/v1",
"embedding_base_url": "http://127.0.0.1:15410/v1",
"core_health": "http://127.0.0.1:5300/healthz"
},
"plugins": [
{"source": "local", "file": "${plugin_dir}/LocalAgent.zip"},
{"source": "local", "file": "${plugin_dir}/LangRAG.zip"},
{"source": "local", "file": "${plugin_dir}/GeneralParsers.zip"}
],
"resources": [
{"id": "mock_provider", "kind": "provider", "body": {
"name": "regression mock", "requester": "openai-chat-completions",
"base_url": "${mock_base_url}", "api_keys": ["test-only"]
}},
{"id": "embedding_provider", "kind": "provider", "body": {
"name": "regression embedding", "requester": "openai-chat-completions",
"base_url": "${embedding_base_url}", "api_keys": ["test-only"]
}},
{"id": "chat_model", "kind": "model", "type": "llm", "body": {
"name": "mock-chat", "provider_uuid": "${mock_provider}",
"abilities": ["func_call"], "context_length": 8192,
"extra_args": {"request_policy": {"retries": 0}}
}},
{"id": "box_model", "kind": "model", "type": "llm", "body": {
"name": "mock-box", "provider_uuid": "${mock_provider}",
"abilities": ["func_call"], "context_length": 8192,
"extra_args": {"request_policy": {"retries": 0}}
}},
{"id": "rag_model", "kind": "model", "type": "llm", "body": {
"name": "mock-rag", "provider_uuid": "${mock_provider}",
"abilities": ["func_call"], "context_length": 8192,
"extra_args": {"request_policy": {"retries": 0}}
}},
{"id": "embedding_model", "kind": "model", "type": "embedding", "body": {
"name": "all-MiniLM-L6-v2", "provider_uuid": "${embedding_provider}"
}},
{"id": "knowledge_base", "kind": "knowledge-base", "body": {
"name": "regression parser golden", "knowledge_engine_plugin_id": "langbot-team/LangRAG",
"creation_settings": {"embedding_model_uuid": "${embedding_model}", "index_type": "chunk", "chunk_size": 512, "overlap": 50},
"retrieval_settings": {"search_type": "vector", "top_k": 5, "query_rewrite": "off", "rerank": "off"}
}, "documents": [{
"file": "../../skills/skills/langbot-testing/fixtures/rag/parser-golden.html",
"parser_plugin_id": "langbot-team/GeneralParsers"
}]},
{"id": "chat_pipeline", "kind": "pipeline", "body": {
"name": "regression chat", "description": "Normal deterministic chat",
"config": {
"ai": {
"runner": {"id": "plugin:langbot-team/LocalAgent/default", "expire-time": 0},
"runner_config": {"plugin:langbot-team/LocalAgent/default": {
"model": {"primary": "${chat_model}", "fallbacks": []},
"prompt": [{"role": "system", "content": "You are a QA regression assistant."}],
"knowledge-bases": [], "box-enabled": false, "timeout": 30
}}
},
"safety": {"rate-limit": {"limitation": 10000}},
"output": {"force-delay": {"min": 0, "max": 0}}
}
}},
{"id": "box_pipeline", "kind": "pipeline", "body": {
"name": "regression Box", "description": "Real Box write and read",
"config": {
"ai": {
"runner": {"id": "plugin:langbot-team/LocalAgent/default", "expire-time": 0},
"runner_config": {"plugin:langbot-team/LocalAgent/default": {
"model": {"primary": "${box_model}", "fallbacks": []},
"prompt": [{"role": "system", "content": "You are a QA regression assistant."}],
"knowledge-bases": [], "box-enabled": true, "box-session-id-template": "{global}",
"timeout": 30, "max-tool-iterations": 5
}}
},
"safety": {"rate-limit": {"limitation": 10000}},
"output": {"force-delay": {"min": 0, "max": 0}}
}
}},
{"id": "rag_pipeline", "kind": "pipeline", "body": {
"name": "regression RAG", "description": "Retrieved model input oracle",
"config": {
"ai": {
"runner": {"id": "plugin:langbot-team/LocalAgent/default", "expire-time": 0},
"runner_config": {"plugin:langbot-team/LocalAgent/default": {
"model": {"primary": "${rag_model}", "fallbacks": []},
"prompt": [{"role": "system", "content": "You are a QA regression assistant."}],
"knowledge-bases": ["${knowledge_base}"], "box-enabled": false,
"timeout": 30, "retrieval-top-k": 5
}}
},
"safety": {"rate-limit": {"limitation": 10000}},
"output": {"force-delay": {"min": 0, "max": 0}}
}
}}
],
"cases": [
{"name": "chat", "weight": 2, "args": ["pipeline", "run", "${chat_pipeline}", "--message", "Run the normal QA scenario", "--timeout", "90s"], "expect": "LANGBOT_CHAT_OK"},
{"name": "box", "weight": 1, "args": ["pipeline", "run", "${box_pipeline}", "--message", "Run the normal QA scenario", "--timeout", "90s"], "expect": "LANGBOT_BOX_OK"},
{"name": "rag_chat", "weight": 1, "args": ["pipeline", "run", "${rag_pipeline}", "--message", "What is the documented parser-rag sentinel?", "--timeout", "90s"], "expect": "LANGBOT_RAG_OK"},
{"name": "rag_retrieve", "weight": 1, "args": ["knowledge-base", "retrieve", "${knowledge_base}"], "body": {"query": "What is the parser-rag golden sentinel and parser-engine pair?", "retrieval_settings": {"search_type": "vector", "top_k": 5}}, "expect": "aurora-parser-rag-9137"}
],
"phases": [{"name": "normal", "duration_seconds": 600, "rps": 2, "workers": 4, "cooldown_seconds": 30}],
"observe": {
"health": {"core": "${core_health}", "mock": "http://127.0.0.1:18080/healthz", "box": "http://127.0.0.1:5410/readyz"},
"pids": {}, "pid_files": {}, "containers": []
},
"cleanup": "on-success",
"thresholds": {"min_rps_ratio": 0.95, "max_p95_seconds": {"chat": 3, "box": 6, "rag_chat": 4, "rag_retrieve": 1}}
}
tools/regression_suite.py
+409
View File
@@ -0,0 +1,409 @@
#!/usr/bin/env python3
"""Prepare an lbctl regression from JSON, run load phases, and clean owned resources."""
import argparse
import copy
import json
import math
import os
from pathlib import Path
import re
import subprocess
import sys
KINDS = {'provider', 'model', 'pipeline', 'knowledge-base'}
REFERENCE = re.compile(r'\$\{([a-zA-Z][a-zA-Z0-9_-]*)\}')
def resolve(value, variables):
if isinstance(value, dict):
return {key: resolve(item, variables) for key, item in value.items()}
if isinstance(value, list):
return [resolve(item, variables) for item in value]
if isinstance(value, str):
return REFERENCE.sub(lambda match: str(variables[match[1]]), value)
return value
def merge(base, patch):
result = copy.deepcopy(base)
for key, value in patch.items():
result[key] = (
merge(result[key], value) if isinstance(value, dict) and isinstance(result.get(key), dict) else value
)
return result
def number(value, label, *, minimum=0, integer=False):
if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
raise ValueError(f'{label} must be a finite number')
if value < minimum or (integer and not isinstance(value, int)):
raise ValueError(f'{label} must be {"an integer " if integer else ""}>= {minimum}')
def validate(profile):
if not isinstance(profile, dict) or profile.get('version') != 1:
raise ValueError('profile version must be 1')
allowed = {'version', 'cli', 'vars', 'plugins', 'resources', 'cases', 'phases', 'observe', 'cleanup', 'thresholds'}
if extra := profile.keys() - allowed:
raise ValueError(f'unknown profile fields: {sorted(extra)}')
for field in ('cli', 'vars', 'observe', 'thresholds'):
if field in profile and not isinstance(profile[field], dict):
raise ValueError(f'{field} must be an object')
for field in ('plugins', 'resources', 'cases', 'phases'):
if field in profile and (
not isinstance(profile[field], list) or any(not isinstance(item, dict) for item in profile[field])
):
raise ValueError(f'{field} must be an array of objects')
cli = profile.get('cli', {})
if not isinstance(cli, dict) or cli.keys() - {'config', 'executable'}:
raise ValueError('cli accepts config and executable')
if not isinstance(cli.get('config'), str) or not cli['config']:
raise ValueError('cli.config must name a Workspace-bound lbctl config')
variables = {'profile_dir': '.', 'run_dir': '.', 'workspace_uuid': 'workspace'}
supplied = profile.get('vars', {})
if not isinstance(supplied, dict) or variables.keys() & supplied.keys():
raise ValueError('vars must be an object without reserved names')
if any(not isinstance(value, (str, int, float)) or '${' in str(value) for value in supplied.values()):
raise ValueError('vars values must be scalar literals, without nested references')
variables.update(supplied)
for plugin in profile.get('plugins', []):
if plugin.keys() - {'source', 'file'}:
raise ValueError('plugins accept source and file')
if plugin.get('source') not in {'local', 'github', 'marketplace'} or not plugin.get('file'):
raise ValueError('plugins need source local/github/marketplace and file')
resolve(plugin, variables)
for item in profile.get('resources', []):
if item.keys() - {'id', 'kind', 'type', 'body', 'documents'}:
raise ValueError('unknown resource field')
name = item.get('id', '')
if not re.fullmatch(r'[a-zA-Z][a-zA-Z0-9_-]*', name) or name in variables:
raise ValueError(f'invalid or duplicate resource id: {name}')
if item.get('kind') not in KINDS or not isinstance(item.get('body'), dict):
raise ValueError(f'{name}: resource needs kind and body')
if 'uuid' in item['body']:
raise ValueError(f'{name}: resources must create new objects; body.uuid is forbidden')
if item['kind'] == 'model' and item.get('type') not in {'llm', 'embedding', 'rerank'}:
raise ValueError(f'{name}: model type must be llm/embedding/rerank')
resolve(item['body'], variables) # References must point backwards; never guess missing UUIDs.
variables[name] = name
documents = item.get('documents', [])
if not isinstance(documents, list) or any(not isinstance(doc, dict) for doc in documents):
raise ValueError(f'{name}: documents must be an array of objects')
for document in documents:
if item['kind'] != 'knowledge-base' or not document.get('file') or not document.get('parser_plugin_id'):
raise ValueError(f'{name}: documents need file and parser_plugin_id on a knowledge-base')
resolve(document, variables)
cases = profile.get('cases')
if not isinstance(cases, list) or not cases:
raise ValueError('cases must be a non-empty array')
names = set()
for case in cases:
if case.keys() - {'name', 'weight', 'args', 'expect', 'body'}:
raise ValueError('unknown case field')
if not isinstance(case.get('name'), str) or not case['name'] or case['name'] in names:
raise ValueError('case names must be non-empty and unique')
names.add(case['name'])
if (
not isinstance(case.get('args'), list)
or not case['args']
or not all(isinstance(x, str) for x in case['args'])
):
raise ValueError('case args must be a non-empty string array')
if not isinstance(case.get('expect'), str) or not case['expect']:
raise ValueError('case expect must be a non-empty string')
number(case.get('weight', 1), 'case weight', minimum=1, integer=True)
resolve(case, variables)
if 'body' in case and (not isinstance(case['body'], dict) or '--file' in case['args']):
raise ValueError('case body must be an object and cannot coexist with --file')
if sum(case.get('weight', 1) for case in cases) > 1000:
raise ValueError('total case weight must be <= 1000')
phases = profile.get('phases')
if not isinstance(phases, list) or not phases:
raise ValueError('phases must be a non-empty array')
names = set()
for phase in phases:
name = phase.get('name', '')
if not re.fullmatch(r'[a-zA-Z][a-zA-Z0-9_-]*', name) or name in names:
raise ValueError('phase names must be unique simple names')
names.add(name)
if phase.keys() - {'name', 'duration_seconds', 'rps', 'workers', 'cooldown_seconds'}:
raise ValueError(f'{name}: unknown phase field')
number(phase.get('duration_seconds'), 'duration_seconds', minimum=0.001)
number(phase.get('rps'), 'rps', minimum=0.001)
number(phase.get('workers'), 'workers', minimum=1, integer=True)
number(phase.get('cooldown_seconds', 0), 'cooldown_seconds')
if phase['workers'] > 32:
raise ValueError('workers must be <= 32')
if profile.get('cleanup', 'on-success') not in {'always', 'on-success', 'never'}:
raise ValueError('cleanup must be always/on-success/never')
thresholds = profile.get('thresholds', {})
if extra := thresholds.keys() - {'min_rps_ratio', 'max_p95_seconds'}:
raise ValueError(f'unknown thresholds: {sorted(extra)}')
number(thresholds.get('min_rps_ratio', 0), 'min_rps_ratio')
if thresholds.get('min_rps_ratio', 0) > 1:
raise ValueError('min_rps_ratio must be <= 1')
if not isinstance(thresholds.get('max_p95_seconds', {}), dict):
raise ValueError('max_p95_seconds must map case names to seconds')
for name, value in thresholds.get('max_p95_seconds', {}).items():
if name not in {case['name'] for case in cases}:
raise ValueError(f'unknown case in max_p95_seconds: {name}')
number(value, 'max_p95_seconds', minimum=0.001)
observed = resolve(profile.get('observe', {}), variables)
if not isinstance(observed, dict) or observed.keys() - {'health', 'pids', 'pid_files', 'containers'}:
raise ValueError('observe accepts health, pids, pid_files, containers')
if not isinstance(observed.get('health'), dict) or not observed['health']:
raise ValueError('observe.health must contain at least one health endpoint')
for name, url in observed['health'].items():
if not isinstance(name, str) or not isinstance(url, str) or not url.startswith(('http://', 'https://')):
raise ValueError('health endpoints must map names to HTTP(S) URLs')
for field in ('pids', 'pid_files'):
if not isinstance(observed.get(field, {}), dict):
raise ValueError(f'observe.{field} must be an object')
for pid in observed.get('pids', {}).values():
number(pid, 'PID', minimum=1, integer=True)
if not isinstance(observed.get('containers', []), list) or any(
not isinstance(item, str) or not item for item in observed.get('containers', [])
):
raise ValueError('observe.containers must be an array of container names/IDs')
class Suite:
def __init__(self, profile, profile_path, output, env=None):
self.profile = profile
self.output = output.resolve()
self.variables = {
**profile.get('vars', {}),
'profile_dir': str(profile_path.parent.resolve()),
'run_dir': str(self.output),
}
self.cli = [
profile.get('cli', {}).get('executable', 'lbctl'),
'--config',
self.path(profile['cli']['config']),
'-o',
'json',
]
self.owned = []
self.env = env
self.secrets = [value for key, value in (env or os.environ).items() if 'API_KEY' in key and value]
def path(self, value):
path = Path(resolve(value, self.variables)).expanduser()
return str(path if path.is_absolute() else Path(self.variables['profile_dir']) / path)
def write(self, name, value):
path = self.output / name
path.write_text(json.dumps(value, ensure_ascii=False, indent=2))
path.chmod(0o600)
return str(path)
def call(self, args, body=None):
command = [*self.cli, *args]
if body is not None:
command += ['--file', self.write('request.json', body)]
try:
result = subprocess.run(command, capture_output=True, text=True, timeout=360, env=self.env)
except subprocess.TimeoutExpired as exc:
raise RuntimeError('lbctl timed out; outcome unknown, no automatic retry') from exc
if result.returncode:
error = result.stdout + result.stderr
for secret in self.secrets:
error = error.replace(secret, '[REDACTED]')
raise RuntimeError(error[:3000])
return json.loads(result.stdout)['data']
def journal(self):
self.write('owned.json', {'workspace_uuid': self.variables['workspace_uuid'], 'resources': self.owned})
def prepare(self):
identity = self.call(['whoami'])
self.variables['workspace_uuid'] = identity['workspace_uuid']
self.journal()
for plugin in self.profile.get('plugins', []):
self.call(['plugin', 'install', plugin['source'], '--file', self.path(plugin['file']), '--wait'])
for item in self.profile.get('resources', []):
kind = item['kind']
body = resolve(item['body'], self.variables)
config = body.pop('config', None) if kind == 'pipeline' else None
args = [kind, 'apply' if kind == 'pipeline' else 'create']
if kind == 'model':
args += ['--type', item['type']]
created = self.call(args, body)
created = created.get('pipeline', created) if kind == 'pipeline' else created
uuid = created['uuid']
self.variables[item['id']] = uuid
self.owned.append({'kind': kind, 'uuid': uuid, **({'type': item['type']} if kind == 'model' else {})})
self.journal() # Record ownership before any later configuration or ingestion can fail.
if config is not None:
current = self.call(['pipeline', 'get', uuid])
current = current.get('pipeline', current)
self.call(
['pipeline', 'apply'],
{
'uuid': uuid,
'name': current['name'],
'description': current['description'],
'config': merge(current['config'], config),
},
)
for document in item.get('documents', []):
self.call(
[
'knowledge-base',
'ingest',
uuid,
'--file',
self.path(document['file']),
'--parser-plugin-id',
document['parser_plugin_id'],
'--wait',
]
)
self.write(
'resources.json',
{key: value for key, value in self.variables.items() if key not in self.profile.get('vars', {})},
)
def cleanup(self):
saved = json.loads((self.output / 'owned.json').read_text())
identity = self.call(['whoami'])
if saved['workspace_uuid'] != identity['workspace_uuid']:
raise RuntimeError('cleanup refused: Workspace differs from owned.json')
failures = []
for resource in reversed(saved['resources'].copy()):
args = [resource['kind'], 'delete', resource['uuid'], '--yes']
if resource['kind'] == 'model':
args += ['--type', resource['type']]
try:
self.call(args)
except (RuntimeError, ValueError, OSError) as exc:
failures.append(str(exc))
else:
saved['resources'].remove(resource)
self.write('owned.json', saved)
return failures
def run(self):
cases = []
for original in self.profile['cases']:
case = resolve(original, self.variables)
if 'body' in case:
case['args'] += ['--file', self.write(f'case-{len(cases)}.json', case.pop('body'))]
cases.extend([{key: case[key] for key in ('name', 'args', 'expect')}] * case.get('weight', 1))
cases_path = self.write('cases.json', cases)
observed = resolve(self.profile.get('observe', {}), self.variables)
observe_args = []
for name, pid in observed.get('pids', {}).items():
observe_args += ['--pid', f'{name}={int(pid)}']
for name, path in observed.get('pid_files', {}).items():
observe_args += ['--pid', f'{name}={int(Path(self.path(path)).read_text().strip())}']
for name, url in observed.get('health', {}).items():
observe_args += ['--health', f'{name}={url}']
for container in observed.get('containers', []):
observe_args += ['--container', container]
results = []
errors = []
for phase in self.profile['phases']:
destination = self.output / phase['name']
command = [
sys.executable,
str(Path(__file__).with_name('regression_traffic.py')),
'--lbctl',
self.cli[0],
'--config',
self.cli[2],
'--cases',
cases_path,
'--output',
str(destination),
'--duration',
str(phase['duration_seconds']),
'--rps',
str(phase['rps']),
'--workers',
str(phase['workers']),
'--cooldown',
str(phase.get('cooldown_seconds', 0)),
'--shared-case-sequence',
*observe_args,
]
with (self.output / f'{phase["name"]}.log').open('w') as log:
result = subprocess.run(command, stdout=log, stderr=subprocess.STDOUT, env=self.env)
summary = json.loads((destination / 'summary.json').read_text())
summary['name'] = phase['name']
summary['achieved_rps'] = summary['requests'] / summary['load_elapsed_seconds']
results.append(summary)
if result.returncode:
errors.append(f'{phase["name"]}: business or health/resource checks failed')
thresholds = self.profile.get('thresholds', {})
if summary['achieved_rps'] < phase['rps'] * thresholds.get('min_rps_ratio', 0):
errors.append(f'{phase["name"]}: achieved RPS below configured threshold')
for name, maximum in thresholds.get('max_p95_seconds', {}).items():
p95 = summary['cases'][name]['p95_seconds']
if p95 is None or p95 > maximum:
errors.append(f'{phase["name"]}/{name}: P95 missing or above configured threshold')
return {'phases': results, 'errors': errors}
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('action', choices=('validate', 'run', 'cleanup'))
parser.add_argument('--profile', required=True, type=Path)
parser.add_argument('--output', type=Path, help='New run directory; existing run directory for cleanup')
parser.add_argument('--duration', type=float, help='Override duration_seconds in every phase')
parser.add_argument('--rps', type=float, help='Override RPS in every phase')
parser.add_argument('--workers', type=int, help='Override workers in every phase')
args = parser.parse_args()
try:
profile = json.loads(args.profile.read_text())
for phase in profile.get('phases', []):
for flag, field in [('duration', 'duration_seconds'), ('rps', 'rps'), ('workers', 'workers')]:
if getattr(args, flag) is not None:
phase[field] = getattr(args, flag)
validate(profile)
if args.action == 'validate':
print('Profile valid (no services or resources changed).')
return
if args.output is None:
raise ValueError('--output is required for run/cleanup')
suite = Suite(profile, args.profile.resolve(), args.output)
if args.action == 'cleanup':
errors = suite.cleanup()
if errors:
raise RuntimeError('\n'.join(errors))
return
args.output.mkdir(parents=True, exist_ok=False, mode=0o700)
suite.write('profile.json', profile)
result = {'phases': [], 'errors': []}
completed = False
try:
suite.prepare()
result = suite.run()
completed = True
except KeyboardInterrupt:
result['errors'].append('Interrupted; in-flight outcomes may be unknown. Inspect retained resources.')
except (RuntimeError, ValueError, OSError, KeyError) as exc:
result['errors'].append(str(exc))
finally:
if not completed and not result['errors']:
result['errors'].append('Run aborted before completion; inspect logs and retained resources.')
policy = profile.get('cleanup', 'on-success')
if policy == 'always' or (policy == 'on-success' and not result['errors']):
if (args.output / 'owned.json').exists():
try:
result['errors'].extend(suite.cleanup())
except (RuntimeError, ValueError, OSError, KeyError) as exc:
result['errors'].append(f'cleanup failed: {exc}')
result['passed'] = not result['errors']
suite.write('summary.json', result)
print(json.dumps(result, ensure_ascii=False, indent=2))
if not result['passed']:
raise SystemExit(1)
except (RuntimeError, ValueError, OSError, KeyError) as exc:
parser.exit(1, f'{type(exc).__name__}: {exc}\n')
if __name__ == '__main__':
main()
tools/regression_traffic.py
+220
View File
@@ -0,0 +1,220 @@
#!/usr/bin/env python3
"""Drive normal lbctl traffic and retain business checks, health and Linux resources."""
import argparse
from collections import Counter
from concurrent.futures import ThreadPoolExecutor
import json
import math
import os
from pathlib import Path
import subprocess
import sys
import threading
import time
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts'))
from cloud_runtime_soak import read_endpoint_snapshot, read_process_snapshot
def percentile(values, fraction):
return sorted(values)[max(0, math.ceil(len(values) * fraction) - 1)] if values else None
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--lbctl', required=True)
parser.add_argument('--config', required=True)
parser.add_argument('--cases', type=Path, required=True, help='JSON array of {name, args, expect}')
parser.add_argument('--output', type=Path, required=True)
parser.add_argument('--duration', type=float, default=600)
parser.add_argument('--rps', type=float, default=2)
parser.add_argument('--workers', type=int, default=4)
parser.add_argument('--cooldown', type=float, default=30)
parser.add_argument('--pid', action='append', default=[], metavar='NAME=PID')
parser.add_argument('--health', action='append', default=[], metavar='NAME=URL')
parser.add_argument('--container', action='append', default=[])
parser.add_argument(
'--shared-case-sequence', action='store_true', help='Share the weighted case sequence across workers'
)
args = parser.parse_args()
if (
not all(math.isfinite(value) for value in (args.duration, args.rps, args.cooldown))
or args.duration <= 0
or args.rps <= 0
or not 1 <= args.workers <= 32
or args.cooldown < 0
):
parser.error('duration/rps must be positive, workers 1..32, cooldown nonnegative')
cases = json.loads(args.cases.read_text())
if (
not isinstance(cases, list)
or not cases
or any(
not isinstance(case, dict)
or not isinstance(case.get('name'), str)
or not isinstance(case.get('args'), list)
or not all(isinstance(item, str) for item in case['args'])
or not isinstance(case.get('expect'), str)
or not case['expect']
for case in cases
)
):
parser.error('cases must contain name, string args, and expected response text')
args.output.mkdir(parents=True, exist_ok=False)
pids = {name: int(pid) for name, pid in (value.split('=', 1) for value in args.pid)}
endpoints = dict(value.split('=', 1) for value in args.health)
rows = []
sampling_errors = []
lock = threading.Lock()
stop = threading.Event()
phase = ['load']
next_case = 0
started = time.monotonic()
deadline = started + args.duration
redactions = [value for name, value in os.environ.items() if 'API_KEY' in name and value]
def redact(text):
for secret in redactions:
text = text.replace(secret, '[REDACTED]')
return text
def sample():
with (args.output / 'resources.jsonl').open('w') as stream:
while not stop.is_set():
row = {
'time': time.time(),
'elapsed': time.monotonic() - started,
'phase': phase[0],
'targets': {},
'health': {},
}
for name, pid in pids.items():
try:
row['targets'][name] = read_process_snapshot(pid)
except (OSError, RuntimeError) as exc:
row['targets'][name] = {'error': str(exc)}
for name, url in endpoints.items():
try:
row['health'][name] = read_endpoint_snapshot(url, timeout_seconds=3)
except RuntimeError as exc:
row['health'][name] = {'error': str(exc)}
if args.container:
result = subprocess.run(
['docker', 'stats', '--no-stream', '--format', '{{json .}}', *args.container],
capture_output=True,
text=True,
timeout=10,
)
row['containers'] = [json.loads(line) for line in result.stdout.splitlines() if line]
if result.returncode:
row['container_error'] = result.stderr.strip()
stream.write(json.dumps(row) + '\n')
stream.flush()
with lock:
progress = {
'elapsed': row['elapsed'],
'phase': phase[0],
'requests': len(rows),
'status': dict(Counter(item['status'] for item in rows)),
}
(args.output / 'progress.json').write_text(json.dumps(progress))
stop.wait(3)
def worker(index):
nonlocal next_case
iteration = 0
due = started + index / args.rps
while due < deadline:
if stop.wait(max(0, due - time.monotonic())) or time.monotonic() >= deadline:
break
if args.shared_case_sequence:
with lock:
case = cases[next_case % len(cases)]
next_case += 1
else:
case = cases[(iteration * args.workers + index) % len(cases)]
before = time.monotonic()
row = {'case': case['name'], 'time': time.time(), 'elapsed': before - started, 'status': 'failed'}
try:
result = subprocess.run(
[args.lbctl, '--config', args.config, '-o', 'json', *case['args']],
capture_output=True,
text=True,
timeout=90,
)
row['exit_code'] = result.returncode
row['output'] = json.loads(redact(result.stdout))
if result.returncode == 0 and case['expect'] in json.dumps(row['output'], ensure_ascii=False):
row['status'] = 'ok'
elif 'result_unknown' in result.stdout:
row['status'] = 'unknown'
else:
row['error'] = redact(result.stderr)[:2000] or 'exit code or response oracle failed'
except subprocess.TimeoutExpired:
row.update(status='unknown', error='CLI timed out; server outcome unknown; request was not retried')
except (OSError, ValueError) as exc:
row['error'] = str(exc)
row['seconds'] = time.monotonic() - before
with lock:
rows.append(row)
with (args.output / 'requests.jsonl').open('a') as stream:
stream.write(json.dumps(row, ensure_ascii=False) + '\n')
iteration += 1
due = max(due + args.workers / args.rps, time.monotonic())
def sampling_guard():
try:
sample()
except Exception as exc:
sampling_errors.append(f'{type(exc).__name__}: {exc}')
stop.set()
sampler = threading.Thread(target=sampling_guard)
sampler.start()
try:
with ThreadPoolExecutor(max_workers=args.workers) as pool:
list(pool.map(worker, range(args.workers)))
stop.wait(max(0, deadline - time.monotonic()))
load_elapsed = time.monotonic() - started
phase[0] = 'cooldown'
stop.wait(args.cooldown)
finally:
stop.set()
sampler.join()
summary = {
'duration_requested_seconds': args.duration,
'load_elapsed_seconds': load_elapsed,
'target_rps': args.rps,
'workers': args.workers,
'cooldown_seconds': args.cooldown,
'requests': len(rows),
'status': dict(Counter(row['status'] for row in rows)),
'cases': {},
}
samples = [json.loads(line) for line in (args.output / 'resources.jsonl').read_text().splitlines()]
summary['health_samples'] = len(samples)
summary['health_failures'] = sum(
any('error' in target for target in [*sample['targets'].values(), *sample['health'].values()])
or bool(sample.get('container_error'))
for sample in samples
)
summary['sampling_errors'] = sampling_errors
for name in dict.fromkeys(case['name'] for case in cases):
selected = [row for row in rows if row['case'] == name]
times = [row['seconds'] for row in selected if row['status'] == 'ok']
summary['cases'][name] = {
'requests': len(selected),
'status': dict(Counter(row['status'] for row in selected)),
'p50_seconds': percentile(times, 0.5),
'p95_seconds': percentile(times, 0.95),
'p99_seconds': percentile(times, 0.99),
}
(args.output / 'summary.json').write_text(json.dumps(summary, indent=2))
print(json.dumps(summary, indent=2), flush=True)
if not rows or any(row['status'] != 'ok' for row in rows) or summary['health_failures'] or sampling_errors:
raise SystemExit(1)
if __name__ == '__main__':
main()
tools/test_mock_llm.py
+196
View File
@@ -0,0 +1,196 @@
"""Exercise the real HTTP server without LangBot, credentials or dependencies."""
from concurrent.futures import ThreadPoolExecutor
import base64
import json
from pathlib import Path
import tempfile
import struct
import threading
import unittest
from urllib.error import HTTPError
from urllib.request import Request, urlopen
from mock_llm import MockServer
class MockLLMTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.tmp = tempfile.TemporaryDirectory()
cls.record = Path(cls.tmp.name) / 'requests.jsonl'
scenarios = json.loads(Path(__file__).with_name('mock_llm_scenarios.json').read_text())
cls.server = MockServer(('127.0.0.1', 0), scenarios, cls.record)
cls.thread = threading.Thread(target=cls.server.serve_forever)
cls.thread.start()
cls.url = f'http://127.0.0.1:{cls.server.server_port}'
@classmethod
def tearDownClass(cls):
cls.server.shutdown()
cls.server.server_close()
cls.thread.join()
cls.tmp.cleanup()
def post(self, model='mock-chat', **kwargs):
body = {
'model': model,
'messages': [{'role': 'user', 'content': 'test'}],
**kwargs,
}
request = Request(
self.url + '/v1/chat/completions',
data=json.dumps(body).encode(),
headers={
'Content-Type': 'application/json',
'Authorization': 'Bearer must-not-be-recorded',
},
)
with urlopen(request, timeout=5) as response:
return response.read().decode()
def test_chat_and_model_discovery(self):
with urlopen(self.url + '/v1/models') as response:
models = json.load(response)
self.assertIn('mock-box', [m['id'] for m in models['data']])
self.assertEqual(
json.loads(self.post())['choices'][0]['message']['content'],
'LANGBOT_CHAT_OK',
)
def test_stream_and_usage(self):
events = self.post(stream=True, stream_options={'include_usage': True}).splitlines()
self.assertEqual(events[-2], 'data: [DONE]')
chunks = [json.loads(line[6:]) for line in events if line.startswith('data: {')]
content = ''.join(c['choices'][0]['delta'].get('content', '') for c in chunks if c['choices'])
self.assertEqual(content, 'LANGBOT_CHAT_OK')
self.assertEqual(chunks[-1]['usage']['total_tokens'], 15)
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'])
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']
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.assertEqual(caught.exception.code, 400)
def box_turn(self, streaming=False):
messages = [{'role': 'user', 'content': 'run box checks'}]
tools = [
{
'type': 'function',
'function': {'name': 'exec', 'parameters': {'type': 'object'}},
}
]
commands = []
for _ in range(2):
wire = self.post('mock-box', messages=messages, tools=tools, stream=streaming)
if streaming:
deltas = [
json.loads(line[6:])['choices'][0]['delta']
for line in wire.splitlines()
if line.startswith('data: {')
]
call = next(d['tool_calls'][0] for d in deltas if d.get('tool_calls') and 'id' in d['tool_calls'][0])
arguments = ''.join(d['tool_calls'][0]['function']['arguments'] for d in deltas if d.get('tool_calls'))
call = {
**call,
'function': {**call['function'], 'arguments': arguments},
}
call.pop('index')
message = {'role': 'assistant', 'content': None, 'tool_calls': [call]}
else:
message = json.loads(wire)['choices'][0]['message']
call = message['tool_calls'][0]
commands.append(json.loads(call['function']['arguments'])['command'])
messages.extend(
[
message,
{
'role': 'tool',
'tool_call_id': call['id'],
'content': 'LANGBOT_BOX_SENTINEL',
},
]
)
final = json.loads(self.post('mock-box', messages=messages, tools=tools))
self.assertEqual(final['choices'][0]['message']['content'], 'LANGBOT_BOX_OK')
self.assertIn('printf LANGBOT_BOX_SENTINEL', commands[0])
self.assertEqual(commands[1], 'cat /workspace/mock-llm.txt')
def test_tool_rounds_and_fragmented_stream(self):
self.box_turn()
self.box_turn(streaming=True)
def test_concurrent_turns_do_not_share_progress(self):
with ThreadPoolExecutor(max_workers=4) as pool:
list(pool.map(lambda _: self.box_turn(), range(8)))
def test_rag_requires_expected_model_input(self):
with self.assertRaises(HTTPError) as caught:
self.post('mock-rag')
self.assertEqual(caught.exception.code, 400)
response = json.loads(
self.post(
'mock-rag',
messages=[
{'role': 'system', 'content': 'Retrieved document: aurora-parser-rag-9137'},
{'role': 'user', 'content': 'What is the documented sentinel?'},
],
)
)
self.assertEqual(response['choices'][0]['message']['content'], 'LANGBOT_RAG_OK')
def test_errors_and_invalid_requests(self):
for status in (401, 429, 503):
with self.subTest(status=status), self.assertRaises(HTTPError) as caught:
self.post(f'mock-{status}')
self.assertEqual(caught.exception.code, status)
for kwargs in (
{'model': 'unknown'},
{'messages': []},
{'stream_options': []},
{'model': 'mock-box', 'tools': []},
):
with self.subTest(kwargs=kwargs), self.assertRaises(HTTPError) as caught:
self.post(**kwargs)
self.assertEqual(caught.exception.code, 400)
messages = [
{'role': 'user', 'content': 'test'},
{'role': 'assistant', 'tool_calls': [{'id': 'call1'}]},
]
with self.assertRaises(HTTPError) as caught:
self.post('mock-box', messages=messages)
self.assertEqual(caught.exception.code, 400)
messages.append({'role': 'tool', 'tool_call_id': 'call1', 'content': 'wrong output'})
with self.assertRaises(HTTPError) as caught:
self.post('mock-box', messages=messages)
self.assertEqual(caught.exception.code, 400)
def test_record_omits_authorization(self):
self.post()
with self.server.record_lock:
recorded = self.record.read_text()
self.assertNotIn('must-not-be-recorded', recorded)
self.assertIn('"request_id"', recorded)
if __name__ == '__main__':
unittest.main()
tools/test_regression_job.py
+179
View File
@@ -0,0 +1,179 @@
"""Exercise Job ownership and lifecycle with controlled CLI/Docker responses."""
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import os
from pathlib import Path
import shutil
import sys
import tempfile
import threading
import unittest
from unittest.mock import patch
import yaml
from regression_job import Job, main
from test_regression_suite import FAKE_CLI, profile
class ControlledJob(Job):
def __init__(self, manifest, suite, path, output, health, fail_install=False):
super().__init__(manifest, suite, path, output)
self.health = health
self.fail_install = fail_install
self.calls = []
self.env['LBCTL_TEST_ROOT'] = str(output.parent)
def command(self, args, timeout=900):
self.calls.append(args)
if args[: len(self.cli)] == self.cli:
operation = args[len(self.cli) :]
if operation[0] == 'install':
self.deploy.mkdir()
(self.deploy / 'data').mkdir()
(self.deploy / 'compose.yaml').write_text('services: {}')
(self.deploy / '.env').write_text(
'LANGBOT_PLUGIN_RUNTIME_CONTROL_TOKEN=' + 't' * 32 + '\nLANGBOT_BOX_ENABLED=true\n'
)
record = {
'deployment_id': 'owned-deployment',
'compose_project': 'test-project',
'compose_file': str(self.deploy / 'compose.yaml'),
'data_dir': str(self.deploy / 'data'),
'env_file': str(self.deploy / '.env'),
'endpoint': self.health,
}
(self.deploy / 'deployment.yaml').write_text(yaml.safe_dump(record))
# Simulate a timeout/failed final probe with no returned record.
return (
(1, json.dumps({'data': {}}), 'install failed')
if self.fail_install
else (0, json.dumps({'data': {'record': record}}), '')
)
if operation[0] == '--endpoint':
return 0, json.dumps({'data': {'workspace_uuid': 'test-workspace'}}), ''
if operation[0] == 'uninstall':
shutil.rmtree(self.deploy / 'data')
return 0, json.dumps({'data': {}}), ''
if args[:2] == ['docker', 'ps']:
if 'label=langbot.box=true' in args:
return 0, '', ''
return (
0,
'\n'.join(
f'id-{name} {name}'
for name in ('langbot', 'langbot_plugin_runtime', 'langbot_box', 'regression_mock')
),
'',
)
if args[:2] == ['docker', 'inspect']:
if '{{.State.Pid}}' in args:
return 0, str(os.getpid()), ''
return 0, json.dumps({'network': {'IPAddress': '127.0.0.1'}}), ''
if args[:2] == ['docker', 'compose']:
if 'logs' in args:
return 0, self.redact(self.env.get('LANGBOT_QA_API_KEY', '') + '\n' + 't' * 32), ''
return 0, '', ''
raise AssertionError(args)
@staticmethod
def ready(url):
pass
def prepare(self):
suite = super().prepare()
# Real traffic driver and HTTP health probe; infrastructure calls are controlled above.
suite.profile['observe'] = {'health': {'core': self.health}}
return suite
class JobTests(unittest.TestCase):
def test_lifecycle_success_and_partial_install_recovery(self):
class Health(BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(200)
self.end_headers()
self.wfile.write(b'{"code":0}')
def log_message(self, *args):
pass
server = ThreadingHTTPServer(('127.0.0.1', 0), Health)
thread = threading.Thread(target=server.serve_forever)
thread.start()
try:
with tempfile.TemporaryDirectory() as folder:
root = Path(folder)
cli = root / 'lbctl-test'
cli.write_text(f'#!{sys.executable}\n' + FAKE_CLI)
cli.chmod(0o700)
manifest = {'lbctl': str(cli), 'deployment': {'version': 'v4.11.0', 'port': 15351}}
suite = profile()
suite['resources'] = [{'id': 'chat', 'kind': 'pipeline', 'body': {'name': 'QA', 'config': {'ai': {}}}}]
suite['cases'][0].update(args=['pipeline', 'run', '${chat}'], expect='TEST_OK')
suite['phases'][0].update(duration_seconds=0.5, rps=4, workers=2)
health = f'http://127.0.0.1:{server.server_port}'
job = ControlledJob(manifest, suite, root / 'profile.json', root / 'success', health)
result = job.run()
self.assertTrue(result['passed'], result)
self.assertEqual(result['phase'], 'cleaned')
self.assertFalse((job.deploy / 'data').exists())
self.assertFalse((job.private / 'api-key').exists())
self.assertIn('--purge-data', job.calls[-1])
self.assertNotIn('t' * 32, (job.output / 'services.log').read_text())
job = ControlledJob(manifest, suite, root / 'profile.json', root / 'failed', health, fail_install=True)
result = job.run()
self.assertFalse(result['passed'])
self.assertEqual(job.state['record']['deployment_id'], 'owned-deployment')
self.assertTrue((job.deploy / 'data').exists())
self.assertFalse(any('uninstall' in call for call in job.calls))
with (
patch.object(sys, 'argv', ['regression_job.py', 'cleanup', '--output', str(job.output)]),
patch('regression_job.Job', return_value=job),
):
main()
self.assertFalse((job.deploy / 'data').exists())
self.assertEqual(job.state['phase'], 'cleaned')
finally:
server.shutdown()
server.server_close()
thread.join()
def test_cleanup_refuses_changed_deployment(self):
with tempfile.TemporaryDirectory() as folder:
root = Path(folder)
job = Job({}, profile(), root / 'profile.json', root)
job.private.mkdir()
job.deploy.mkdir()
(job.private / 'job-id').write_text(job.state['job_id'])
record = {
'deployment_id': 'old',
'compose_file': str(job.deploy / 'compose.yaml'),
'data_dir': str(job.deploy / 'data'),
'env_file': str(job.deploy / '.env'),
}
job.state['record'] = record
(job.deploy / 'deployment.yaml').write_text(yaml.safe_dump({**record, 'deployment_id': 'new'}))
with self.assertRaisesRegex(RuntimeError, 'identity changed'):
job.cleanup()
def test_sandbox_cleanup_uses_owned_mount_roots(self):
with tempfile.TemporaryDirectory() as folder:
root = Path(folder)
job = Job({}, profile(), root / 'profile.json', root)
job.state['record'] = {'data_dir': str(job.deploy / 'data')}
def checked(args, timeout=900):
if args[1] == 'ps':
return 'ours\nother\n'
source = str(job.deploy / 'data/box/tenant') if args[-1] == 'ours' else '/unrelated/box/tenant'
return json.dumps([{'Source': source}])
job.checked = checked
self.assertEqual(job.owned_sandboxes(), ['ours'])
if __name__ == '__main__':
unittest.main()
tools/test_regression_suite.py
+228
View File
@@ -0,0 +1,228 @@
"""Focused checks for profile validation, ownership and the CLI preparation sequence."""
import copy
from contextlib import redirect_stdout
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import io
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import threading
import unittest
from unittest.mock import patch
from regression_suite import Suite, main, merge, resolve, validate
FAKE_CLI = r"""
import json, os, sys
from pathlib import Path
args = sys.argv[5:]
root = Path(os.environ['LBCTL_TEST_ROOT'])
state_path = root / 'cli-state.json'
state = json.loads(state_path.read_text()) if state_path.exists() else {}
body = json.loads(Path(args[args.index('--file') + 1]).read_text()) if '--file' in args else {}
with (root / 'cli-calls.jsonl').open('a') as log:
log.write(json.dumps({'args': args, 'body': body}) + '\n')
if args == ['whoami']:
data = {'workspace_uuid': 'test-workspace'}
elif args[:2] == ['pipeline', 'run']:
assert args[2] in state
data = {'status': 'completed', 'reply': 'TEST_OK'}
elif args[1] == 'delete':
del state[args[2]]
state_path.write_text(json.dumps(state))
data = {}
elif args[:2] == ['pipeline', 'get']:
data = state[args[2]]
elif args[1] in ['create', 'apply']:
uuid = body.get('uuid', f'created-{len(state)}')
data = {'uuid': uuid, 'name': body.get('name', 'test'), 'description': '', 'config': body.get('config', {'ai': {'default': True}})}
state[uuid] = data
state_path.write_text(json.dumps(state))
else:
raise RuntimeError(args)
print(json.dumps({'ok': True, 'data': data}))
"""
def profile():
return {
'version': 1,
'cli': {'config': 'lbctl.yaml'},
'cases': [{'name': 'chat', 'args': ['pipeline', 'run', 'existing'], 'expect': 'OK'}],
'phases': [{'name': 'normal', 'duration_seconds': 1, 'rps': 2, 'workers': 4}],
'observe': {'health': {'core': 'http://127.0.0.1:5300/healthz'}},
}
class ProfileTests(unittest.TestCase):
def test_normal_profile_and_overrides_contract(self):
validate(json.loads(Path(__file__).with_name('regression_profiles').joinpath('normal.json').read_text()))
bad = profile()
for field, value in [('rps', float('nan')), ('workers', 33), ('duration_seconds', -1)]:
with self.subTest(field=field):
current = copy.deepcopy(bad)
current['phases'][0][field] = value
with self.assertRaises(ValueError):
validate(current)
def test_missing_refs_and_existing_resource_overwrite_are_rejected(self):
current = profile()
current['cases'][0]['args'][-1] = '${missing}'
with self.assertRaises(KeyError):
validate(current)
current = profile()
current['resources'] = [{'id': 'pipeline', 'kind': 'pipeline', 'body': {'uuid': 'existing'}}]
with self.assertRaisesRegex(ValueError, 'uuid is forbidden'):
validate(current)
def test_merge_keeps_defaults_but_replaces_lists(self):
base = {'ai': {'enabled': True, 'models': ['old']}, 'output': {'x': 1}}
actual = merge(base, {'ai': {'models': ['new']}})
self.assertEqual(actual, {'ai': {'enabled': True, 'models': ['new']}, 'output': {'x': 1}})
self.assertEqual(base['ai']['models'], ['old'])
self.assertEqual(resolve(['${id}', {'id': '${id}'}], {'id': 'created'}), ['created', {'id': 'created'}])
def test_ownership_recorded_before_configuration_failure(self):
with tempfile.TemporaryDirectory() as folder:
root = Path(folder)
current = profile()
current['resources'] = [{'id': 'chat', 'kind': 'pipeline', 'body': {'name': 'QA', 'config': {'ai': {}}}}]
suite = Suite(current, root / 'profile.json', root)
calls = []
def call(args, body=None):
calls.append((args, body))
if args == ['whoami']:
return {'workspace_uuid': 'workspace-a'}
if args == ['pipeline', 'apply'] and 'uuid' not in body:
return {'uuid': 'owned-pipeline'}
if args[:2] == ['pipeline', 'get']:
return {'uuid': 'owned-pipeline', 'name': 'QA', 'description': '', 'config': {'ai': {'old': 1}}}
if args == ['pipeline', 'apply']:
raise RuntimeError('configuration failed')
return {}
suite.call = call
with self.assertRaisesRegex(RuntimeError, 'configuration failed'):
suite.prepare()
self.assertEqual(
json.loads((root / 'owned.json').read_text())['resources'],
[{'kind': 'pipeline', 'uuid': 'owned-pipeline'}],
)
self.assertEqual(suite.cleanup(), [])
self.assertEqual(calls[-1][0], ['pipeline', 'delete', 'owned-pipeline', '--yes'])
self.assertEqual(json.loads((root / 'owned.json').read_text())['resources'], [])
def test_cleanup_refuses_another_workspace(self):
with tempfile.TemporaryDirectory() as folder:
root = Path(folder)
(root / 'owned.json').write_text(json.dumps({'workspace_uuid': 'other', 'resources': []}))
suite = Suite(profile(), root / 'profile.json', root)
suite.call = lambda args: {'workspace_uuid': 'current'}
with self.assertRaisesRegex(RuntimeError, 'Workspace differs'):
suite.cleanup()
def test_aborted_run_is_not_treated_as_successful_cleanup(self):
with tempfile.TemporaryDirectory() as folder:
root = Path(folder)
config = root / 'profile.json'
config.write_text(json.dumps(profile()))
for error_type in (KeyboardInterrupt, TypeError):
output = root / error_type.__name__
def interrupt(suite):
suite.variables['workspace_uuid'] = 'test-workspace'
suite.journal()
raise error_type
argv = ['regression_suite.py', 'run', '--profile', str(config), '--output', str(output)]
with (
patch.object(sys, 'argv', argv),
patch.object(Suite, 'prepare', interrupt),
patch.object(Suite, 'cleanup') as cleanup,
redirect_stdout(io.StringIO()),
):
with self.assertRaises(SystemExit if error_type is KeyboardInterrupt else TypeError):
main()
cleanup.assert_not_called()
self.assertFalse(json.loads((output / 'summary.json').read_text())['passed'])
def test_subprocess_phases_and_cleanup_with_controlled_cli(self):
class Health(BaseHTTPRequestHandler):
def do_GET(self):
payload = b'{"code":0}'
self.send_response(200)
self.send_header('Content-Length', str(len(payload)))
self.end_headers()
self.wfile.write(payload)
def log_message(self, *args):
pass
server = ThreadingHTTPServer(('127.0.0.1', 0), Health)
thread = threading.Thread(target=server.serve_forever)
thread.start()
try:
with tempfile.TemporaryDirectory() as folder:
root = Path(folder)
cli = root / 'lbctl-test'
cli.write_text(f'#!{sys.executable}\n' + FAKE_CLI)
cli.chmod(0o700)
current = profile()
current['cli']['executable'] = str(cli)
current['resources'] = [
{'id': 'provider', 'kind': 'provider', 'body': {'name': 'test'}},
{
'id': 'pipeline',
'kind': 'pipeline',
'body': {'name': 'test', 'config': {'ai': {'model': '${provider}'}}},
},
]
current['cases'] = [
{'name': 'chat', 'args': ['pipeline', 'run', '${pipeline}'], 'expect': 'TEST_OK', 'weight': 2},
{'name': 'box', 'args': ['pipeline', 'run', '${pipeline}'], 'expect': 'TEST_OK', 'weight': 1},
]
current['phases'] = [
{'name': name, 'duration_seconds': 0.5, 'rps': 12, 'workers': 3} for name in ('low', 'high')
]
current['observe'] = {'health': {'core': f'http://127.0.0.1:{server.server_port}/healthz'}}
current['cleanup'] = 'always'
config = root / 'profile.json'
command = [
sys.executable,
str(Path(__file__).with_name('regression_suite.py')),
'run',
'--profile',
str(config),
]
environment = {**os.environ, 'LBCTL_TEST_ROOT': folder}
for enforce_latency in (False, True):
current['thresholds'] = {'max_p95_seconds': {'chat': 0.001}} if enforce_latency else {}
config.write_text(json.dumps(current))
output = root / ('failure' if enforce_latency else 'success')
result = subprocess.run(
[*command, '--output', str(output)], env=environment, capture_output=True, text=True, timeout=30
)
summary = json.loads((output / 'summary.json').read_text())
self.assertEqual(result.returncode, int(enforce_latency), result.stdout + result.stderr)
self.assertEqual(summary['passed'], not enforce_latency)
self.assertEqual(len(summary['phases']), 2)
for phase in summary['phases']:
self.assertGreater(phase['achieved_rps'], 0)
self.assertEqual(phase['health_failures'], 0)
self.assertGreaterEqual(phase['cases']['chat']['requests'], phase['cases']['box']['requests'])
self.assertEqual(json.loads((output / 'owned.json').read_text())['resources'], [])
self.assertEqual(json.loads((root / 'cli-state.json').read_text()), {})
finally:
server.shutdown()
server.server_close()
thread.join()
if __name__ == '__main__':
unittest.main()