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()