132 lines
3.9 KiB
Python
132 lines
3.9 KiB
Python
import asyncio
|
|
import os
|
|
import sys
|
|
import wave
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
|
|
|
|
import discord
|
|
|
|
import bot
|
|
import voice
|
|
|
|
|
|
class FakeChannel:
|
|
def __init__(self, failures_before_success=0, channel_id=999):
|
|
self.failures_before_success = failures_before_success
|
|
self.messages = []
|
|
self.attempts = 0
|
|
self.id = channel_id
|
|
|
|
async def send(self, text):
|
|
self.attempts += 1
|
|
if self.attempts <= self.failures_before_success:
|
|
response = SimpleNamespace(status=403, reason="Forbidden")
|
|
raise discord.Forbidden(response, {"message": "Missing Permissions", "code": 50013})
|
|
self.messages.append(text)
|
|
|
|
|
|
def test_safe_send_chunked_falls_back_to_secondary_channel():
|
|
async def run():
|
|
primary = FakeChannel(failures_before_success=1, channel_id=100)
|
|
fallback = FakeChannel(channel_id=200)
|
|
delivered = await bot.safe_send_chunked(
|
|
primary,
|
|
"hello world",
|
|
fallback_channels=[fallback],
|
|
purpose="test notification",
|
|
)
|
|
return delivered, fallback.messages
|
|
|
|
delivered, messages = asyncio.run(run())
|
|
assert messages == ["hello world"]
|
|
assert delivered is not None
|
|
|
|
|
|
def test_safe_send_chunked_returns_none_when_all_channels_fail():
|
|
async def run():
|
|
primary = FakeChannel(failures_before_success=10, channel_id=100)
|
|
fallback = FakeChannel(failures_before_success=10, channel_id=200)
|
|
return await bot.safe_send_chunked(
|
|
primary,
|
|
"hello world",
|
|
fallback_channels=[fallback],
|
|
purpose="test notification",
|
|
)
|
|
|
|
assert asyncio.run(run()) is None
|
|
|
|
|
|
def test_build_retry_notice_mentions_reason_and_path():
|
|
message = bot.build_retry_notice(
|
|
file_path=str(Path("recordings") / "1" / "session" / "meeting.wav"),
|
|
reason="OpenRouter 502",
|
|
)
|
|
|
|
assert "OpenRouter 502" in message
|
|
assert "meeting.wav" in message
|
|
assert "`/retry`" in message
|
|
|
|
|
|
def test_meeting_recorder_wraps_wave_sink_with_silence_generator(tmp_path, monkeypatch):
|
|
created = {}
|
|
|
|
class FakeWaveSink:
|
|
def __init__(self, output_path):
|
|
self.output_path = output_path
|
|
created["wave"] = self
|
|
|
|
class FakeSilenceGeneratorSink:
|
|
def __init__(self, destination):
|
|
self.destination = destination
|
|
created["silence"] = self
|
|
|
|
class FakeVoiceClient:
|
|
def __init__(self):
|
|
self.listened_sink = None
|
|
|
|
def is_listening(self):
|
|
return False
|
|
|
|
def listen(self, sink, after):
|
|
self.listened_sink = sink
|
|
created["after"] = after
|
|
|
|
monkeypatch.setattr(voice.voice_recv, "WaveSink", FakeWaveSink)
|
|
monkeypatch.setattr(voice.voice_recv, "SilenceGeneratorSink", FakeSilenceGeneratorSink)
|
|
|
|
vc = FakeVoiceClient()
|
|
output_path = tmp_path / "meeting.wav"
|
|
recorder = voice.MeetingRecorder(vc, str(output_path))
|
|
|
|
asyncio.run(recorder.start(lambda error: None))
|
|
|
|
assert created["wave"].output_path == str(output_path)
|
|
assert created["silence"].destination is created["wave"]
|
|
assert vc.listened_sink is created["silence"]
|
|
assert recorder.recording is True
|
|
|
|
|
|
def test_wait_for_file_ready_rejects_stable_but_invalid_wav(tmp_path):
|
|
broken = tmp_path / "meeting.wav"
|
|
broken.write_bytes(b"not a real wav file")
|
|
|
|
ready = asyncio.run(bot.wait_for_file_ready(str(broken), attempts=3, delay=0.001))
|
|
|
|
assert ready is False
|
|
|
|
|
|
def test_wait_for_file_ready_accepts_valid_wav(tmp_path):
|
|
valid = tmp_path / "meeting.wav"
|
|
with wave.open(str(valid), "wb") as wav_file:
|
|
wav_file.setnchannels(1)
|
|
wav_file.setsampwidth(2)
|
|
wav_file.setframerate(16000)
|
|
wav_file.writeframes(b"\x00\x00" * 160)
|
|
|
|
ready = asyncio.run(bot.wait_for_file_ready(str(valid), attempts=3, delay=0.001))
|
|
|
|
assert ready is True
|