1414
1515import subprocess
1616from unittest import mock
17- import pytest
17+
1818import numpy as np
19+ import pytest
20+
1921from nemo .collections .asr .parts .preprocessing .perturb import TranscodePerturbation
2022from nemo .collections .asr .parts .preprocessing .segment import AudioSegment
2123
24+
2225class TestTranscodePerturbationSubprocess :
2326 @pytest .mark .unit
2427 def test_transcode_perturbation_no_shell_true (self ):
@@ -27,29 +30,28 @@ def test_transcode_perturbation_no_shell_true(self):
2730 This uses source inspection/mocking to ensure security compliance.
2831 """
2932 # Mock subprocess.Popen and subprocess.run
30- with mock .patch ("subprocess.Popen" ) as mock_popen , \
31- mock .patch ("subprocess.run" ) as mock_run :
32-
33+ with mock .patch ("subprocess.Popen" ) as mock_popen , mock .patch ("subprocess.run" ) as mock_run :
34+
3335 # Setup mock for Popen context manager
3436 mock_process = mock .Mock ()
3537 mock_process .stdout = mock .Mock ()
3638 mock_popen .return_value .__enter__ .return_value = mock_process
37-
39+
3840 # Setup dummy audio data
3941 tp = TranscodePerturbation (codecs = ["amr-nb" , "ogg" , "g711" ])
4042 data = mock .Mock (spec = AudioSegment )
4143 data ._samples = np .zeros ((16000 ,), dtype = np .float32 )
4244 data .sample_rate = 16000
43-
45+
4446 # Run perturbation multiple times to cover different codecs
45- # We can't easily force a specific codec due to random choice,
47+ # We can't easily force a specific codec due to random choice,
4648 # so we run enough times to likely hit them all or checking the implementation
47-
49+
4850 # Instead of relying on random, let's force the codec choice by mocking random
4951 with mock .patch ("random.randint" , side_effect = [0 , 0 , 1 , 0 , 2 ]):
5052 # 1. Test amr-nb (index 0)
5153 tp .perturb (data )
52-
54+
5355 # Check Popen calls for amr-nb
5456 # Should be called with list args, not string, and shell=False (default)
5557 if mock_popen .called :
@@ -59,55 +61,57 @@ def test_transcode_perturbation_no_shell_true(self):
5961
6062 # Check run calls
6163 if mock_run .called :
62- args , kwargs = mock_run .call_args
63- assert isinstance (args [0 ], list ), "subprocess.run should be called with a list of arguments"
64- assert kwargs .get ("shell" , False ) is False , "subprocess.run should not use shell=True"
65-
64+ args , kwargs = mock_run .call_args
65+ assert isinstance (args [0 ], list ), "subprocess.run should be called with a list of arguments"
66+ assert kwargs .get ("shell" , False ) is False , "subprocess.run should not use shell=True"
67+
6668 mock_popen .reset_mock ()
6769 mock_run .reset_mock ()
6870
6971 # 2. Test ogg (index 1)
7072 tp .perturb (data )
71-
73+
7274 if mock_popen .called :
7375 args , kwargs = mock_popen .call_args
7476 assert isinstance (args [0 ], list ), "subprocess.Popen should be called with a list of arguments"
7577 assert kwargs .get ("shell" , False ) is False , "subprocess.Popen should not use shell=True"
76-
78+
7779 mock_popen .reset_mock ()
7880 mock_run .reset_mock ()
7981
8082 # 3. Test g711 (index 2)
8183 tp .perturb (data )
82-
84+
8385 if mock_run .called :
84- args , kwargs = mock_run .call_args
85- assert isinstance (args [0 ], list ), "subprocess.run should be called with a list of arguments"
86- assert kwargs .get ("shell" , False ) is False , "subprocess.run should not use shell=True"
86+ args , kwargs = mock_run .call_args
87+ assert isinstance (args [0 ], list ), "subprocess.run should be called with a list of arguments"
88+ assert kwargs .get ("shell" , False ) is False , "subprocess.run should not use shell=True"
8789
8890 @pytest .mark .unit
8991 def test_transcode_perturbation_call_structure (self ):
9092 """
9193 Verify the exact structure of the subprocess calls for accuracy.
9294 """
93- with mock .patch ("subprocess.Popen" ) as mock_popen , \
94- mock .patch ("subprocess.run" ) as mock_run , \
95- mock .patch ("soundfile.write" ), \
96- mock .patch ("nemo.collections.asr.parts.preprocessing.segment.AudioSegment.from_file" ):
97-
95+ with (
96+ mock .patch ("subprocess.Popen" ) as mock_popen ,
97+ mock .patch ("subprocess.run" ) as mock_run ,
98+ mock .patch ("soundfile.write" ),
99+ mock .patch ("nemo.collections.asr.parts.preprocessing.segment.AudioSegment.from_file" ),
100+ ):
101+
98102 # Setup mock
99103 mock_process = mock .Mock ()
100104 mock_process .stdout = "PIPE"
101105 mock_popen .return_value .__enter__ .return_value = mock_process
102-
106+
103107 tp = TranscodePerturbation (codecs = ["g711" ])
104108 data = mock .Mock (spec = AudioSegment )
105109 data ._samples = np .zeros ((16000 ,), dtype = np .float32 )
106-
110+
107111 # Test g711 which uses subprocess.run directly
108112 with mock .patch ("random.randint" , return_value = 0 ):
109113 tp .perturb (data )
110-
114+
111115 assert mock_run .called
112116 args , _ = mock_run .call_args
113117 cmd_list = args [0 ]
0 commit comments