"""Network-free provider contracts, safe failures and single-provider selection."""
import asyncio
import base64
import json
import logging
import subprocess
import sys
from types import SimpleNamespace

import httpx
import pytest

from device_gateway.voice import (
    CF_STT_MODEL, CloudflareWorkersAISttProvider, OpenAICompatibleSTT,
    TranscriptionError, VoiceManager, VoiceRequest, configured_stt, wav_pcm,
)

KEY = 'test-private-key-never-log'
WAV = wav_pcm(b'public-audio-marker!' * 100)


def mock_http(monkeypatch, handler):
    client = httpx.AsyncClient
    calls = []

    def handle(request):
        calls.append(request)
        return handler(request)

    def factory(**kwargs):
        assert kwargs == {'timeout': 25, 'follow_redirects': False}
        return client(**kwargs, transport=httpx.MockTransport(handle))

    monkeypatch.setattr(httpx, 'AsyncClient', factory)
    return calls


def make_provider(name, prompt=None):
    if name == 'cloudflare':
        return CloudflareWorkersAISttProvider('test-account', KEY, initial_prompt=prompt)
    return OpenAICompatibleSTT('https://api.groq.com/openai/v1/', KEY,
                               'whisper-large-v3-turbo', initial_prompt=prompt)


@pytest.mark.parametrize('prompt', [None, 'Codex, Pi, session, branch, pytest'])
def test_cloudflare_serialization_and_success(monkeypatch, prompt):
    def handle(request):
        assert str(request.url) == (
            'https://api.cloudflare.com/client/v4/accounts/test-account/ai/run/' + CF_STT_MODEL)
        assert request.headers['authorization'] == 'Bearer ' + KEY
        assert request.headers['content-type'] == 'application/json'
        payload = json.loads(request.content)
        assert base64.b64decode(payload.pop('audio'), validate=True) == WAV
        assert payload == ({'task': 'transcribe', 'initial_prompt': prompt}
                           if prompt else {'task': 'transcribe'})
        return httpx.Response(200, json={'success': True, 'errors': [],
            'result': {'text': ' 请 resume Codex session，然后 checkout branch。 ',
                       'transcription_info': {'language': 'zh'}}})

    calls = mock_http(monkeypatch, handle)
    provider = make_provider('cloudflare', prompt)
    assert asyncio.run(provider.transcribe(WAV)) == '请 resume Codex session，然后 checkout branch。'
    assert len(calls) == 1
    assert provider.last_request_seconds >= 0


@pytest.mark.parametrize('prompt', [None, 'Codex, pytest'])
def test_groq_openai_compatible_regression(monkeypatch, prompt):
    def handle(request):
        assert str(request.url) == 'https://api.groq.com/openai/v1/audio/transcriptions'
        assert request.headers['authorization'] == 'Bearer ' + KEY
        assert request.headers['content-type'].startswith('multipart/form-data; boundary=')
        assert WAV in request.content
        assert b'name="model"\r\n\r\nwhisper-large-v3-turbo' in request.content
        assert b'filename="voice.wav"' in request.content
        assert b'name="language"' not in request.content
        assert b'name="initial_prompt"' not in request.content
        assert (b'name="prompt"' in request.content) == bool(prompt)
        return httpx.Response(200, json={'text': 'run pytest'})

    calls = mock_http(monkeypatch, handle)
    assert asyncio.run(make_provider('groq', prompt).transcribe(WAV)) == 'run pytest'
    assert len(calls) == 1


FAILURES = [401, 403, 429, 500, 503, 302, 'timeout', 'network', 'invalid-json',
            'empty', 'blank', 'missing', 'non-string', 'non-object', 'bad-result', 'envelope-error']


@pytest.mark.parametrize('name', ['cloudflare', 'groq'])
@pytest.mark.parametrize('failure', FAILURES)
def test_safe_failures_no_retry_or_secret_logs(monkeypatch, caplog, name, failure):
    def handle(request):
        if failure in ('timeout', 'network'):
            error = httpx.ReadTimeout if failure == 'timeout' else httpx.ConnectError
            raise error(KEY + ' public-audio-marker!', request=request)
        if isinstance(failure, int):
            return httpx.Response(failure, text=KEY + ' public-audio-marker!',
                                  headers={'Location': 'https://untrusted.example/'})
        if failure == 'invalid-json':
            return httpx.Response(200, text=KEY)
        payload = {'text': {'empty': '', 'blank': ' \n\x00', 'non-string': 42}.get(failure)}
        if failure == 'missing':
            payload = {}
        if failure == 'non-object':
            payload = []
        if name == 'cloudflare':
            payload = {'success': True, 'result': payload, 'errors': []}
            if failure == 'bad-result':
                payload['result'] = None
            if failure == 'envelope-error':
                payload.update(success=False, errors=[{'message': KEY}])
        return httpx.Response(200, json=payload)

    calls = mock_http(monkeypatch, handle)
    caplog.set_level(logging.DEBUG)
    with pytest.raises(TranscriptionError) as exc:
        asyncio.run(make_provider(name).transcribe(WAV))
    assert str(exc.value) == 'Transcription failed'
    assert len(calls) == 1
    assert 'STT provider=' in caplog.text and 'request_seconds=' in caplog.text
    expected = ('http_status' if isinstance(failure, int) else
                'timeout' if failure == 'timeout' else
                'network' if failure == 'network' else 'invalid_response')
    assert 'outcome=' + expected in caplog.text
    for private in (KEY, 'public-audio-marker!', base64.b64encode(WAV).decode(), 'Authorization'):
        assert private not in caplog.text


@pytest.mark.parametrize('failure', [401, 403, 429, 500, 'timeout', 'empty'])
def test_cloudflare_failure_never_sends_agent_input(monkeypatch, failure):
    def handle(request):
        if failure == 'timeout':
            raise httpx.ReadTimeout(KEY, request=request)
        if failure == 'empty':
            return httpx.Response(200, json={'success': True, 'result': {'text': ''}})
        return httpx.Response(failure, text=KEY)

    mock_http(monkeypatch, handle)

    async def exercise():
        sent, inputs = [], []
        async def send(device, message): sent.append(message)
        async def send_input(*args): inputs.append(args)
        manager = VoiceManager(make_provider('cloudflare'), lambda _: 'agent:pi:s',
            lambda _: SimpleNamespace(send_input=send_input), send, lambda _: True, lambda: False)
        device = SimpleNamespace(connection_id='c', device_id='d')
        request = VoiceRequest('v', 'c', 'agent:pi:s', 'pi', 's', 'session',
                               state='TRANSCRIBING', pcm=bytearray(b'\0\0' * 4410))
        manager.requests['c'] = request
        try:
            await manager.transcribe(device, request)
            assert sent[-1]['type'] == 'voice_error'
            assert sent[-1]['state'] == 'FAILED'
            assert sent[-1]['message'] == 'Transcription failed'
            await manager.handle(device, {'type': 'voice_decision', 'voice_id': 'v', 'decision': 'send'})
            assert inputs == [] and request.transcript == '' and not request.pcm
        finally:
            await manager.close()

    asyncio.run(exercise())


def test_cancellation_propagates(monkeypatch):
    def handle(request): raise asyncio.CancelledError()
    calls = mock_http(monkeypatch, handle)
    with pytest.raises(asyncio.CancelledError):
        asyncio.run(make_provider('cloudflare').transcribe(WAV))
    assert len(calls) == 1


def test_voice_diagnostics_visible_with_production_uvicorn_logging():
    # Configure in a child process so pytest's capture handlers remain untouched.
    result = subprocess.run([sys.executable, '-c', '''
from logging.config import dictConfig
from uvicorn.config import LOGGING_CONFIG
from device_gateway.voice import logger
dictConfig(LOGGING_CONFIG)
logger.info('[VOICE_RX] production-logging-probe')
'''], capture_output=True, text=True, check=True)
    assert '[VOICE_RX] production-logging-probe' in result.stderr


def test_configuration_is_explicit_no_fallback(monkeypatch):
    import os
    for key in list(os.environ):
        if key.startswith(('GATEWAY_STT_', 'GATEWAY_CF_')) or key == 'GROQ_API_KEY':
            monkeypatch.delenv(key)
    monkeypatch.setenv('GROQ_API_KEY', KEY)
    assert configured_stt() is None
    monkeypatch.setenv('GATEWAY_STT_PROVIDER', 'cloudflare-workers-ai')
    assert configured_stt() is None  # Never fall back to the existing Groq key.
    monkeypatch.setenv('GATEWAY_CF_ACCOUNT_ID', 'test-account')
    assert configured_stt() is None
    monkeypatch.setenv('GATEWAY_CF_API_TOKEN', KEY)
    provider = configured_stt()
    assert isinstance(provider, CloudflareWorkersAISttProvider)
    assert provider.model == CF_STT_MODEL and provider.initial_prompt is None
    monkeypatch.setenv('GATEWAY_STT_PROVIDER', 'openai-compatible')
    monkeypatch.setenv('GATEWAY_STT_BASE_URL', 'https://api.groq.com/openai/v1')
    monkeypatch.setenv('GATEWAY_STT_MODEL', 'whisper-large-v3-turbo')
    provider = configured_stt()
    assert isinstance(provider, OpenAICompatibleSTT) and provider.api_key == KEY
    monkeypatch.setenv('GATEWAY_STT_BASE_URL', 'https://another-provider.example/v1')
    assert configured_stt() is None  # Never disclose Groq's key to another host.
    monkeypatch.setenv('GATEWAY_STT_API_KEY', 'other-authorized-key')
    assert isinstance(configured_stt(), OpenAICompatibleSTT)
    monkeypatch.setenv('GATEWAY_STT_INITIAL_PROMPT', 'Codex, pytest')
    assert configured_stt().initial_prompt == 'Codex, pytest'
    monkeypatch.setenv('GATEWAY_STT_PROVIDER', 'unknown')
    assert configured_stt() is None