mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-10-30 20:35:22 +08:00
huanghuoguoguo
f837083781
* test(regression): expand finite scenarios and share bounded execution lifecycle * fix(regression): continue job cleanup after state persistence failure
329 lines
15 KiB
Python
329 lines
15 KiB
Python
"""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
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import yaml
|
|
|
|
from regression_job import Job, load_profile, main
|
|
from regression_scenarios import ExecutionWindow
|
|
from regression_suite import Suite
|
|
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_published_resolution_failure_is_preparation_failure_without_deployment(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
root = Path(folder)
|
|
job = ControlledJob({}, profile(), root / 'profile.json', root / 'run', 'unused')
|
|
with patch('regression_job.prepare_published_plugins', side_effect=RuntimeError('release unavailable')):
|
|
result = job.run()
|
|
self.assertFalse(result['passed'])
|
|
self.assertFalse(result['completed'])
|
|
self.assertEqual(result['phase'], 'plugins')
|
|
self.assertEqual(result['errors'], ['release unavailable'])
|
|
self.assertEqual(job.calls, [])
|
|
|
|
def test_job_defaults_to_published_plugins_and_keeps_explicit_profile_mode(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
root = Path(folder)
|
|
suite = profile()
|
|
suite['plugins'] = [{'source': 'local', 'file': 'developer.zip'}]
|
|
(root / 'suite.json').write_text(json.dumps(suite))
|
|
path = root / 'job.json'
|
|
manifest = {'version': 1, 'suite': 'suite.json', 'deployment': {'version': 'v4.11.0'}}
|
|
path.write_text(json.dumps(manifest))
|
|
self.assertEqual(load_profile(path)[1]['plugins'], [])
|
|
manifest['plugin_mode'] = 'profile'
|
|
path.write_text(json.dumps(manifest))
|
|
self.assertEqual(load_profile(path)[1]['plugins'], suite['plugins'])
|
|
manifest['plugin_manifest'] = 'previous/manifest.json'
|
|
path.write_text(json.dumps(manifest))
|
|
with self.assertRaisesRegex(ValueError, 'requires published'):
|
|
load_profile(path)
|
|
|
|
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},
|
|
'plugin_mode': 'profile',
|
|
}
|
|
suite = profile()
|
|
suite['coverage_suite'] = 'smoke'
|
|
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.assertTrue((job.output / 'suite/coverage.json').exists())
|
|
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_shared_finish_cleans_owned_resources_before_environment_teardown(self):
|
|
for failure in ('prepare', 'finish-error', 'finish-raises'):
|
|
with self.subTest(failure=failure), tempfile.TemporaryDirectory() as folder:
|
|
root = Path(folder)
|
|
current = profile()
|
|
current.pop('phases')
|
|
current.update(
|
|
smoke={'ledger_db': 'unused.db'},
|
|
cleanup='always',
|
|
execution={'total_seconds': 1, 'drain_seconds': 0.1, 'cooldown_seconds': 0.01},
|
|
)
|
|
job = Job({'cleanup': 'always'}, current, root / 'profile.json', root / 'job')
|
|
events, visible = [], ['owned']
|
|
|
|
def prepare():
|
|
destination = job.output / 'suite'
|
|
destination.mkdir()
|
|
suite = job.suite = Suite(current, root / 'profile.json', destination)
|
|
suite.variables['workspace_uuid'] = 'workspace'
|
|
suite.owned = [{'kind': 'provider', 'uuid': 'owned'}]
|
|
suite.journal()
|
|
|
|
def call(args, body=None):
|
|
if args == ['whoami']:
|
|
return {'workspace_uuid': 'workspace'}
|
|
if args == ['provider', 'delete', 'owned', '--yes']:
|
|
events.append('delete-owned')
|
|
visible.remove('owned')
|
|
return {}
|
|
if args == ['provider', 'list']:
|
|
return {'providers': [{'uuid': item} for item in visible]}
|
|
raise AssertionError(args)
|
|
|
|
suite.call = call
|
|
suite.run = lambda: {'errors': [], 'phases': []}
|
|
suite.execution = ExecutionWindow(suite)
|
|
suite.execution.start()
|
|
if failure == 'prepare':
|
|
raise RuntimeError('prepare failed after creating resource')
|
|
finish = suite.finish
|
|
|
|
def failing_finish(summary):
|
|
if failure == 'finish-error':
|
|
suite.execution.errors.append('health failed during execution')
|
|
finish(summary)
|
|
if failure == 'finish-raises':
|
|
raise TypeError('finalization defect')
|
|
|
|
suite.finish = failing_finish
|
|
return suite
|
|
|
|
with (
|
|
patch.object(job, 'prepare', side_effect=prepare),
|
|
patch.object(job, 'collect', side_effect=lambda: events.append('collect')),
|
|
patch.object(job, 'cleanup', side_effect=lambda: events.append('teardown')),
|
|
patch('regression_traffic.read_endpoint_snapshot', return_value={'http.ok': 1}),
|
|
):
|
|
result = job.run()
|
|
self.assertFalse(result['passed'])
|
|
self.assertFalse(result['suite']['passed'])
|
|
self.assertEqual(result['suite']['outcome'], 'FAIL')
|
|
self.assertFalse(job.suite.execution.thread.is_alive())
|
|
self.assertEqual(events, ['delete-owned', 'collect', 'teardown'])
|
|
self.assertEqual(json.loads((job.output / 'suite/owned.json').read_text())['resources'], [])
|
|
self.assertIn('cleanup', [phase['stage'] for phase in result['suite']['execution']['phases']])
|
|
self.assertEqual(result['errors'], result['suite']['errors'])
|
|
self.assertEqual(result['completed'], failure != 'prepare')
|
|
|
|
def test_finished_state_save_failure_does_not_skip_collection_or_cleanup(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
root = Path(folder)
|
|
job = Job({'cleanup': 'always'}, profile(), root / 'profile.json', root / 'job')
|
|
suite = SimpleNamespace(run=lambda: {'errors': []}, finish=lambda summary: None)
|
|
events = []
|
|
save = job.save
|
|
|
|
def failing_save():
|
|
if job.state['phase'] == 'finished':
|
|
raise OSError('state disk unavailable')
|
|
save()
|
|
|
|
with (
|
|
patch.object(job, 'prepare', return_value=suite),
|
|
patch.object(job, 'save', side_effect=failing_save),
|
|
patch.object(job, 'collect', side_effect=lambda: events.append('collect')),
|
|
patch.object(job, 'cleanup', side_effect=lambda: events.append('cleanup')),
|
|
):
|
|
result = job.run()
|
|
self.assertEqual(events, ['collect', 'cleanup'])
|
|
self.assertTrue(result['completed'])
|
|
self.assertFalse(result['passed'])
|
|
self.assertEqual(result['errors'], ['state persistence: state disk unavailable'])
|
|
self.assertEqual(json.loads((job.output / 'summary.json').read_text()), result)
|
|
|
|
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')}
|
|
containers = ['ours', 'other']
|
|
|
|
def checked(args, timeout=900):
|
|
if args[1] == 'ps':
|
|
return '\n'.join(containers)
|
|
return '\n'.join(
|
|
json.dumps(
|
|
{
|
|
'Id': container,
|
|
'Mounts': [
|
|
{
|
|
'Source': str(job.deploy / 'data/box/tenant')
|
|
if container != 'other'
|
|
else '/unrelated/box/tenant'
|
|
}
|
|
],
|
|
}
|
|
)
|
|
for container in args[4:]
|
|
)
|
|
|
|
job.checked = checked
|
|
self.assertEqual(job.owned_sandboxes(), ['ours'])
|
|
containers.append('later')
|
|
self.assertEqual(job.owned_sandboxes(), ['ours', 'later'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|