aussie/EscapistsReintroduced
1
0
Fork 0
forked from escapists/EscapistsReintroduced
Code Pull Requests Activity

Refresh integrated playtest packages and handle disappearing test audio streams

Browse Source
This commit is contained in:
jakub
2026-09-20 00:33:58 +02:00
parent 5e742691ad
commit e7b4141f69
46 changed files with 4035 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_native_package.py
+58
View File
@@ -203,6 +203,64 @@ class NativePackageTests(unittest.TestCase):
self.assertEqual([call.args[0] for call in run.call_args_list],
[['pactl', 'move-sink-input', '1', 'private-sink'], ['pactl', 'set-sink-input-mute', '1', '1']])
def test_audio_disappearing_move_skips_mute_and_continues_owned_streams(self):
streams = [{'index': 1, 'properties': {'application.id': 'own'}},
{'index': 2, 'properties': {'application.id': 'user-music'}},
{'index': 3, 'properties': {'application.id': 'own'}}]
error = subprocess.CalledProcessError(1, ['pactl', 'move-sink-input', '1', 'private-sink'])
# A newly observed stream is not authority to modify it in this pass.
fresh = streams[1:] + [{'index': 4, 'properties': {'application.id': 'user-music'}}]
with patch.object(n.subprocess, 'check_output', side_effect=[json.dumps(streams), json.dumps(fresh)]) as listing, patch.object(n.subprocess, 'run', side_effect=[error, None, None]) as run:
n.quiet_streams('own', 'private-sink', self.compat)
self.assertEqual(listing.call_count, 2)
self.assertEqual([call.args[0] for call in run.call_args_list], [
['pactl', 'move-sink-input', '1', 'private-sink'],
['pactl', 'move-sink-input', '3', 'private-sink'],
['pactl', 'set-sink-input-mute', '3', '1']])
def test_audio_disappearing_mute_is_confirmed_by_fresh_listing(self):
stream = [{'index': 1, 'properties': {'application.id': 'own'}}]
error = subprocess.CalledProcessError(1, ['pactl', 'set-sink-input-mute', '1', '1'])
with patch.object(n.subprocess, 'check_output', side_effect=[json.dumps(stream), '[]']) as listing, patch.object(n.subprocess, 'run', side_effect=[None, error]) as run:
n.quiet_streams('own', 'private-sink', self.compat)
self.assertEqual(listing.call_count, 2)
self.assertEqual(run.call_count, 2)
def test_audio_operation_failure_remains_when_same_index_is_present(self):
stream = [{'index': 1, 'properties': {'application.id': 'own'}}]
# Even changed ownership at the same index cannot excuse the failure or
# permit another mute/move against that now unrelated stream.
for fresh in (stream, [{'index': 1, 'properties': {'application.id': 'user-music'}}]):
for position in (0, 1):
with self.subTest(fresh=fresh, operation=position):
error = subprocess.CalledProcessError(1, ['pactl', 'failed-operation'])
with patch.object(n.subprocess, 'check_output', side_effect=[json.dumps(stream), json.dumps(fresh)]), patch.object(n.subprocess, 'run', side_effect=[None]*position+[error]) as run:
with self.assertRaises(subprocess.CalledProcessError) as raised:
n.quiet_streams('own', 'private-sink', self.compat)
self.assertIs(raised.exception, error)
self.assertEqual(run.call_count, position+1)
def test_audio_failed_or_malformed_recheck_cannot_prove_absence(self):
stream = [{'index': 1, 'properties': {'application.id': 'own'}}]
reads = [OSError('pactl unavailable'), subprocess.CalledProcessError(1, ['pactl', 'list']),
subprocess.TimeoutExpired(['pactl', 'list'], 5), 'invalid json', '{}', 'null',
'[{}]', '[null]', '[{"index":"1"}]', '[{"index":true}]', '[{"index":-1}]']
for fresh in reads:
with self.subTest(read=fresh):
error = subprocess.CalledProcessError(1, ['pactl', 'move-sink-input'])
with patch.object(n.subprocess, 'check_output', side_effect=[json.dumps(stream), fresh]), patch.object(n.subprocess, 'run', side_effect=error) as run:
with self.assertRaises(subprocess.CalledProcessError) as raised:
n.quiet_streams('own', 'private-sink', self.compat)
self.assertIs(raised.exception, error)
run.assert_called_once()
def test_audio_operation_timeout_is_not_a_disappeared_stream_success(self):
stream = [{'index': 1, 'properties': {'application.id': 'own'}}]
with patch.object(n.subprocess, 'check_output', return_value=json.dumps(stream)) as listing, patch.object(n.subprocess, 'run', side_effect=subprocess.TimeoutExpired(['pactl', 'move-sink-input'], 5)):
with self.assertRaises(subprocess.TimeoutExpired):
n.quiet_streams('own', 'private-sink', self.compat)
listing.assert_called_once()
def test_orchestration_stops_after_first_failed_dependency_without_launch(self):
import uuid
import contextlib