Merge pull request #107 from hyades/api_record_file_format

Fix tests for record file option parsing

.gitignore

@@ -71,3 +71,4 @@ test-driver
 *.dat*
 *.avi
 *.mkv
+*.log

python-api/gstswitch/server.py

@@ -40,7 +40,7 @@ def __init__(
             video_port=3000,
             audio_port=4000,
             control_port=5000,
-            record_file='record.data'):
+            record_file=False):
 
         super(Server, self).__init__()
 
@@ -159,19 +159,24 @@ def record_file(self):
     @record_file.setter
     def record_file(self, record_file):
         """Set Record File
-        :raises ValueError: Record File cannot be left blank
+        :raises ValueError: Non-string file format
         :raises ValueError: Record File cannot have forward slashes
         """
-        if not record_file:
-            raise ValueError("Record File '{0}' cannot be blank"
-                             .format(record_file))
+        if record_file is False:
+            self._record_file = False
+        elif record_file is True:
+            self._record_file = True
         else:
-            rec = str(record_file)
-            if rec.find('/') < 0:
-                self._record_file = rec
-            else:
+            if not record_file:
                 raise ValueError("Record File: '{0}' "
-                                 "cannot have forward slashes".format(rec))
+                                 "Non-string file format".format(record_file))
+            else:
+                rec = str(record_file)
+                if rec.find('/') < 0:
+                    self._record_file = rec
+                else:
+                    raise ValueError("Record File: '{0}' "
+                                     "cannot have forward slashes".format(rec))
 
     def run(self, gst_option=''):
         """Launch the server process
@@ -196,26 +201,29 @@ def run(self, gst_option=''):
     def _run_process(self):
         """Non-public method: Runs the gst-switch-srv process
         """
-        cmd = ''
+        cmd = ['']
         if not self.path:
             srv_location = spawn.find_executable('gst-switch-srv')
             if srv_location:
-                cmd = srv_location
+                cmd[0] = srv_location
             else:
                 raise PathError("Cannot find gst-switch-srv in $PATH.\
                     Please specify the path.")
         else:
-            cmd += os.path.join(self.path, 'gst-switch-srv')
-        cmd += """  {0} \
-                    --video-input-port={1} \
-                    --audio-input-port={2} \
-                    --control-port={3} \
-                    --record={4} """ .format(self.gst_option_string,
-                                             self.video_port,
-                                             self.audio_port,
-                                             self.control_port,
-                                             self.record_file)
-        cmd = " ".join(cmd.split())
+            cmd[0] += os.path.join(self.path, 'gst-switch-srv')
+        if self.gst_option_string:
+            cmd += [self.gst_option_string]
+        cmd.append("--video-input-port={0}".format(self.video_port))
+        cmd.append("--audio-input-port={0}".format(self.audio_port))
+        cmd.append("--control-port={0}".format(self.control_port))
+        if self.record_file is False:
+            pass
+        elif self.record_file is True:
+            cmd.append("-r")
+        else:
+            if self.record_file is not False:
+                cmd.append("--record={0}".format(self.record_file))
+
         proc = self._start_process(cmd)
         return proc
 
@@ -229,7 +237,7 @@ def _start_process(self, cmd):
         try:
             with open('server.log', 'w') as tempf:
                 process = subprocess.Popen(
-                    cmd.split(),
+                    cmd,
                     stdout=tempf,
                     stderr=tempf,
                     bufsize=-1,

python-api/tests/integrationtests/test_controller.py

@@ -366,7 +366,7 @@ def new_record(self):
     def test_new_record(self):
         """Test new_record"""
         for _ in range(self.NUM):
-            serv = Server(path=PATH, record_file="test.data")
+            serv = Server(path=PATH, record_file="test-%Y.data")
             try:
                 serv.run()
 
@@ -375,18 +375,14 @@ def test_new_record(self):
                 sources.new_test_video()
 
                 curr_time = datetime.datetime.now()
-                alt_curr_time = curr_time + datetime.timedelta(0, 1)
-                time_str = curr_time.strftime('%Y-%m-%d %H%M%S')
-                alt_time_str = alt_curr_time.strftime('%Y-%m-%d %H%M%S')
-                test_filename = "test {0}.data".format(time_str)
-                alt_test_filename = "test {0}.data".format(alt_time_str)
+                time_str = curr_time.strftime('%Y')
+                test_filename = "test-{0}.data".format(time_str)
 
                 res = self.new_record()
                 print res
                 sources.terminate_video()
                 serv.terminate(1)
-                assert ((os.path.exists(test_filename)) or
-                        (os.path.exists(alt_test_filename))) is True
+                assert os.path.exists(test_filename) is True
             finally:
                 if serv.proc:
                     poll = serv.proc.poll()

python-api/tests/unittests/test_server_unit.py

@@ -27,9 +27,9 @@ def mock_method(arg):
         path = '/usr/'
         serv = Server(path=path)
         serv._start_process = mock_method
-        assert serv._run_process().split() == "/usr/gst-switch-srv \
+        assert serv._run_process() == "/usr/gst-switch-srv \
 --video-input-port=3000 --audio-input-port=4000 \
---control-port=5000 --record=record.data".split()
+--control-port=5000".split()
 
     def test_path_provided_no_slash(self):
         """Test if a path is provided"""
@@ -39,9 +39,9 @@ def mock_method(arg):
         path = '/usr'
         serv = Server(path=path)
         serv._start_process = mock_method
-        assert serv._run_process().split() == "/usr/gst-switch-srv \
+        assert serv._run_process() == "/usr/gst-switch-srv \
 --video-input-port=3000 --audio-input-port=4000 \
---control-port=5000 --record=record.data".split()
+--control-port=5000".split()
 
     def test_path_empty(self, monkeypatch):
         """Test if null path is given"""
@@ -58,12 +58,13 @@ def mockreturn(path):
         for path in paths:
             serv = Server(path=path)
             serv._start_process = mock_method
-            assert serv._run_process().split() == "/usr/gst-switch-srv \
+            assert serv._run_process() == "/usr/gst-switch-srv \
 --video-input-port=3000 --audio-input-port=4000 \
---control-port=5000 --record=record.data".split()
+--control-port=5000".split()
 
 
 class TestVideoPort(object):
+
     """Test for video_port parameter"""
     # Video Port Tests
 
@@ -151,8 +152,69 @@ class TestRecordFile(object):
     """Test the record_file parameter"""
     # Record File
 
-    def test_record_file_blank(self):
-        """Test when the record_file is null"""
+    def test_record_file_false(self):
+        """Test if record file is False"""
+        def mock_method(arg):
+            """Mocking _start_process"""
+            return arg
+        path = '/usr'
+        serv = Server(path=path, record_file=False)
+        serv._start_process = mock_method
+        assert serv._run_process() == "/usr/gst-switch-srv \
+--video-input-port=3000 --audio-input-port=4000 \
+--control-port=5000".split()
+
+    def test_record_file_true(self):
+        """Test if record file is True"""
+        def mock_method(arg):
+            """Mocking _start_process"""
+            return arg
+        path = '/usr'
+        serv = Server(path=path, record_file=True)
+        serv._start_process = mock_method
+        assert serv._run_process() == "/usr/gst-switch-srv \
+--video-input-port=3000 --audio-input-port=4000 \
+--control-port=5000 -r".split()
+
+    def test_record_file_valid(self):
+        """Test if record file is valid"""
+        def mock_method(arg):
+            """Mocking _start_process"""
+            return arg
+        path = '/usr'
+        serv = Server(path=path, record_file="record.data")
+        serv._start_process = mock_method
+        assert serv._run_process() == "/usr/gst-switch-srv \
+--video-input-port=3000 --audio-input-port=4000 \
+--control-port=5000 --record=record.data".split()
+
+    def test_record_file_valid_date(self):
+        """Test if record file is valid"""
+        def mock_method(arg):
+            """Mocking _start_process"""
+            return arg
+        path = '/usr'
+        serv = Server(path=path, record_file="record_%Y.data")
+        serv._start_process = mock_method
+        assert serv._run_process() == "/usr/gst-switch-srv \
+--video-input-port=3000 --audio-input-port=4000 \
+--control-port=5000 \
+--record=record_%Y.data".split()
+
+    def test_record_file_valid_space(self):
+        """Test if record file is valid and has a space"""
+        def mock_method(arg):
+            """Mocking _start_process"""
+            return arg
+        path = '/usr'
+        serv = Server(path=path, record_file='record 1.data')
+        serv._start_process = mock_method
+        assert serv._run_process() == "/usr/gst-switch-srv \
+--video-input-port=3000 --audio-input-port=4000 \
+--control-port=5000".split() + ["--record=record 1.data"]
+
+    def test_record_file_invalid(self):
+        """Test when the record_file is invalid"""
         files = ['', None, [], {}]
         for record_file in files:
             with pytest.raises(ValueError):