78 lines
3.9 KiB
Python
78 lines
3.9 KiB
Python
"""Offline Azure catalog tests: no Azure credentials or live services needed."""
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import httpx
|
|
from functools import wraps
|
|
|
|
|
|
def async_test(fn):
|
|
@wraps(fn)
|
|
def run(*args, **kwargs):
|
|
return asyncio.run(fn(*args, **kwargs))
|
|
return run
|
|
from web.voice_catalog import DEFAULT_VOICE, VoiceCatalog, english_voices, normalize_default_voice
|
|
from web import token_server
|
|
|
|
|
|
def test_english_catalog_and_omni_variants():
|
|
voices = english_voices([
|
|
{'ShortName': 'en-GB-SoniaNeural', 'DisplayName': 'Sonia', 'Locale': 'en-GB', 'Gender': 'Female'},
|
|
{'ShortName': 'en-US-Ava:DragonHDOmniLatestNeural', 'DisplayName': 'Ava', 'Locale': 'en-US', 'Gender': 'Female'},
|
|
{'ShortName': 'fr-FR-RemyMultilingualNeural', 'Locale': 'fr-FR', 'SecondaryLocaleList': ['en-US']},
|
|
{'ShortName': 'fr-FR-DeniseNeural', 'Locale': 'fr-FR'},
|
|
{'ShortName': 'en-US-MultiTalker-Ava-Andrew:DragonHDLatestNeural', 'Locale': 'en-US'},
|
|
])
|
|
ids = [v['id'] for v in voices]
|
|
assert ids[0] == DEFAULT_VOICE
|
|
assert 'en-GB-Sonia:DragonHDOmniLatestNeural' in ids
|
|
assert 'en-GB-SoniaNeural' in ids
|
|
assert 'fr-FR-RemyMultilingualNeural' in ids
|
|
assert not any('Denise' in v or 'MultiTalker' in v for v in ids)
|
|
assert len(ids) == len(set(ids))
|
|
|
|
|
|
def test_default_migration():
|
|
assert normalize_default_voice('') == DEFAULT_VOICE
|
|
assert normalize_default_voice('en-US-Ava:DragonHDLatestNeural') == DEFAULT_VOICE
|
|
assert normalize_default_voice('en-GB-SoniaNeural') == 'en-GB-SoniaNeural'
|
|
|
|
|
|
@async_test
|
|
async def test_cache_and_safe_fallback():
|
|
catalog = VoiceCatalog('eastus', 'private-key')
|
|
response = httpx.Response(200, json=[{'ShortName': 'en-US-AvaNeural', 'Locale': 'en-US', 'Gender': 'Female'}], request=httpx.Request('GET', 'https://example.test'))
|
|
with patch('httpx.AsyncClient.get', new=AsyncMock(return_value=response)) as get:
|
|
a, b = await asyncio.gather(catalog.get(), catalog.get())
|
|
assert a == b and a['catalog_available']
|
|
assert get.await_count == 1
|
|
a['voices'].clear()
|
|
assert (await catalog.get())['voices']
|
|
catalog._expires = 0
|
|
with patch('httpx.AsyncClient.get', new=AsyncMock(side_effect=httpx.ConnectError('private-key'))):
|
|
result = await catalog.get()
|
|
assert not result['catalog_available'] and len(result['voices']) > 1
|
|
assert 'private-key' not in str(result)
|
|
invalid = VoiceCatalog('evil.test/path', 'private-key')
|
|
with patch('httpx.AsyncClient.get', new=AsyncMock()) as get:
|
|
assert not (await invalid.get())['catalog_available']
|
|
get.assert_not_awaited()
|
|
|
|
|
|
@async_test
|
|
async def test_catalog_requires_authenticated_account(monkeypatch):
|
|
monkeypatch.setattr(token_server, 'CONTROL_SECRET', 'control-secret')
|
|
transport = httpx.ASGITransport(app=token_server.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url='https://hope.test') as client:
|
|
assert (await client.get('/voices')).status_code == 401
|
|
response = httpx.Response(200, json={'user': {'id': 'alice'}}, request=httpx.Request('POST', 'https://example.test'))
|
|
with patch('httpx.AsyncClient.post', new=AsyncMock(return_value=response)), patch.object(token_server.catalog, 'get', new=AsyncMock(return_value={'voices': [], 'catalog_available': True})):
|
|
result = await client.get('/voices', headers={'Authorization': 'Bearer session'})
|
|
assert result.status_code == 200
|
|
assert result.headers['cache-control'] == 'no-store'
|
|
assert 'control-secret' not in result.text
|
|
response = httpx.Response(401, request=httpx.Request('POST', 'https://example.test'))
|
|
with patch('httpx.AsyncClient.post', new=AsyncMock(return_value=response)), patch.object(token_server.catalog, 'get', new=AsyncMock()) as get:
|
|
assert (await client.get('/voices', headers={'Authorization': 'Bearer revoked'})).status_code == 401
|
|
get.assert_not_awaited()
|