From 5cd28491526c4d48c4d6318c3a5922d95dd4d459 Mon Sep 17 00:00:00 2001 From: hyades Date: Fri, 2 Aug 2013 02:13:22 +0530 Subject: [PATCH] Fixed server.py unittests --- python-api/gstswitch/test_controller.py | 73 +++++++++++++++++++-- python-api/gstswitch/test_server.py | 86 ++++++++++++++++++------- python-api/test.py | 3 +- 3 files changed, 135 insertions(+), 27 deletions(-) diff --git a/python-api/gstswitch/test_controller.py b/python-api/gstswitch/test_controller.py index 60ad72c..1b73d52 100644 --- a/python-api/gstswitch/test_controller.py +++ b/python-api/gstswitch/test_controller.py @@ -94,8 +94,26 @@ def __init__(self, mode): self.mode = mode def get_compose_port(self): - if self.mode is True: - return GLib.Variant('(i)', (0,)) + if self.mode is False: + return GLib.Variant('(i)', (3001,)) + else: + return (0,) + + def get_encode_port(self): + if self.mode is False: + return GLib.Variant('(i)', (3002,)) + else: + return (0,) + + def get_audio_port(self): + if self.mode is False: + return GLib.Variant('(i)', (4000,)) + else: + return (0,) + + def get_preview_ports(self): + if self.mode is False: + return GLib.Variant('(s)', ('[(3002, 1, 7), (3003, 1, 8)]',)) else: return (0,) @@ -104,6 +122,53 @@ class TestGetComposePort(object): def test_unpack(self): controller = Controller(address='unix:abstract=abcdefghijk') - controller.connection = MockConnection(False) - with pytest.raises(AttributeError): + controller.connection = MockConnection(True) + with pytest.raises(ConnectionError): controller.get_compose_port() + + def test_normal_unpack(self): + controller = Controller(address='unix:abstract=abcdef') + controller.connection = MockConnection(False) + assert controller.get_compose_port() == 3001 + + +class TestGetEncodePort(object): + + def test_unpack(self): + controller = Controller(address='unix:abstract=abcdefghijk') + controller.connection = MockConnection(True) + with pytest.raises(ConnectionError): + controller.get_encode_port() + + def test_normal_unpack(self): + controller = Controller(address='unix:abstract=abcdef') + controller.connection = MockConnection(False) + assert controller.get_encode_port() == 3002 + + +class TestGetAudioPort(object): + + def test_unpack(self): + controller = Controller(address='unix:abstract=abcdefghijk') + controller.connection = MockConnection(True) + with pytest.raises(ConnectionError): + controller.get_audio_port() + + def test_normal_unpack(self): + controller = Controller(address='unix:abstract=abcdef') + controller.connection = MockConnection(False) + assert controller.get_audio_port() == 4000 + + +# class TestGetPreviewPorts(object): + +# def test_unpack(self): +# controller = Controller(address='unix:abstract=abcdefghijk') +# controller.connection = MockConnection(True) +# with pytest.raises(ConnectionError): +# controller.get_preview_ports() + +# def test_normal_unpack(self): +# controller = Controller(address='unix:abstract=abcdef') +# controller.connection = MockConnection(False) +# assert controller.get_preview_ports() == '[(3002, 1, 7), (3003, 1, 8)]' diff --git a/python-api/gstswitch/test_server.py b/python-api/gstswitch/test_server.py index b186f9c..d564237 100644 --- a/python-api/gstswitch/test_server.py +++ b/python-api/gstswitch/test_server.py @@ -1,7 +1,9 @@ from server import Server import pytest from exception import * +import os import subprocess +from mock import Mock class TestPath(object): @@ -107,7 +109,7 @@ def test_record_file_slashes(self): Server(path=path, record_file=file) -class TestProcess(object): +class TestKillTerminate(object): # OS Errors def test_terminate_fail(self): class fake_proc(object): @@ -131,45 +133,85 @@ def test_kill_fail(self): with pytest.raises(ServerProcessError): s.kill() - def test_start_process(self): + def test_no_process_kill(self): path = '/home/hyades/gst/master/gstreamer/tools/' s = Server(path=path) with pytest.raises(ServerProcessError): - s._start_process('/usr/') + s.kill() - def test_no_process_kill(self): + def test_no_process_terminate(self): path = '/home/hyades/gst/master/gstreamer/tools/' s = Server(path=path) with pytest.raises(ServerProcessError): - s.kill() + s.terminate() -class TestNormal(object): - # Normal Functioning Tests - def test_normal_terminate(self): - path = '/home/hyades/gst/master/gstreamer/tools/' - s = Server(path=path) +class TestRun(object): + + def test_run(self): + s = Server(path='abc') + s._run_process = Mock(return_value=MockProcess()) s.run() + assert s.pid == 1 assert s.proc is not None + + def test_run_process(self): + s = Server(path='abc') + s._start_process = Mock(return_value=MockProcess()) + s.gst_option_string = '' + ret = s._run_process() + assert ret is not None + + def test_start_process_error(self, monkeypatch): + s = Server(path='abc') + monkeypatch.setattr(subprocess, 'Popen', Mock(side_effect=OSError)) + with pytest.raises(ServerProcessError): + s._start_process('cmd') + + def test_start_process_normal(self, monkeypatch): + s = Server(path='abc') + monkeypatch.setattr(subprocess, 'Popen', Mock(return_value=MockProcess())) + s._start_process('cmd') + + +class MockProcess(object): + def __init__(self, mode=True): + self.mode = mode + self.pid = 1 + + def terminate(self): + if self.mode == True: + pass + if self.mode == False: + raise OSError('Testing terminate') + + +class TestNormal(object): + # Normal Functioning Tests + + def test_normal_terminate(self): + s = Server(path='abc') + s.proc = MockProcess(True) s.terminate() assert s.proc is None - def test_normal_kill(self): - path = '/home/hyades/gst/master/gstreamer/tools/' - s = Server(path=path) - s.run() - assert s.proc is not None - s.kill() + def test_normal_kill(self, monkeypatch): + s = Server(path='abc') + s.proc = Mock() + monkeypatch.setattr(os, 'kill', Mock()) + res = s.kill() + assert res == True assert s.proc is None def test_terminate(self): - path = '/home/hyades/gst/master/gstreamer/tools/' - s = Server(path=path) + s = Server(path='abc') + s.proc = MockProcess(False) with pytest.raises(ServerProcessError): s.terminate() - def test_kill(self): - path = '/home/hyades/gst/master/gstreamer/tools/' - s = Server(path=path) + def test_kill(self, monkeypatch): + s = Server(path='abc') + s.proc = Mock() + monkeypatch.setattr(os, 'kill', Mock(side_effect=OSError)) with pytest.raises(ServerProcessError): - s.terminate() + s.kill() diff --git a/python-api/test.py b/python-api/test.py index 38be562..43a6884 100755 --- a/python-api/test.py +++ b/python-api/test.py @@ -10,7 +10,8 @@ import time # all executables (gst-launch-1.0, gst-switch-srv, gst-switch-ui, gst-switch-cap) at this path -path = '/home/hyades/gst/master/gstreamer/tools/' +# path = '/home/hyades/gst/master/gstreamer/tools/' +path = '/usr/local/bin/' s = Server(path) try: s.run() # launches the server default parameters