mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-11-03 22:32:27 +08:00
test(tools): install latest published regression plugins
This commit is contained in:
huanghuoguoguo
@@ -0,0 +1,192 @@
|
||||
"""Exercise published resolution, replay integrity and installation readback."""
|
||||
|
||||
from collections import Counter
|
||||
import copy
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
import io
|
||||
import json
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
import zipfile
|
||||
|
||||
import regression_plugins as plugins
|
||||
|
||||
|
||||
def archive(plugin_id, version='1.2.3'):
|
||||
author, name = plugin_id.split('/')
|
||||
output = io.BytesIO()
|
||||
with zipfile.ZipFile(output, 'w') as package:
|
||||
package.writestr(
|
||||
zipfile.ZipInfo('manifest.yaml'),
|
||||
f'kind: Plugin\nmetadata:\n author: {author}\n name: {name}\n version: {version}\n',
|
||||
)
|
||||
package.writestr(zipfile.ZipInfo('main.py'), '# Published fixture\n')
|
||||
return output.getvalue()
|
||||
|
||||
|
||||
class PublishedPluginsTests(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
class Marketplace(BaseHTTPRequestHandler):
|
||||
def do_GET(self):
|
||||
cls.requests.append(self.path)
|
||||
status, body = cls.responses.get(self.path, (404, b'missing published release'))
|
||||
self.send_response(status)
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def log_message(self, *args):
|
||||
pass
|
||||
|
||||
cls.server = ThreadingHTTPServer(('127.0.0.1', 0), Marketplace)
|
||||
cls.thread = threading.Thread(target=cls.server.serve_forever)
|
||||
cls.thread.start()
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
cls.server.shutdown()
|
||||
cls.server.server_close()
|
||||
cls.thread.join()
|
||||
|
||||
def setUp(self):
|
||||
self.folder = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.folder.cleanup)
|
||||
self.root = Path(self.folder.name)
|
||||
self.marketplace = patch.object(plugins, 'MARKETPLACE', f'http://127.0.0.1:{self.server.server_port}')
|
||||
self.marketplace.start()
|
||||
self.addCleanup(self.marketplace.stop)
|
||||
type(self).requests = []
|
||||
type(self).responses = {}
|
||||
for plugin_id in plugins.PLUGIN_IDS:
|
||||
author, name = plugin_id.split('/')
|
||||
data = {
|
||||
'code': 0,
|
||||
'data': {
|
||||
'plugin': {
|
||||
'plugin_id': plugin_id,
|
||||
'author': author,
|
||||
'name': name,
|
||||
'latest_version': '1.2.3',
|
||||
'status': 'live',
|
||||
}
|
||||
},
|
||||
}
|
||||
self.responses[f'/api/v1/marketplace/plugins/{plugin_id}'] = (200, json.dumps(data).encode())
|
||||
self.responses[f'/api/v1/marketplace/plugins/download/{plugin_id}/1.2.3'] = (200, archive(plugin_id))
|
||||
|
||||
def test_latest_resolved_once_and_replay_uses_same_published_bytes(self):
|
||||
directory = self.root / 'latest'
|
||||
manifest = plugins.prepare_published_plugins(directory)
|
||||
self.assertEqual(Counter(self.requests), {path: 1 for path in self.responses})
|
||||
installed = []
|
||||
|
||||
def call(args):
|
||||
if args[:3] == ['plugin', 'install', 'local']:
|
||||
source = Path(args[4])
|
||||
self.assertEqual(source.read_bytes(), archive(plugins.PLUGIN_IDS[len(installed)]))
|
||||
author, name = plugins.PLUGIN_IDS[len(installed)].split('/')
|
||||
installed.append(
|
||||
{'author': author, 'name': name, 'version': '1.2.3', 'enabled': True, 'status': 'initialized'}
|
||||
)
|
||||
return {}
|
||||
self.assertEqual(args, ['plugin', 'list'])
|
||||
return {'plugins': installed}
|
||||
|
||||
plugins.install_published_plugins(call, directory, manifest)
|
||||
recorded = json.loads((directory / 'manifest.json').read_text())
|
||||
self.assertTrue(all(entry['installed'][0]['status'] == 'initialized' for entry in recorded['plugins']))
|
||||
self.requests.clear()
|
||||
replay = plugins.prepare_published_plugins(self.root / 'replay', directory / 'manifest.json')
|
||||
self.assertEqual(
|
||||
[entry['sha256'] for entry in replay['plugins']], [entry['sha256'] for entry in manifest['plugins']]
|
||||
)
|
||||
self.assertEqual(len(self.requests), 3)
|
||||
self.assertTrue(all('/download/' in path for path in self.requests))
|
||||
|
||||
def test_bad_metadata_and_unpublished_versions_fail_without_download(self):
|
||||
url = '/api/v1/marketplace/plugins/langbot-team/LocalAgent'
|
||||
valid = json.loads(self.responses[url][1])
|
||||
invalid = [[], {'code': 0, 'data': []}, {'code': 1, 'data': valid['data']}]
|
||||
for field, value in [
|
||||
('latest_version', ''),
|
||||
('latest_version', '../bad'),
|
||||
('status', 'draft'),
|
||||
('name', 'Other'),
|
||||
]:
|
||||
data = copy.deepcopy(valid)
|
||||
data['data']['plugin'][field] = value
|
||||
invalid.append(data)
|
||||
for index, payload in enumerate(invalid):
|
||||
with self.subTest(payload=payload):
|
||||
self.requests.clear()
|
||||
self.responses[url] = (200, json.dumps(payload).encode())
|
||||
with self.assertRaises(ValueError):
|
||||
plugins.prepare_published_plugins(self.root / str(index))
|
||||
self.assertEqual(self.requests, [url])
|
||||
|
||||
def test_missing_or_wrong_release_does_not_fall_back(self):
|
||||
url = '/api/v1/marketplace/plugins/download/langbot-team/LocalAgent/1.2.3'
|
||||
self.responses[url] = (404, b'missing')
|
||||
with self.assertRaisesRegex(RuntimeError, 'HTTP Error 404'):
|
||||
plugins.prepare_published_plugins(self.root / 'missing')
|
||||
self.responses[url] = (200, archive('langbot-team/LocalAgent', '0.0.1'))
|
||||
with self.assertRaisesRegex(ValueError, 'identity/version differs'):
|
||||
plugins.prepare_published_plugins(self.root / 'wrong')
|
||||
|
||||
def test_replay_rejects_changed_packages_and_untrusted_identity(self):
|
||||
directory = self.root / 'latest'
|
||||
manifest = plugins.prepare_published_plugins(directory)
|
||||
replay = directory / 'manifest.json'
|
||||
url = '/api/v1/marketplace/plugins/download/langbot-team/LocalAgent/1.2.3'
|
||||
self.responses[url] = (200, b'different package')
|
||||
with self.assertRaisesRegex(ValueError, 'sha256 changed'):
|
||||
plugins.prepare_published_plugins(self.root / 'changed', replay)
|
||||
for change in ('identity', 'duplicate', 'origin', 'hash'):
|
||||
invalid = copy.deepcopy(manifest)
|
||||
if change == 'identity':
|
||||
invalid['plugins'][0]['id'] = 'untrusted/Other'
|
||||
elif change == 'duplicate':
|
||||
invalid['plugins'][1] = invalid['plugins'][0]
|
||||
elif change == 'origin':
|
||||
invalid['marketplace'] = 'https://untrusted.example'
|
||||
else:
|
||||
del invalid['plugins'][0]['sha256']
|
||||
replay.write_text(json.dumps(invalid))
|
||||
with self.subTest(change=change), self.assertRaises(ValueError):
|
||||
plugins.prepare_published_plugins(self.root / change, replay)
|
||||
|
||||
def test_readback_records_actual_failed_version_and_stops_installing(self):
|
||||
directory = self.root / 'latest'
|
||||
manifest = plugins.prepare_published_plugins(directory)
|
||||
calls = []
|
||||
|
||||
def call(args):
|
||||
calls.append(args)
|
||||
return {
|
||||
'plugins': [
|
||||
{
|
||||
'author': 'langbot-team',
|
||||
'name': 'LocalAgent',
|
||||
'version': '0.0.1',
|
||||
'enabled': True,
|
||||
'status': 'initialized',
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
with self.assertRaisesRegex(RuntimeError, 'not ready at the selected version'):
|
||||
plugins.install_published_plugins(call, directory, manifest)
|
||||
self.assertEqual(len(calls), 2)
|
||||
recorded = json.loads((directory / 'manifest.json').read_text())
|
||||
self.assertEqual(recorded['plugins'][0]['installed'][0]['version'], '0.0.1')
|
||||
(directory / manifest['plugins'][0]['archive']).write_bytes(b'changed after download')
|
||||
with self.assertRaisesRegex(ValueError, 'changed before installation'):
|
||||
plugins.install_published_plugins(call, directory, manifest)
|
||||
self.assertEqual(len(calls), 2)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user