wehub-resource-sync
928 lines
29 KiB
Python
928 lines
29 KiB
Python
import asyncio
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from free_claude_code.messaging.models import MessageScope
|
|
from free_claude_code.messaging.platforms.voice_flow import (
|
|
VOICE_DISABLED_MESSAGE,
|
|
VOICE_TRANSCRIPTION_ERROR_MESSAGE,
|
|
VoiceNoteFlow,
|
|
VoiceNoteRequest,
|
|
audio_suffix_from_metadata,
|
|
is_audio_metadata,
|
|
)
|
|
from free_claude_code.messaging.trees.runtime import MessageTree
|
|
from free_claude_code.messaging.voice import Transcriber
|
|
from free_claude_code.messaging.workflow import MessagingWorkflow
|
|
|
|
VOICE_SCOPE = MessageScope(platform="telegram", chat_id="chat")
|
|
|
|
|
|
class FatalVoiceError(BaseException):
|
|
"""Fatal sentinel used to verify ownership finalization."""
|
|
|
|
|
|
class MockTranscriber:
|
|
def __init__(self, result: str = "hello from voice") -> None:
|
|
self.run = AsyncMock(return_value=result)
|
|
self.close_run = AsyncMock()
|
|
self.paths: list[Path] = []
|
|
|
|
async def transcribe(self, file_path: Path) -> str:
|
|
self.paths.append(file_path)
|
|
return await self.run(file_path)
|
|
|
|
async def close(self) -> None:
|
|
await self.close_run()
|
|
|
|
|
|
def _flow(*, enabled: bool = True) -> tuple[VoiceNoteFlow, MockTranscriber]:
|
|
transcriber = MockTranscriber()
|
|
configured: Transcriber | None = transcriber if enabled else None
|
|
return (
|
|
VoiceNoteFlow(
|
|
transcriber=configured,
|
|
log_raw_messaging_content=False,
|
|
log_api_error_tracebacks=False,
|
|
),
|
|
transcriber,
|
|
)
|
|
|
|
|
|
def _request(
|
|
*,
|
|
download_to=None,
|
|
reply_text=None,
|
|
message_id: str = "voice",
|
|
) -> VoiceNoteRequest:
|
|
async def default_download_to(path: Path) -> None:
|
|
path.write_bytes(b"voice")
|
|
|
|
return VoiceNoteRequest(
|
|
platform="telegram",
|
|
chat_id="chat",
|
|
user_id="user",
|
|
message_id=message_id,
|
|
raw_event={"raw": True},
|
|
content_type="audio/ogg",
|
|
temp_suffix=".ogg",
|
|
status_text="transcribing",
|
|
status_parse_mode="MarkdownV2",
|
|
message_thread_id="thread",
|
|
reply_to_message_id="reply",
|
|
download_to=download_to or default_download_to,
|
|
reply_text=reply_text or AsyncMock(),
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_flow_success_builds_incoming_message() -> None:
|
|
flow, transcriber = _flow()
|
|
handler = AsyncMock()
|
|
queue_send = AsyncMock(return_value="status")
|
|
queue_delete = AsyncMock()
|
|
downloaded_paths: list[Path] = []
|
|
|
|
async def download_to(path: Path) -> None:
|
|
downloaded_paths.append(path)
|
|
path.write_bytes(b"voice")
|
|
|
|
handled = await flow.handle(
|
|
_request(download_to=download_to),
|
|
message_handler=handler,
|
|
queue_send_message=queue_send,
|
|
queue_delete_messages=queue_delete,
|
|
)
|
|
|
|
assert handled is True
|
|
queue_send.assert_awaited_once_with(
|
|
"chat",
|
|
"transcribing",
|
|
reply_to="voice",
|
|
parse_mode="MarkdownV2",
|
|
fire_and_forget=False,
|
|
message_thread_id="thread",
|
|
)
|
|
queue_delete.assert_not_awaited()
|
|
handler.assert_awaited_once()
|
|
incoming = handler.call_args.args[0]
|
|
assert incoming.text == "hello from voice"
|
|
assert incoming.chat_id == "chat"
|
|
assert incoming.message_id == "voice"
|
|
assert incoming.reply_to_message_id == "reply"
|
|
assert incoming.message_thread_id == "thread"
|
|
assert incoming.status_message_id == "status"
|
|
transcriber.run.assert_awaited_once()
|
|
assert transcriber.paths == downloaded_paths
|
|
assert downloaded_paths and not downloaded_paths[0].exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_flow_disabled_replies_without_transcribing() -> None:
|
|
flow, transcriber = _flow(enabled=False)
|
|
reply_text = AsyncMock()
|
|
|
|
handled = await flow.handle(
|
|
_request(reply_text=reply_text),
|
|
message_handler=AsyncMock(),
|
|
queue_send_message=AsyncMock(),
|
|
queue_delete_messages=AsyncMock(),
|
|
)
|
|
|
|
assert handled is True
|
|
reply_text.assert_awaited_once_with(VOICE_DISABLED_MESSAGE)
|
|
transcriber.run.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_flow_missing_status_id_stops_before_transcription() -> None:
|
|
flow, transcriber = _flow()
|
|
reply_text = AsyncMock()
|
|
handler = AsyncMock()
|
|
queue_delete = AsyncMock()
|
|
|
|
handled = await flow.handle(
|
|
_request(reply_text=reply_text),
|
|
message_handler=handler,
|
|
queue_send_message=AsyncMock(return_value=None),
|
|
queue_delete_messages=queue_delete,
|
|
)
|
|
|
|
assert handled is True
|
|
transcriber.run.assert_not_awaited()
|
|
handler.assert_not_awaited()
|
|
queue_delete.assert_not_awaited()
|
|
reply_text.assert_awaited_once_with(VOICE_TRANSCRIPTION_ERROR_MESSAGE)
|
|
assert await flow.cancel_pending_voice(VOICE_SCOPE, "voice") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_flow_cancelled_transcription_preserves_owned_status() -> None:
|
|
flow, transcriber = _flow()
|
|
|
|
async def canceling_transcribe(_path: Path) -> str:
|
|
await flow.cancel_pending_voice(VOICE_SCOPE, "voice")
|
|
return "ignored"
|
|
|
|
transcriber.run.side_effect = canceling_transcribe
|
|
handler = AsyncMock()
|
|
queue_send = AsyncMock(return_value="status")
|
|
queue_delete = AsyncMock()
|
|
|
|
handled = await flow.handle(
|
|
_request(),
|
|
message_handler=handler,
|
|
queue_send_message=queue_send,
|
|
queue_delete_messages=queue_delete,
|
|
)
|
|
|
|
assert handled is True
|
|
handler.assert_not_awaited()
|
|
queue_delete.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_flow_cancel_during_status_delivery_prevents_transcription() -> (
|
|
None
|
|
):
|
|
flow, transcriber = _flow()
|
|
status_send_started = asyncio.Event()
|
|
release_status_send = asyncio.Event()
|
|
handler = AsyncMock()
|
|
queue_delete = AsyncMock()
|
|
|
|
async def send_status(*_args, **_kwargs) -> str:
|
|
status_send_started.set()
|
|
await release_status_send.wait()
|
|
return "status"
|
|
|
|
handle_task = asyncio.create_task(
|
|
flow.handle(
|
|
_request(),
|
|
message_handler=handler,
|
|
queue_send_message=send_status,
|
|
queue_delete_messages=queue_delete,
|
|
)
|
|
)
|
|
|
|
try:
|
|
await asyncio.wait_for(status_send_started.wait(), timeout=1)
|
|
cancelled = await flow.cancel_pending_voice(VOICE_SCOPE, "voice")
|
|
|
|
assert cancelled is not None
|
|
assert cancelled.voice_message_id == "voice"
|
|
assert cancelled.status_message_id is None
|
|
|
|
release_status_send.set()
|
|
assert await asyncio.wait_for(handle_task, timeout=1) is True
|
|
finally:
|
|
release_status_send.set()
|
|
if not handle_task.done():
|
|
handle_task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await handle_task
|
|
|
|
transcriber.run.assert_not_awaited()
|
|
handler.assert_not_awaited()
|
|
queue_delete.assert_awaited_once_with("chat", ["status"])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_flow_cancel_during_status_delivery_suppresses_late_failure() -> (
|
|
None
|
|
):
|
|
flow, transcriber = _flow()
|
|
status_send_started = asyncio.Event()
|
|
release_status_send = asyncio.Event()
|
|
reply_text = AsyncMock()
|
|
handler = AsyncMock()
|
|
|
|
async def fail_status_send(*_args, **_kwargs) -> str:
|
|
status_send_started.set()
|
|
await release_status_send.wait()
|
|
raise RuntimeError("late status failure")
|
|
|
|
handle_task = asyncio.create_task(
|
|
flow.handle(
|
|
_request(reply_text=reply_text),
|
|
message_handler=handler,
|
|
queue_send_message=fail_status_send,
|
|
queue_delete_messages=AsyncMock(),
|
|
)
|
|
)
|
|
|
|
try:
|
|
await asyncio.wait_for(status_send_started.wait(), timeout=1)
|
|
assert await flow.cancel_pending_voice(VOICE_SCOPE, "voice") is not None
|
|
|
|
release_status_send.set()
|
|
assert await asyncio.wait_for(handle_task, timeout=1) is True
|
|
finally:
|
|
release_status_send.set()
|
|
if not handle_task.done():
|
|
handle_task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await handle_task
|
|
|
|
transcriber.run.assert_not_awaited()
|
|
handler.assert_not_awaited()
|
|
reply_text.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_flow_cancel_during_transcription_suppresses_late_failure() -> None:
|
|
flow, transcriber = _flow()
|
|
transcription_started = asyncio.Event()
|
|
release_transcription = asyncio.Event()
|
|
reply_text = AsyncMock()
|
|
handler = AsyncMock()
|
|
queue_delete = AsyncMock()
|
|
|
|
async def fail_transcription(_path: Path) -> str:
|
|
transcription_started.set()
|
|
await release_transcription.wait()
|
|
raise RuntimeError("late transcription failure")
|
|
|
|
transcriber.run.side_effect = fail_transcription
|
|
handle_task = asyncio.create_task(
|
|
flow.handle(
|
|
_request(reply_text=reply_text),
|
|
message_handler=handler,
|
|
queue_send_message=AsyncMock(return_value="status"),
|
|
queue_delete_messages=queue_delete,
|
|
)
|
|
)
|
|
|
|
try:
|
|
await asyncio.wait_for(transcription_started.wait(), timeout=1)
|
|
cancelled = await flow.cancel_pending_voice(VOICE_SCOPE, "voice")
|
|
assert cancelled is not None
|
|
assert cancelled.status_message_id == "status"
|
|
|
|
release_transcription.set()
|
|
assert await asyncio.wait_for(handle_task, timeout=1) is True
|
|
finally:
|
|
release_transcription.set()
|
|
if not handle_task.done():
|
|
handle_task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await handle_task
|
|
|
|
handler.assert_not_awaited()
|
|
reply_text.assert_not_awaited()
|
|
queue_delete.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reply_stop_status_survives_late_transcription_success(
|
|
mock_platform,
|
|
mock_cli_manager,
|
|
mock_session_store,
|
|
incoming_message_factory,
|
|
) -> None:
|
|
flow, transcriber = _flow()
|
|
workflow = MessagingWorkflow(
|
|
mock_platform,
|
|
mock_cli_manager,
|
|
mock_session_store,
|
|
platform_name="telegram",
|
|
voice_cancellation=flow,
|
|
)
|
|
transcription_started = asyncio.Event()
|
|
release_transcription = asyncio.Event()
|
|
|
|
async def delayed_transcription(_path: Path) -> str:
|
|
transcription_started.set()
|
|
await release_transcription.wait()
|
|
return "late voice prompt"
|
|
|
|
transcriber.run.side_effect = delayed_transcription
|
|
mock_platform.queue_send_message.return_value = "voice_status"
|
|
voice_task = asyncio.create_task(
|
|
flow.handle(
|
|
_request(),
|
|
message_handler=workflow.handle_message,
|
|
queue_send_message=mock_platform.queue_send_message,
|
|
queue_delete_messages=mock_platform.queue_delete_messages,
|
|
)
|
|
)
|
|
await asyncio.wait_for(transcription_started.wait(), timeout=1)
|
|
|
|
try:
|
|
await workflow.handle_message(
|
|
incoming_message_factory(
|
|
text="/stop",
|
|
chat_id="chat",
|
|
message_id="stop_command",
|
|
reply_to_message_id="voice",
|
|
)
|
|
)
|
|
finally:
|
|
release_transcription.set()
|
|
assert await asyncio.wait_for(voice_task, timeout=1) is True
|
|
await asyncio.sleep(0)
|
|
|
|
stopped = workflow.format_status("⏹", "Stopped.")
|
|
assert any(
|
|
call.args[:3] == ("chat", "voice_status", stopped)
|
|
for call in mock_platform.queue_edit_message.await_args_list
|
|
)
|
|
mock_platform.queue_delete_messages.assert_not_awaited()
|
|
assert await workflow.tree_queue.resolve_node_id(VOICE_SCOPE, "voice") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_flow_cancel_during_handoff_stops_and_drains_handler() -> None:
|
|
flow, _transcriber = _flow()
|
|
handler_started = asyncio.Event()
|
|
handler_cancelled = asyncio.Event()
|
|
release_handler = asyncio.Event()
|
|
queue_delete = AsyncMock()
|
|
|
|
async def handler(_incoming) -> None:
|
|
handler_started.set()
|
|
try:
|
|
await asyncio.Event().wait()
|
|
except asyncio.CancelledError:
|
|
handler_cancelled.set()
|
|
await release_handler.wait()
|
|
raise
|
|
|
|
handle_task = asyncio.create_task(
|
|
flow.handle(
|
|
_request(),
|
|
message_handler=handler,
|
|
queue_send_message=AsyncMock(return_value="status"),
|
|
queue_delete_messages=queue_delete,
|
|
)
|
|
)
|
|
|
|
try:
|
|
await asyncio.wait_for(handler_started.wait(), timeout=1)
|
|
cancel_task = asyncio.create_task(
|
|
flow.cancel_pending_voice(VOICE_SCOPE, "voice")
|
|
)
|
|
await asyncio.wait_for(handler_cancelled.wait(), timeout=1)
|
|
|
|
assert not cancel_task.done()
|
|
release_handler.set()
|
|
cancelled = await asyncio.wait_for(cancel_task, timeout=1)
|
|
assert cancelled is not None
|
|
assert cancelled.status_message_id == "status"
|
|
assert await asyncio.wait_for(handle_task, timeout=1) is True
|
|
finally:
|
|
release_handler.set()
|
|
if not handle_task.done():
|
|
handle_task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await handle_task
|
|
|
|
queue_delete.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_flow_cancel_during_handoff_suppresses_late_handler_error() -> None:
|
|
flow, _transcriber = _flow()
|
|
handler_started = asyncio.Event()
|
|
reply_text = AsyncMock()
|
|
|
|
async def handler(_incoming) -> None:
|
|
handler_started.set()
|
|
try:
|
|
await asyncio.Event().wait()
|
|
except asyncio.CancelledError:
|
|
raise RuntimeError("late handler failure") from None
|
|
|
|
handle_task = asyncio.create_task(
|
|
flow.handle(
|
|
_request(reply_text=reply_text),
|
|
message_handler=handler,
|
|
queue_send_message=AsyncMock(return_value="status"),
|
|
queue_delete_messages=AsyncMock(),
|
|
)
|
|
)
|
|
|
|
await asyncio.wait_for(handler_started.wait(), timeout=1)
|
|
assert await flow.cancel_pending_voice(VOICE_SCOPE, "status") is not None
|
|
assert await asyncio.wait_for(handle_task, timeout=1) is True
|
|
reply_text.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("command", ["/stop", "/clear"])
|
|
async def test_reply_command_cancels_voice_at_tree_admission_commit(
|
|
command: str,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
mock_platform,
|
|
mock_cli_manager,
|
|
mock_session_store,
|
|
incoming_message_factory,
|
|
) -> None:
|
|
flow, _transcriber = _flow()
|
|
workflow = MessagingWorkflow(
|
|
mock_platform,
|
|
mock_cli_manager,
|
|
mock_session_store,
|
|
platform_name="telegram",
|
|
voice_cancellation=flow,
|
|
)
|
|
admission_mutated = asyncio.Event()
|
|
release_admission = asyncio.Event()
|
|
original_enqueue_or_claim = MessageTree.enqueue_or_claim
|
|
|
|
async def mutate_then_block(tree: MessageTree, node_id: str):
|
|
decision = await original_enqueue_or_claim(tree, node_id)
|
|
if node_id == "voice":
|
|
admission_mutated.set()
|
|
await release_admission.wait()
|
|
return decision
|
|
|
|
monkeypatch.setattr(MessageTree, "enqueue_or_claim", mutate_then_block)
|
|
mock_platform.queue_send_message.return_value = "voice_status"
|
|
voice_task = asyncio.create_task(
|
|
flow.handle(
|
|
_request(),
|
|
message_handler=workflow.handle_message,
|
|
queue_send_message=mock_platform.queue_send_message,
|
|
queue_delete_messages=mock_platform.queue_delete_messages,
|
|
)
|
|
)
|
|
await asyncio.wait_for(admission_mutated.wait(), timeout=1)
|
|
|
|
command_task = asyncio.create_task(
|
|
workflow.handle_message(
|
|
incoming_message_factory(
|
|
text=command,
|
|
chat_id="chat",
|
|
message_id="command",
|
|
reply_to_message_id="voice",
|
|
)
|
|
)
|
|
)
|
|
try:
|
|
await asyncio.sleep(0)
|
|
assert not command_task.done()
|
|
finally:
|
|
release_admission.set()
|
|
await asyncio.wait_for(command_task, timeout=1)
|
|
voice_result = await asyncio.wait_for(voice_task, timeout=1)
|
|
assert voice_result is True
|
|
await asyncio.sleep(0)
|
|
|
|
mock_session_store.save_tree_snapshot.assert_called()
|
|
if command == "/clear":
|
|
assert workflow.get_tree_count() == 0
|
|
assert await workflow.tree_queue.resolve_node_id(VOICE_SCOPE, "voice") is None
|
|
else:
|
|
assert workflow.get_tree_count() == 1
|
|
assert (
|
|
await workflow.tree_queue.resolve_node_id(VOICE_SCOPE, "voice") == "voice"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("command", ["/stop", "/clear"])
|
|
async def test_global_command_rejects_transcription_that_finishes_late(
|
|
command: str,
|
|
mock_platform,
|
|
mock_cli_manager,
|
|
mock_session_store,
|
|
incoming_message_factory,
|
|
) -> None:
|
|
flow, transcriber = _flow()
|
|
workflow = MessagingWorkflow(
|
|
mock_platform,
|
|
mock_cli_manager,
|
|
mock_session_store,
|
|
platform_name="telegram",
|
|
voice_cancellation=flow,
|
|
)
|
|
transcription_started = asyncio.Event()
|
|
release_transcription = asyncio.Event()
|
|
|
|
async def delayed_transcription(_path: Path) -> str:
|
|
transcription_started.set()
|
|
await release_transcription.wait()
|
|
return "late voice prompt"
|
|
|
|
transcriber.run.side_effect = delayed_transcription
|
|
mock_platform.queue_send_message.return_value = "voice_status"
|
|
voice_task = asyncio.create_task(
|
|
flow.handle(
|
|
_request(),
|
|
message_handler=workflow.handle_message,
|
|
queue_send_message=mock_platform.queue_send_message,
|
|
queue_delete_messages=mock_platform.queue_delete_messages,
|
|
)
|
|
)
|
|
await asyncio.wait_for(transcription_started.wait(), timeout=1)
|
|
|
|
await asyncio.wait_for(
|
|
workflow.handle_message(
|
|
incoming_message_factory(
|
|
text=command,
|
|
message_id="command",
|
|
chat_id=VOICE_SCOPE.chat_id,
|
|
)
|
|
),
|
|
timeout=1,
|
|
)
|
|
release_transcription.set()
|
|
assert await asyncio.wait_for(voice_task, timeout=1) is True
|
|
|
|
assert workflow.get_tree_count() == 0
|
|
assert await workflow.tree_queue.resolve_node_id(VOICE_SCOPE, "voice") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("command", ["/stop", "/clear"])
|
|
async def test_transcribed_global_command_does_not_cancel_its_own_handoff(
|
|
command: str,
|
|
mock_platform,
|
|
mock_cli_manager,
|
|
mock_session_store,
|
|
) -> None:
|
|
flow, transcriber = _flow()
|
|
transcriber.run.return_value = command
|
|
workflow = MessagingWorkflow(
|
|
mock_platform,
|
|
mock_cli_manager,
|
|
mock_session_store,
|
|
platform_name="telegram",
|
|
voice_cancellation=flow,
|
|
)
|
|
mock_platform.queue_send_message.return_value = "voice_status"
|
|
|
|
assert (
|
|
await asyncio.wait_for(
|
|
flow.handle(
|
|
_request(),
|
|
message_handler=workflow.handle_message,
|
|
queue_send_message=mock_platform.queue_send_message,
|
|
queue_delete_messages=mock_platform.queue_delete_messages,
|
|
),
|
|
timeout=1,
|
|
)
|
|
is True
|
|
)
|
|
|
|
assert await flow.cancel_pending_voice(VOICE_SCOPE, "voice") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_flow_caller_cancellation_drains_handoff_child() -> None:
|
|
flow, _transcriber = _flow()
|
|
handler_started = asyncio.Event()
|
|
handler_cancelled = asyncio.Event()
|
|
release_handler = asyncio.Event()
|
|
queue_delete = AsyncMock()
|
|
|
|
async def handler(_incoming) -> None:
|
|
handler_started.set()
|
|
try:
|
|
await asyncio.Event().wait()
|
|
except asyncio.CancelledError:
|
|
handler_cancelled.set()
|
|
await release_handler.wait()
|
|
raise
|
|
|
|
handle_task = asyncio.create_task(
|
|
flow.handle(
|
|
_request(),
|
|
message_handler=handler,
|
|
queue_send_message=AsyncMock(return_value="status"),
|
|
queue_delete_messages=queue_delete,
|
|
)
|
|
)
|
|
await asyncio.wait_for(handler_started.wait(), timeout=1)
|
|
handle_task.cancel()
|
|
await asyncio.wait_for(handler_cancelled.wait(), timeout=1)
|
|
|
|
assert not handle_task.done()
|
|
release_handler.set()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await handle_task
|
|
assert await flow.cancel_pending_voice(VOICE_SCOPE, "voice") is None
|
|
queue_delete.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_flow_task_cancellation_waits_then_cleans_pending_state() -> None:
|
|
flow, transcriber = _flow()
|
|
started = asyncio.Event()
|
|
cancellation_received = asyncio.Event()
|
|
release = asyncio.Event()
|
|
stopped = asyncio.Event()
|
|
|
|
async def cancellation_safe_transcribe(_path: Path) -> str:
|
|
started.set()
|
|
try:
|
|
await asyncio.Event().wait()
|
|
return "unreachable"
|
|
except asyncio.CancelledError:
|
|
cancellation_received.set()
|
|
await release.wait()
|
|
stopped.set()
|
|
raise
|
|
|
|
transcriber.run.side_effect = cancellation_safe_transcribe
|
|
handler = AsyncMock()
|
|
queue_delete = AsyncMock()
|
|
handle_task = asyncio.create_task(
|
|
flow.handle(
|
|
_request(),
|
|
message_handler=handler,
|
|
queue_send_message=AsyncMock(return_value="status"),
|
|
queue_delete_messages=queue_delete,
|
|
)
|
|
)
|
|
|
|
await started.wait()
|
|
handle_task.cancel()
|
|
await cancellation_received.wait()
|
|
|
|
assert not handle_task.done()
|
|
queue_delete.assert_not_awaited()
|
|
|
|
release.set()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await handle_task
|
|
|
|
assert stopped.is_set()
|
|
handler.assert_not_awaited()
|
|
queue_delete.assert_awaited_once_with("chat", ["status"])
|
|
assert await flow.cancel_pending_voice(VOICE_SCOPE, "voice") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_flow_repeated_cancellation_cannot_interrupt_cleanup(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
flow, transcriber = _flow()
|
|
transcription_started = asyncio.Event()
|
|
cleanup_started = asyncio.Event()
|
|
release_cleanup = asyncio.Event()
|
|
|
|
async def blocked_transcription(_path: Path) -> str:
|
|
transcription_started.set()
|
|
await asyncio.Event().wait()
|
|
return "unreachable"
|
|
|
|
original_discard = flow._pending_voice.discard
|
|
|
|
async def delayed_discard(claim) -> bool:
|
|
cleanup_started.set()
|
|
await release_cleanup.wait()
|
|
return await original_discard(claim)
|
|
|
|
transcriber.run.side_effect = blocked_transcription
|
|
monkeypatch.setattr(flow._pending_voice, "discard", delayed_discard)
|
|
queue_delete = AsyncMock()
|
|
handle_task = asyncio.create_task(
|
|
flow.handle(
|
|
_request(),
|
|
message_handler=AsyncMock(),
|
|
queue_send_message=AsyncMock(return_value="status"),
|
|
queue_delete_messages=queue_delete,
|
|
)
|
|
)
|
|
|
|
await asyncio.wait_for(transcription_started.wait(), timeout=1)
|
|
handle_task.cancel("first cancellation")
|
|
await asyncio.wait_for(cleanup_started.wait(), timeout=1)
|
|
handle_task.cancel("second cancellation")
|
|
await asyncio.sleep(0)
|
|
|
|
assert not handle_task.done()
|
|
release_cleanup.set()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await asyncio.wait_for(handle_task, timeout=1)
|
|
|
|
queue_delete.assert_awaited_once_with("chat", ["status"])
|
|
assert await flow.cancel_pending_voice(VOICE_SCOPE, "voice") is None
|
|
assert await flow.cancel_pending_voice(VOICE_SCOPE, "status") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_flow_fatal_status_failure_releases_claim() -> None:
|
|
flow, transcriber = _flow()
|
|
queue_delete = AsyncMock()
|
|
|
|
async def fatal_status(*_args, **_kwargs) -> str:
|
|
raise FatalVoiceError
|
|
|
|
with pytest.raises(FatalVoiceError):
|
|
await flow.handle(
|
|
_request(),
|
|
message_handler=AsyncMock(),
|
|
queue_send_message=fatal_status,
|
|
queue_delete_messages=queue_delete,
|
|
)
|
|
|
|
transcriber.run.assert_not_awaited()
|
|
queue_delete.assert_not_awaited()
|
|
assert await flow.cancel_pending_voice(VOICE_SCOPE, "voice") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_flow_fatal_download_failure_releases_status_ownership() -> None:
|
|
flow, transcriber = _flow()
|
|
queue_delete = AsyncMock()
|
|
|
|
async def fatal_download(_path: Path) -> None:
|
|
raise FatalVoiceError
|
|
|
|
with pytest.raises(FatalVoiceError):
|
|
await flow.handle(
|
|
_request(download_to=fatal_download),
|
|
message_handler=AsyncMock(),
|
|
queue_send_message=AsyncMock(return_value="status"),
|
|
queue_delete_messages=queue_delete,
|
|
)
|
|
|
|
transcriber.run.assert_not_awaited()
|
|
queue_delete.assert_awaited_once_with("chat", ["status"])
|
|
assert await flow.cancel_pending_voice(VOICE_SCOPE, "voice") is None
|
|
assert await flow.cancel_pending_voice(VOICE_SCOPE, "status") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_flow_fatal_transcription_releases_status_ownership() -> None:
|
|
flow, transcriber = _flow()
|
|
transcriber.run.side_effect = FatalVoiceError()
|
|
handler = AsyncMock()
|
|
queue_delete = AsyncMock()
|
|
|
|
with pytest.raises(FatalVoiceError):
|
|
await flow.handle(
|
|
_request(),
|
|
message_handler=handler,
|
|
queue_send_message=AsyncMock(return_value="status"),
|
|
queue_delete_messages=queue_delete,
|
|
)
|
|
|
|
handler.assert_not_awaited()
|
|
queue_delete.assert_awaited_once_with("chat", ["status"])
|
|
assert await flow.cancel_pending_voice(VOICE_SCOPE, "voice") is None
|
|
assert await flow.cancel_pending_voice(VOICE_SCOPE, "status") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_flow_download_failure_cleans_pending_state() -> None:
|
|
flow, transcriber = _flow()
|
|
reply_text = AsyncMock()
|
|
queue_delete = AsyncMock()
|
|
|
|
async def failing_download(_path: Path) -> None:
|
|
raise RuntimeError("download failed")
|
|
|
|
handled = await flow.handle(
|
|
_request(download_to=failing_download, reply_text=reply_text),
|
|
message_handler=AsyncMock(),
|
|
queue_send_message=AsyncMock(return_value="status"),
|
|
queue_delete_messages=queue_delete,
|
|
)
|
|
|
|
assert handled is True
|
|
transcriber.run.assert_not_awaited()
|
|
queue_delete.assert_awaited_once_with("chat", ["status"])
|
|
reply_text.assert_awaited_once_with(VOICE_TRANSCRIPTION_ERROR_MESSAGE)
|
|
assert await flow.cancel_pending_voice(VOICE_SCOPE, "voice") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_flow_transcription_failure_cleans_pending_state() -> None:
|
|
flow, transcriber = _flow()
|
|
transcriber.run.side_effect = RuntimeError("transcription failed")
|
|
reply_text = AsyncMock()
|
|
queue_delete = AsyncMock()
|
|
|
|
handled = await flow.handle(
|
|
_request(reply_text=reply_text),
|
|
message_handler=AsyncMock(),
|
|
queue_send_message=AsyncMock(return_value="status"),
|
|
queue_delete_messages=queue_delete,
|
|
)
|
|
|
|
assert handled is True
|
|
queue_delete.assert_awaited_once_with("chat", ["status"])
|
|
reply_text.assert_awaited_once_with(VOICE_TRANSCRIPTION_ERROR_MESSAGE)
|
|
assert await flow.cancel_pending_voice(VOICE_SCOPE, "voice") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_flow_handler_failure_cleans_pending_without_deleting_status() -> (
|
|
None
|
|
):
|
|
flow, _transcriber = _flow()
|
|
reply_text = AsyncMock()
|
|
queue_delete = AsyncMock()
|
|
|
|
async def failing_handler(_incoming) -> None:
|
|
raise RuntimeError("handler failed")
|
|
|
|
handled = await flow.handle(
|
|
_request(reply_text=reply_text),
|
|
message_handler=failing_handler,
|
|
queue_send_message=AsyncMock(return_value="status"),
|
|
queue_delete_messages=queue_delete,
|
|
)
|
|
|
|
assert handled is True
|
|
queue_delete.assert_not_awaited()
|
|
reply_text.assert_awaited_once_with(VOICE_TRANSCRIPTION_ERROR_MESSAGE)
|
|
assert await flow.cancel_pending_voice(VOICE_SCOPE, "voice") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_flow_rejects_oversized_audio_before_transcription(
|
|
monkeypatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
"free_claude_code.messaging.platforms.voice_flow.MAX_AUDIO_SIZE_BYTES",
|
|
3,
|
|
)
|
|
flow, transcriber = _flow()
|
|
reply_text = AsyncMock()
|
|
queue_delete = AsyncMock()
|
|
|
|
async def download(path: Path) -> None:
|
|
path.write_bytes(b"four")
|
|
|
|
handled = await flow.handle(
|
|
_request(download_to=download, reply_text=reply_text),
|
|
message_handler=AsyncMock(),
|
|
queue_send_message=AsyncMock(return_value="status"),
|
|
queue_delete_messages=queue_delete,
|
|
)
|
|
|
|
assert handled is True
|
|
transcriber.run.assert_not_awaited()
|
|
queue_delete.assert_awaited_once_with("chat", ["status"])
|
|
assert reply_text.await_args is not None
|
|
assert "too large" in reply_text.await_args.args[0]
|
|
|
|
|
|
def test_audio_metadata_helpers() -> None:
|
|
assert is_audio_metadata("voice.ogg", "application/octet-stream") is True
|
|
assert is_audio_metadata("file.txt", "audio/ogg") is True
|
|
assert is_audio_metadata("file.txt", "text/plain") is False
|
|
assert (
|
|
audio_suffix_from_metadata(filename="voice.ogg", content_type="audio/mp4")
|
|
== ".mp4"
|
|
)
|
|
assert (
|
|
audio_suffix_from_metadata(filename="clip.m4a", content_type="audio/mp4")
|
|
== ".m4a"
|
|
)
|
|
assert (
|
|
audio_suffix_from_metadata(filename="clip.m4a", content_type="audio/mpeg")
|
|
== ".mp3"
|
|
)
|
|
assert audio_suffix_from_metadata(content_type="audio/mpeg") == ".mp3"
|
|
assert audio_suffix_from_metadata(filename="clip.m4a") == ".m4a"
|
|
assert audio_suffix_from_metadata(content_type="audio/mp4") == ".mp4"
|
|
assert audio_suffix_from_metadata(content_type="audio/wav") == ".wav"
|