Let a recording be held while something else happens

A phone call in the middle of a dictation left two choices: send what
was said so far off to be transcribed, or throw it away. Both end the
sentence you were in the middle of. Now the recording can be held: the
microphone stays ours, what was said before the pause stays in the
buffer, and what is said during it is dropped.

The capture program keeps running and keeps handing blocks over, which
are read and thrown away rather than left in the pipe. Stopping it
instead would mean asking the sound server for the device again on the
way back, and that is the one moment another application can take it: a
recording would be lost to the phone call it was paused for.

The clock stops with it. Paused time is time the recording does not
have, so the indicator and the length limit both go by what was
actually captured, and a five minute limit is not spent waiting.

The indicator says so as well, since a pulsing dot, moving bars and a
counting clock otherwise all say the words are still going in: the
ribbon freezes where the pause found it and turns amber behind two
bars. The tray menu holds and resumes it, `dikte pause` does, and so
does a global shortcut, which starts empty because holding a recording
is not something a keyboard has a habit for.

Dictation and a command to the agent both, whichever is recording. A
meeting is left out: it writes to a file as it goes and keeps two
streams aligned itself, and neither of those wants a hole in it.

tests/test_audio.py

@@ -280,6 +280,26 @@ class _StalledStream:
         self._released.set()
 
 
+class _HeldStream:
+    """A capture that is paused and taken up again partway through, the way a
+    key press lands in the middle of a recording rather than between two."""
+
+    def __init__(self, data, recorder, pause_at, resume_at=None):
+        self._data = io.BytesIO(data)
+        self._recorder = recorder
+        self._pause_at = pause_at
+        self._resume_at = resume_at
+        self.reads = 0
+
+    def read(self, size):
+        if self.reads == self._pause_at:
+            self._recorder.pause()
+        elif self.reads == self._resume_at:
+            self._recorder.pause(False)
+        self.reads += 1
+        return self._data.read(size)
+
+
 class RecordingCommand(OnLinux, DikteTest):
     """Which program captures the microphone, and how it is asked to."""
 
@@ -499,6 +519,51 @@ class RecorderChain(OnLinux, DikteTest):
         self.assertEqual(len(failures), 1)
         self.assertIn("0.3", failures[0])
 
+    def held(self, data, pause_at, resume_at=None):
+        """Record `data` with the recorder paused for part of it."""
+        recorder = audio.Recorder()
+        results = []
+        failures = []
+        recorder.stopped.connect(lambda *args: results.append(args))
+        recorder.failed.connect(failures.append)
+        proc = FakeProcess(data)
+        proc.stdout = _HeldStream(data, recorder, pause_at, resume_at)
+        with only_these_tools("pw-record"), \
+                mock.patch.object(subprocess, "Popen", return_value=proc):
+            recorder.start()
+            recorder._thread.join(timeout=5)
+            # Nothing has ended the capture: a pause holds the microphone.
+            self.assertEqual(proc.signals, [])
+            recorder.stop()
+        return results, failures
+
+    def test_what_is_said_while_it_is_held_is_not_in_the_recording(self):
+        """The phone call in the middle of a dictation is the whole feature: it
+        must not reach the transcript, and the two halves must meet."""
+        results, failures = self.held(tone(2.0), pause_at=8, resume_at=16)
+        self.assertEqual(failures, [])
+        path, duration, _ = results[0]
+        self.addCleanup(os.unlink, path)
+        dropped = 8 * audio.CHUNK_FRAMES
+        self.assertAlmostEqual(duration, (2 * audio.RATE - dropped) / audio.RATE,
+                               places=3)
+
+    def test_a_recording_held_all_the_way_through_captured_nothing(self):
+        results, failures = self.held(tone(2.0), pause_at=0)
+        self.assertEqual(results, [])
+        self.assertIn("0.3", failures[0])
+
+    def test_a_pause_does_not_outlive_the_recording_it_was_asked_for(self):
+        recorder = audio.Recorder()
+        recorder.pause()
+        proc = FakeProcess(tone(0.5))
+        with only_these_tools("pw-record"), \
+                mock.patch.object(subprocess, "Popen", return_value=proc):
+            recorder.start()
+            self.assertFalse(recorder.paused)
+            recorder._thread.join(timeout=5)
+            recorder.cancel()
+
     def test_a_recorder_that_could_not_start(self):
         recorder = audio.Recorder()
         failures = []