| import json |
| import os |
| import tempfile |
| import threading |
| import unittest |
| from types import SimpleNamespace |
| from unittest.mock import patch |
|
|
| from fastapi.testclient import TestClient |
| from lg_control.cloud import cloud_completion |
| from lg_control.local import run_agent |
| from lg_control.server import Settings, create_app |
| from lg_control.tools import Workspace |
|
|
|
|
| class FakeModel: |
| def __init__(self, parsed): |
| self.parsed = parsed |
| self.calls = [] |
|
|
| def create_chat_completion(self, **kwargs): |
| self.calls.append(kwargs) |
| return {'choices': [{'finish_reason': 'stop', 'message': {'content': json.dumps(self.parsed)}}]} |
|
|
|
|
| class CloudTests(unittest.TestCase): |
| def settings(self): |
| return SimpleNamespace(origin='https://lg.example', worker_token='w'*32, |
| client_id='', client_secret='', password='', redirects=set(), oauth_enabled=False) |
|
|
| def test_qwen_structured_proposal(self): |
| model = FakeModel({'answer': 'Je liste les fichiers.', 'tool': 'list_files', 'arguments': {}}) |
| brain = SimpleNamespace(llm=model, _lock=threading.Lock()) |
| result = cloud_completion([{'role': 'user', 'content': 'Liste les fichiers'}], brain) |
| self.assertEqual(result['message']['tool_calls'][0]['function']['name'], 'list_files') |
| self.assertEqual(model.calls[0]['response_format']['type'], 'json_object') |
| self.assertTrue(brain._lock.acquire(blocking=False)) |
| brain._lock.release() |
|
|
| def test_qwen_unknown_tool_rejected(self): |
| model = FakeModel({'answer': '', 'tool': 'delete_entire_disk', 'arguments': {}}) |
| brain = SimpleNamespace(llm=model, _lock=threading.Lock()) |
| with self.assertRaises(ValueError): |
| cloud_completion([], brain) |
|
|
| def test_cloud_loop_and_local_tools(self): |
| observed = [] |
| def model(messages): |
| observed.append(messages) |
| if len(observed) == 1: |
| return {'message': {'role': 'assistant', 'content': '', 'tool_calls': [{'function': {'name': 'computer_status', 'arguments': {}}}]}} |
| self.assertEqual(messages[-1]['role'], 'tool') |
| self.assertIn('workspace', json.loads(messages[-1]['content'])) |
| return {'message': {'role': 'assistant', 'content': 'PC accessible.'}} |
| with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, {'LG_ASSISTANT_ENABLED': '1'}), TestClient(create_app(self.settings(), model)) as client: |
| self.assertEqual(client.post('/control/chat', json={'messages': []}).status_code, 401) |
| headers = {'Authorization': 'Bearer '+'w'*32} |
| def chat(payload): |
| r = client.post('/control/chat', headers=headers, json={'messages': payload['messages']}) |
| self.assertEqual(r.status_code, 200) |
| return r.json() |
| self.assertEqual(run_agent(Workspace(directory), 'Statut du PC', chat=chat), 'PC accessible.') |
| self.assertEqual(client.get('/authorize').status_code, 503) |
| self.assertEqual(len(observed), 2) |
|
|
| def test_cloud_disabled_by_default(self): |
| with patch.dict(os.environ, {'LG_ASSISTANT_ENABLED': '0'}), TestClient(create_app(self.settings())) as client: |
| response = client.post('/control/chat', headers={'Authorization': 'Bearer '+'w'*32}, json={'messages': [{'role': 'user', 'content': 'bonjour'}]}) |
| self.assertEqual(response.status_code, 503) |
|
|
| def test_oauth_optional_for_cloud_mode(self): |
| with patch.dict(os.environ, {'LG_PUBLIC_URL': 'https://lg.example', 'LG_WORKER_TOKEN': 'w'*32}, clear=True): |
| cfg = Settings() |
| self.assertFalse(cfg.oauth_enabled) |
|
|
| def test_concurrent_inference_rejected(self): |
| release, entered = threading.Event(), threading.Event() |
| def model(messages): |
| entered.set() |
| release.wait(5) |
| return {'message': {'role': 'assistant', 'content': 'OK'}} |
| with patch.dict(os.environ, {'LG_ASSISTANT_ENABLED': '1'}), TestClient(create_app(self.settings(), model)) as client: |
| headers = {'Authorization': 'Bearer '+'w'*32} |
| body = {'messages': [{'role': 'user', 'content': 'bonjour'}]} |
| results = [] |
| thread = threading.Thread(target=lambda: results.append(client.post('/control/chat', headers=headers, json=body))) |
| thread.start() |
| try: |
| self.assertTrue(entered.wait(3)) |
| self.assertEqual(client.post('/control/chat', headers=headers, json=body).status_code, 409) |
| finally: |
| release.set() |
| thread.join(5) |
| self.assertEqual(results[0].status_code, 200) |
|
|
| class CloudFailureTests(unittest.TestCase): |
| def test_truncated_response_regenerates_before_any_tool(self): |
| class Model: |
| count = 0 |
| def create_chat_completion(self, **kwargs): |
| self.count += 1 |
| if self.count == 1: |
| return {'choices': [{'finish_reason': 'length', 'message': {'content': '{"tool":"write_file"'}}]} |
| return {'choices': [{'finish_reason': 'stop', 'message': {'content': json.dumps({'answer': 'Structure expliquee.', 'tool': '', 'arguments': {}})}}]} |
| model = Model() |
| result = cloud_completion([], SimpleNamespace(llm=model, _lock=threading.Lock())) |
| self.assertEqual(model.count, 2) |
| self.assertNotIn('tool_calls', result['message']) |
|
|
| def test_context_and_truncation_have_distinct_errors(self): |
| from lg_control.cloud import CloudCompletionError |
| class Model: |
| def create_chat_completion(self, **kwargs): |
| raise ValueError('Requested tokens exceed context window') |
| with self.assertRaises(CloudCompletionError) as caught: |
| cloud_completion([], SimpleNamespace(llm=Model(), _lock=threading.Lock())) |
| self.assertEqual(caught.exception.code, 'context_exceeded') |
| class Truncated: |
| def create_chat_completion(self, **kwargs): |
| return {'choices': [{'finish_reason': 'length'}]} |
| with self.assertRaises(CloudCompletionError) as caught: |
| cloud_completion([], SimpleNamespace(llm=Truncated(), _lock=threading.Lock())) |
| self.assertEqual(caught.exception.code, 'response_truncated') |
|
|
| def test_http_preserves_failure_code(self): |
| from lg_control.cloud import CloudCompletionError |
| def backend(messages): |
| raise CloudCompletionError('response_truncated', 'Truncated') |
| cfg = CloudTests().settings() |
| with patch.dict(os.environ, {'LG_ASSISTANT_ENABLED': '1'}), TestClient(create_app(cfg, backend)) as client: |
| response = client.post('/control/chat', headers={'Authorization': 'Bearer '+'w'*32}, json={'messages': [{'role': 'user', 'content': 'test'}]}) |
| self.assertEqual(response.status_code, 422) |
| self.assertEqual(response.json()['detail']['code'], 'response_truncated') |
|
|
|
|
| if __name__ == '__main__': |
| unittest.main() |
|
|
