discussion_helper.testing module

Fake parts, so the loop can be tested without a room.

Module summary

The behaviour worth testing in this package is the behaviour that is hardest to observe: what happens when somebody interrupts, whether the queued audio is thrown away, whether the floor ends up back where it belongs. None of that is about a microphone, a model or a sound card — and a test that needs all three is a test that runs on one machine, once, by hand.

So the three parts have fakes, and they ship in the package rather than in the test directory. Somebody writing their own responder or their own voice wants to drop it into a loop whose other two parts are known-good, and they should not have to copy them out of a test file to do it.

Each fake is deliberately observable: SilentVoice records what it was asked to say and what it was cut off mid-way through, because “did the interruption actually stop the audio” is exactly the assertion that matters and exactly the one a real voice makes impossible.

Author

Warith HARCHAOUI — https://linkedin.com/in/warith-harchaoui

class discussion_helper.testing.EchoResponder(sentences=1, delay_s=0.0)

Bases: object

Answers with what it was told, one sentence per call.

The simplest responder that still exercises the loop: it proves the history reached the model, because what comes back is what went in.

Parameters:

• sentences (int)

• delay_s (float)

async aclose()

Return type:

None

async reply(messages)

Echo the last user message, as many sentences as configured.

Parameters:

messages (list[dict[str, str]])

Return type:

AsyncIterator[str]

class discussion_helper.testing.ScriptedListener(turns, *, interrupt_after_s=None, pace_s=0.0)

Bases: object

Yields the turns it was given, then stops.

Parameters:

• turns (sequence of str) – What the user says, in order.

• interrupt_after_s (float or None) – When set, speech_started() resolves this many seconds after it is awaited — which is how a barge-in is staged in a test. None never resolves, which is a user who waits politely.

• pace_s (float) – A pause before each turn, so a timing test has something to measure.

async aclose()

Return type:

None

async speech_started()

Resolve when the staged interruption happens, or never.

Return type:

None

async utterances()

Yield each scripted turn as a finished utterance.

Return type:

AsyncIterator[Utterance]

class discussion_helper.testing.SilentVoice(seconds_per_sentence=0.05, first_audio_s=0.0)

Bases: object

Plays nothing, and remembers exactly what it was asked to play.

Parameters:

• seconds_per_sentence (float) – How long a sentence “takes”. Real time, so a barge-in staged at 0.1 s lands inside a 1 s sentence the way it would in a room.

• first_audio_s (float)

spoken

Every sentence it was asked for, including the one that was cut.

Type:

list of str

completed

Only the sentences that finished. The difference between the two lists is the assertion that matters: audio that was started and not heard must not be counted as said.

Type:

list of str

stops

How many times it was told to stop.

Type:

int

async aclose()

Return type:

None

async speak(sentence)

“Play” one sentence, returning early if stopped.

Parameters:

sentence (str)

Return type:

SpokenSentence

async stop()

Cut the current sentence and discard what was queued.

Return type:

None