Skip to content

Commit a48b4b2

Browse files
hanhsuanhanhsuan
committed
Remove assertLog to make code simplier
1 parent 4b0a8df commit a48b4b2

1 file changed

Lines changed: 80 additions & 240 deletions

File tree

‎providers/base/tests/test_pipewire_utils.py‎

Lines changed: 80 additions & 240 deletions
Original file line numberDiff line numberDiff line change
@@ -16,9 +16,7 @@
1616

1717
# There are some issue with assertLog for python version lower than 3.10,
1818
# https://github.com/python/cpython/commit/6fdfcec5b11f44f27aae3d53ddeb004150ae1f61
19-
# and make some duplicate code in this script.
20-
# Therefore, please don't add new test cases of assertLog
21-
# and the old one might be removed in the future.
19+
# Therefore, please don't add new test cases of assertLog.
2220

2321
import unittest
2422
import sys
@@ -112,59 +110,41 @@ def test_fail(self):
112110
pt = PipewireTest()
113111

114112
# return UNKNOWN TYPE
115-
if sys.version_info < (3, 10):
116-
wrong_type1 = "videog"
117-
wrong_type2 = "xxxx"
118-
self.assertEqual("UNKNOWN TYPE",
119-
pt.generate_pw_media_class(wrong_type1,
120-
"source"))
121-
self.assertEqual("UNKNOWN TYPE",
122-
pt.generate_pw_media_class(wrong_type2,
123-
"sources"))
124-
return
125-
126-
with self.assertLogs("root", level="INFO") as cm:
127-
wrong_type1 = "videog"
128-
wrong_type2 = "xxxx"
129-
self.assertEqual("UNKNOWN TYPE",
130-
pt.generate_pw_media_class(wrong_type1,
131-
"source"))
132-
self.assertEqual("UNKNOWN TYPE",
133-
pt.generate_pw_media_class(wrong_type2,
134-
"sources"))
135-
# check log output
136-
self.assertEqual(cm.output,
137-
["INFO:root:Media type:[{}] is unknown"
138-
.format(wrong_type1),
139-
"INFO:root:Media type:[{}] is unknown"
140-
.format(wrong_type2)])
113+
wrong_type1 = "videog"
114+
wrong_type2 = "xxxx"
115+
self.assertEqual("UNKNOWN TYPE",
116+
pt.generate_pw_media_class(wrong_type1,
117+
"source"))
118+
self.assertEqual("UNKNOWN TYPE",
119+
pt.generate_pw_media_class(wrong_type2,
120+
"sources"))
121+
122+
# example code for testing log output in python 3.10 and above
123+
# with self.assertLogs("root", level="INFO") as cm:
124+
# wrong_type1 = "videog"
125+
# wrong_type2 = "xxxx"
126+
# self.assertEqual("UNKNOWN TYPE",
127+
# pt.generate_pw_media_class(wrong_type1,
128+
# "source"))
129+
# self.assertEqual("UNKNOWN TYPE",
130+
# pt.generate_pw_media_class(wrong_type2,
131+
# "sources"))
132+
# # check log output
133+
# self.assertEqual(cm.output,
134+
# ["INFO:root:Media type:[{}] is unknown"
135+
# .format(wrong_type1),
136+
# "INFO:root:Media type:[{}] is unknown"
137+
# .format(wrong_type2)])
141138

142139
# return UNKNOWN CLASS
143-
if sys.version_info < (3, 10):
144-
wrong_class1 = "sourceg"
145-
wrong_class2 = "sinkf"
146-
self.assertEqual("UNKNOWN CLASS",
147-
pt.generate_pw_media_class("video",
148-
wrong_class1))
149-
self.assertEqual("UNKNOWN CLASS",
150-
pt.generate_pw_media_class("video",
151-
wrong_class2))
152-
return
153-
with self.assertLogs("root", level="INFO") as cm:
154-
wrong_class1 = "sourceg"
155-
wrong_class2 = "sinkf"
156-
self.assertEqual("UNKNOWN CLASS",
157-
pt.generate_pw_media_class("video",
158-
wrong_class1))
159-
self.assertEqual("UNKNOWN CLASS",
160-
pt.generate_pw_media_class("video",
161-
wrong_class2))
162-
# check log output
163-
self.assertEqual(cm.output,
164-
["INFO:root:Media class:[{}] is unknown"
165-
.format(wrong_class1),
166-
"INFO:root:Media class:[{}] is unknown"
167-
.format(wrong_class2)])
140+
wrong_class1 = "sourceg"
141+
wrong_class2 = "sinkf"
142+
self.assertEqual("UNKNOWN CLASS",
143+
pt.generate_pw_media_class("video",
144+
wrong_class1))
145+
self.assertEqual("UNKNOWN CLASS",
146+
pt.generate_pw_media_class("video",
147+
wrong_class2))
168148

169149

170150
class DetectDeviceTests(unittest.TestCase):
@@ -242,80 +222,34 @@ class DetectDeviceTests(unittest.TestCase):
242222
def test_succ(self, mock_checkout):
243223
pt = PipewireTest()
244224

245-
if sys.version_info < (3, 10):
246-
# detect audio sink succ
247-
mock_checkout.return_value = self.sink_audio_node
248-
self.assertEqual(PipewireTestError.NO_ERROR,
249-
pt.detect_device("audio", "sink"))
250-
251-
# detect audio source succ
252-
mock_checkout.return_value = self.source_audio_node
253-
self.assertEqual(PipewireTestError.NO_ERROR,
254-
pt.detect_device("audio", "source"))
255-
return
256-
with self.assertLogs("root", level="INFO") as cm:
257-
# detect audio sink succ
258-
mock_checkout.return_value = self.sink_audio_node
259-
self.assertEqual(PipewireTestError.NO_ERROR,
260-
pt.detect_device("audio", "sink"))
261-
262-
# detect audio source succ
263-
mock_checkout.return_value = self.source_audio_node
264-
self.assertEqual(PipewireTestError.NO_ERROR,
265-
pt.detect_device("audio", "source"))
266-
267-
# check log output
268-
self.assertEqual(cm.output,
269-
["INFO:root:device id:[29]"
270-
" media.class:[Audio/Sink]"
271-
" node.name:[Freewheel-Driver]",
272-
"INFO:root:device id:[30]"
273-
" media.class:[Audio/Source]"
274-
" node.name:[Freewheel-Driver]"])
225+
# detect audio sink succ
226+
mock_checkout.return_value = self.sink_audio_node
227+
self.assertEqual(PipewireTestError.NO_ERROR,
228+
pt.detect_device("audio", "sink"))
229+
230+
# detect audio source succ
231+
mock_checkout.return_value = self.source_audio_node
232+
self.assertEqual(PipewireTestError.NO_ERROR,
233+
pt.detect_device("audio", "source"))
275234

276235
@patch("subprocess.check_output")
277236
def test_fail(self, mock_checkout):
278237
pt = PipewireTest()
279238

280-
if sys.version_info < (3, 10):
281-
# detect audio sink fail
282-
mock_checkout.return_value = self.source_audio_node
283-
self.assertEqual(PipewireTestError.NOT_DETECTED,
284-
pt.detect_device("audio", "sink"))
285-
286-
# detect audio source fail
287-
mock_checkout.return_value = self.sink_audio_node
288-
self.assertEqual(PipewireTestError.NOT_DETECTED,
289-
pt.detect_device("audio", "source"))
290-
291-
# wrong media class
292-
mock_checkout.return_value = self.sink_audio_node
293-
self.assertEqual(PipewireTestError.NOT_DETECTED,
294-
pt.detect_device("xxx", "ooo"))
295-
return
296-
with self.assertLogs("root", level="INFO") as cm:
297-
# detect audio sink fail
298-
mock_checkout.return_value = self.source_audio_node
299-
self.assertEqual(PipewireTestError.NOT_DETECTED,
300-
pt.detect_device("audio", "sink"))
301-
302-
# detect audio source fail
303-
mock_checkout.return_value = self.sink_audio_node
304-
self.assertEqual(PipewireTestError.NOT_DETECTED,
305-
pt.detect_device("audio", "source"))
306-
307-
# wrong media class
308-
mock_checkout.return_value = self.sink_audio_node
309-
self.assertEqual(PipewireTestError.NOT_DETECTED,
310-
pt.detect_device("xxx", "ooo"))
311-
312-
# check log output
313-
self.assertEqual(cm.output,
314-
["INFO:root:"
315-
"media.class:[Audio/Sink] couldn't find",
316-
"INFO:root:"
317-
"media.class:[Audio/Source] couldn't find",
318-
"INFO:root:Media type:[xxx] is unknown"])
239+
# detect audio sink fail
240+
mock_checkout.return_value = self.source_audio_node
241+
self.assertEqual(PipewireTestError.NOT_DETECTED,
242+
pt.detect_device("audio", "sink"))
243+
244+
# detect audio source fail
245+
mock_checkout.return_value = self.sink_audio_node
246+
self.assertEqual(PipewireTestError.NOT_DETECTED,
247+
pt.detect_device("audio", "source"))
248+
249+
# wrong media class
250+
mock_checkout.return_value = self.sink_audio_node
251+
self.assertEqual(PipewireTestError.NOT_DETECTED,
252+
pt.detect_device("xxx", "ooo"))
319253

320254

321255
class SelectDeviceTests(unittest.TestCase):
@@ -360,20 +294,9 @@ class SelectDeviceTests(unittest.TestCase):
360294
def test_wrong_pw_dump(self, mock_checkout):
361295
pt = PipewireTest()
362296

363-
if sys.version_info < (3, 10):
364-
mock_checkout.return_value = "xxxx"
365-
self.assertEqual(PipewireTestError.NO_AVAILABLE_PORT,
366-
pt.select_device("video", "sink", "hdmi"))
367-
return
368-
with self.assertLogs("root", level="INFO") as cm:
369-
mock_checkout.return_value = "xxxx"
370-
self.assertEqual(PipewireTestError.NO_AVAILABLE_PORT,
371-
pt.select_device("video", "sink", "hdmi"))
372-
373-
# check log output
374-
self.assertEqual(cm.output,
375-
["ERROR:root:pw-dump Node failed !!!",
376-
"ERROR:root:No available Video/Sink found"])
297+
mock_checkout.return_value = "xxxx"
298+
self.assertEqual(PipewireTestError.NO_AVAILABLE_PORT,
299+
pt.select_device("video", "sink", "hdmi"))
377300

378301
@patch("subprocess.check_output")
379302
def test_no_available_node(self, mock_checkout):
@@ -385,19 +308,9 @@ def test_no_available_node(self, mock_checkout):
385308
self.assertEqual(PipewireTestError.NO_AVAILABLE_PORT,
386309
pt.select_device("video", "xxx", "hdmi"))
387310

388-
if sys.version_info < (3, 10):
389-
mock_checkout.return_value = self.sink_audio_node
390-
self.assertEqual(PipewireTestError.NO_AVAILABLE_PORT,
391-
pt.select_device("video", "sink", "hdmi"))
392-
return
393-
with self.assertLogs("root", level="INFO") as cm:
394-
mock_checkout.return_value = self.sink_audio_node
395-
self.assertEqual(PipewireTestError.NO_AVAILABLE_PORT,
396-
pt.select_device("video", "sink", "hdmi"))
397-
398-
# check log output
399-
self.assertEqual(cm.output,
400-
["ERROR:root:No available Video/Sink found"])
311+
mock_checkout.return_value = self.sink_audio_node
312+
self.assertEqual(PipewireTestError.NO_AVAILABLE_PORT,
313+
pt.select_device("video", "sink", "hdmi"))
401314

402315
@patch("builtins.input")
403316
@patch("subprocess.check_output")
@@ -465,48 +378,19 @@ class GstPipeLineTests(unittest.TestCase):
465378
def test_wrong_gst_pipeline_device(self, mock_checkout):
466379
pt = PipewireTest()
467380

468-
if sys.version_info < (3, 10):
469-
mock_checkout.return_value = self.device
470-
self.assertEqual(PipewireTestError.NO_SPECIFIC_DEVICE,
471-
pt.gst_pipeline("pipe", 10, "qoo"))
472-
return
473-
with self.assertLogs("root", level="INFO") as cm:
474-
mock_checkout.return_value = self.device
475-
self.assertEqual(PipewireTestError.NO_SPECIFIC_DEVICE,
476-
pt.gst_pipeline("pipe", 10, "qoo"))
477-
478-
# check log output
479-
self.assertEqual(cm.output,
480-
["ERROR:root:No abailable output device for qoo"])
381+
mock_checkout.return_value = self.device
382+
self.assertEqual(PipewireTestError.NO_SPECIFIC_DEVICE,
383+
pt.gst_pipeline("pipe", 10, "qoo"))
481384

482385
@patch("subprocess.check_output")
483386
def test_gst_pipeline(self, mock_checkout):
484387
pt = PipewireTest()
485388

486-
if sys.version_info < (3, 10):
487-
mock_checkout.return_value = self.device
488-
self.assertEqual(PipewireTestError.NO_ERROR,
489-
pt.gst_pipeline("audiotestsrc wave=sine freq=512"
490-
" ! audioconvert ! audioresample"
491-
" ! autoaudiosink", 2, "hdmi"))
492-
return
493-
with self.assertLogs("root", level="INFO") as cm:
494-
mock_checkout.return_value = self.device
495-
self.assertEqual(PipewireTestError.NO_ERROR,
496-
pt.gst_pipeline("audiotestsrc wave=sine freq=512"
497-
" ! audioconvert ! audioresample"
498-
" ! autoaudiosink", 2, "hdmi"))
499-
500-
# check log output
501-
self.assertEqual(cm.output,
502-
["INFO:root:" + "[ Audio sink ]".center(80, '='),
503-
"INFO:root:Device: [hdmi demo output] "
504-
"availavle: [yes]",
505-
"INFO:root:Attempting to initialize Gstreamer "
506-
"pipeline: audiotestsrc wave=sine freq=512 ! "
507-
"audioconvert ! audioresample ! autoaudiosink",
508-
"INFO:root:Pipeline initialized, now starting "
509-
"playback."])
389+
mock_checkout.return_value = self.device
390+
self.assertEqual(PipewireTestError.NO_ERROR,
391+
pt.gst_pipeline("audiotestsrc wave=sine freq=512"
392+
" ! audioconvert ! audioresample"
393+
" ! autoaudiosink", 2, "hdmi"))
510394

511395

512396
class MonitorActivePortTests(unittest.TestCase):
@@ -599,47 +483,17 @@ class MonitorActivePortTests(unittest.TestCase):
599483
def test_couldnt_detect_change(self, mock_checkout):
600484
pt = PipewireTest()
601485

602-
if sys.version_info < (3, 10):
603-
mock_checkout.return_value = self.before_device
604-
self.assertEqual(PipewireTestError.NO_CHANGE_DETECTED,
605-
pt.monitor_active_port_change(2, "sink"))
606-
return
607-
with self.assertLogs("root", level="INFO") as cm:
608-
mock_checkout.return_value = self.before_device
609-
self.assertEqual(PipewireTestError.NO_CHANGE_DETECTED,
610-
pt.monitor_active_port_change(2, "sink"))
611-
612-
# check log output
613-
self.assertEqual(cm.output,
614-
["INFO:root:Starting with config: "
615-
"{('sink #29', 'hdmi_demo_output', 'yes')}",
616-
"INFO:root:"
617-
"You have 2 seconds to plug the item in",
618-
"INFO:root:Couldn't detect active port change!"])
486+
mock_checkout.return_value = self.before_device
487+
self.assertEqual(PipewireTestError.NO_CHANGE_DETECTED,
488+
pt.monitor_active_port_change(2, "sink"))
619489

620490
@patch("subprocess.check_output")
621491
def test_could_detect_change(self, mock_checkout):
622492
pt = PipewireTest()
623493

624-
if sys.version_info < (3, 10):
625-
mock_checkout.side_effect = [self.before_device, self.after_device]
626-
self.assertEqual(PipewireTestError.NO_ERROR,
627-
pt.monitor_active_port_change(2, "sink"))
628-
return
629-
with self.assertLogs("root", level="INFO") as cm:
630-
mock_checkout.side_effect = [self.before_device, self.after_device]
631-
self.assertEqual(PipewireTestError.NO_ERROR,
632-
pt.monitor_active_port_change(2, "sink"))
633-
634-
# check log output
635-
self.assertEqual(cm.output,
636-
["INFO:root:Starting with config: "
637-
"{('sink #29', 'hdmi_demo_output', 'yes')}",
638-
"INFO:root:"
639-
"You have 2 seconds to plug the item in",
640-
"INFO:root:Now using config: "
641-
"{('sink #29', 'hdmi_after_output', 'yes')}",
642-
"INFO:root:It seems to work!"])
494+
mock_checkout.side_effect = [self.before_device, self.after_device]
495+
self.assertEqual(PipewireTestError.NO_ERROR,
496+
pt.monitor_active_port_change(2, "sink"))
643497

644498

645499
class GoThroughPortTests(unittest.TestCase):
@@ -691,23 +545,9 @@ class GoThroughPortTests(unittest.TestCase):
691545
def test_though(self, mock_checkout, mock_input):
692546
pt = PipewireTest()
693547

694-
if sys.version_info < (3, 10):
695-
mock_checkout.return_value = self.device
696-
mock_input.side_effect = ["yes", "yes"]
697-
self.assertEqual(None, pt.go_through_ports("echo test", "sink"))
698-
return
699-
with self.assertLogs("root", level="INFO") as cm:
700-
mock_checkout.return_value = self.device
701-
mock_input.side_effect = ["yes", "yes"]
702-
self.assertEqual(None, pt.go_through_ports("echo test", "sink"))
703-
704-
# check log output
705-
self.assertEqual(cm.output,
706-
[("INFO:root:Please select [hdmi demo output] "
707-
"for testing (if selected, please enter 'yes')"),
708-
"INFO:root:test",
709-
("INFO:root:"
710-
"Is working ? please enter 'yes' to leave")])
548+
mock_checkout.return_value = self.device
549+
mock_input.side_effect = ["yes", "yes"]
550+
self.assertEqual(None, pt.go_through_ports("echo test", "sink"))
711551

712552

713553
class ArgsParsingTests(unittest.TestCase):

0 commit comments

Comments
 (0)