Fixed server.py unittests

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)]'

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()

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