Add flowty CLI level test, Shuffle functional tests around

setup.py

@@ -60,7 +60,7 @@ def cython_extension(
 ]
 
 docs_require = ["sphinx"]
-tests_require = ["pytest"]
+tests_require = ["pytest", "imageio"]
 
 setup(
     name=about["__title__"],

tests/functional/io/readers/__init__.py

@@ -0,0 +1 @@
+

tests/functional/test_videoio.py → tests/functional/io/readers/test_video_source.py

@@ -1,23 +1,5 @@
-from pathlib import Path
-
-import pytest
-
 from flowty.cv.videoio import VideoSource
-import os
-import numpy as np
-
-from flowty.videoio import FlowImageWriter
-
-MEDIA_ROOT = os.path.join(
-    os.path.dirname(__file__),
-    '..',  '..', 'media'
-)
-VIDEO_PATHS = {
-    'mp4': os.path.join(MEDIA_ROOT, 'mr-bubz.mp4'),
-    'webm': os.path.join(MEDIA_ROOT, 'mr-bubz.webm'),
-    'avi': os.path.join(MEDIA_ROOT, 'mr-bubz.avi'),
-    'jpeg': os.path.join(MEDIA_ROOT, 'mr-bubz', 'frame_%05d.jpg'),
-}
+from tests.resources import VIDEO_PATHS
 
 
 class TestVideoSource:
@@ -88,40 +70,4 @@ def read_video(self, video_path, backend='ffmpeg'):
         for frame in iter(src):
             shape = frame.shape
             n_frames += 1
-        return n_frames, shape
-
-
-class TestFlowImageWriter:
-    def test_saving_single_flow_image(self, tmp_path):
-        image_writer = self.get_flow_writer(tmp_path)
-        flow = np.random.randint(low=0, high=255, size=(5, 5, 2), dtype=np.uint8)
-
-        image_writer.write(flow)
-
-        for axis in ['u', 'v']:
-            assert (Path(tmp_path) / axis / '00001.jpg').exists()
-
-    def test_throws_error_if_flow_not_uint8(self, tmp_path):
-        image_writer = self.get_flow_writer(tmp_path)
-        flow = np.random.randint(low=0, high=255, size=(5, 5, 2), dtype=np.int32)
-
-        with pytest.raises(ValueError):
-            image_writer.write(flow)
-
-    def test_throws_error_if_flow_not_2_channels(self, tmp_path):
-        image_writer = self.get_flow_writer(tmp_path)
-        flow = np.random.randint(low=0, high=255, size=(5, 5, 3), dtype=np.uint8)
-
-        with pytest.raises(ValueError):
-            image_writer.write(flow)
-
-    def test_throws_error_if_flow_not_3_dimensional(self, tmp_path):
-        image_writer = self.get_flow_writer(tmp_path)
-        flow = np.random.randint(low=0, high=255, size=(5, 5), dtype=np.uint8)
-
-        with pytest.raises(ValueError):
-            image_writer.write(flow)
-
-    def get_flow_writer(self, tmp_path):
-        filename_template = tmp_path / '{axis}/{index:05d}.jpg'
-        return FlowImageWriter(str(filename_template))
+        return n_frames, shape

tests/functional/io/writers/test_flow_writer.py

@@ -0,0 +1,42 @@
+from pathlib import Path
+
+import numpy as np
+import pytest
+
+from flowty.videoio import FlowImageWriter
+
+
+class TestFlowImageWriter:
+    def test_saving_single_flow_image(self, tmp_path):
+        image_writer = self.get_flow_writer(tmp_path)
+        flow = np.random.randint(low=0, high=255, size=(5, 5, 2), dtype=np.uint8)
+
+        image_writer.write(flow)
+
+        for axis in ['u', 'v']:
+            assert (Path(tmp_path) / axis / '00001.jpg').exists()
+
+    def test_throws_error_if_flow_not_uint8(self, tmp_path):
+        image_writer = self.get_flow_writer(tmp_path)
+        flow = np.random.randint(low=0, high=255, size=(5, 5, 2), dtype=np.int32)
+
+        with pytest.raises(ValueError):
+            image_writer.write(flow)
+
+    def test_throws_error_if_flow_not_2_channels(self, tmp_path):
+        image_writer = self.get_flow_writer(tmp_path)
+        flow = np.random.randint(low=0, high=255, size=(5, 5, 3), dtype=np.uint8)
+
+        with pytest.raises(ValueError):
+            image_writer.write(flow)
+
+    def test_throws_error_if_flow_not_3_dimensional(self, tmp_path):
+        image_writer = self.get_flow_writer(tmp_path)
+        flow = np.random.randint(low=0, high=255, size=(5, 5), dtype=np.uint8)
+
+        with pytest.raises(ValueError):
+            image_writer.write(flow)
+
+    def get_flow_writer(self, tmp_path):
+        filename_template = tmp_path / '{axis}/{index:05d}.jpg'
+        return FlowImageWriter(str(filename_template))

tests/functional/test_flowty.py

@@ -1,47 +1,25 @@
-import sys
-from io import StringIO
-
-from unittest.mock import Mock
-
-from flowty.flowty import main, command_parsers
-
-
-class IOCapture:
-    def __init__(self):
-        self._original_stdout = sys.stdout
-        self._original_stderr = sys.stderr
-        self._stdout = StringIO()
-        self._stderr = StringIO()
-
-    def __enter__(self):
-        sys.stdout = self._stdout
-        sys.stderr = self._stderr
-        return self
-
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        sys.stdout = self._original_stdout
-        sys.stderr = self._original_stderr
-
-    @property
-    def stdout(self):
-        return self._stdout.getvalue()
-
-    @property
-    def stderr(self):
-        return self._stderr.getvalue()
-
-
-class TestFlowtyCli:
-    def test_prints_help_when_no_command_is_provided(self):
-        with IOCapture() as capture:
-            main([])
-            assert 'usage: flowty' in capture.stdout
-
-    def test_invokes_registered_command(self):
-        tvl1_parser = command_parsers.add_parser('tvl1')
-        tvl1_command = Mock()
-        tvl1_parser.set_defaults(command=tvl1_command)
-
-        main(['tvl1'])
-
-        tvl1_command.assert_called_once()
+from pathlib import Path
+import numpy as np
+from imageio import imread
+
+from flowty import flowty
+from tests.resources import RUBBER_WHALE
+
+
+class TestFlowty:
+    def test_tvl1_flow_from_mp4_to_uv_images(self, tmpdir):
+        src = RUBBER_WHALE['media_path']['png']
+        output_dir = tmpdir / 'flow'
+        dest = output_dir / '{axis}' / 'frame{index:02d}.png'
+        stride = 2
+
+        flowty.main(["dis", "--video-stride", str(stride), str(src), str(dest)])
+
+        for axis in ['u', 'v']:
+            for i in range(1, RUBBER_WHALE['frame_count'] // stride):
+                flow_frame_path = Path(output_dir / axis / 'frame{:02d}.png'.format(i))
+                assert flow_frame_path.exists()
+                flow_frame = imread(flow_frame_path)
+                assert flow_frame.shape == RUBBER_WHALE['resolution']
+                assert flow_frame.dtype == np.uint8
+            assert not Path(output_dir / axis / 'frame{:02d}.png'.format( RUBBER_WHALE['frame_count'])).exists()

tests/resources.py

@@ -0,0 +1,18 @@
+from pathlib import Path
+
+
+MEDIA_ROOT = Path(__file__).parent.parent / 'media'
+VIDEO_PATHS = {
+    'mp4': str(MEDIA_ROOT / 'mr-bubz.mp4'),
+    'webm': str(MEDIA_ROOT / 'mr-bubz.webm'),
+    'avi': str(MEDIA_ROOT / 'mr-bubz.avi'),
+    'jpeg': str(MEDIA_ROOT / 'mr-bubz' / 'frame_%05d.jpg'),
+}
+RUBBER_WHALE = {
+    'media_path': {
+        'mp4': str(MEDIA_ROOT / 'rubber-whale.mp4'),
+        'png': str(MEDIA_ROOT / 'rubber-whale' / 'frame%02d.png'),
+    },
+    'resolution': (388, 584),
+    'frame_count': 8,
+}

tests/unit/test_flowty.py

@@ -0,0 +1,47 @@
+import sys
+from io import StringIO
+
+from unittest.mock import Mock
+
+from flowty.flowty import main, command_parsers
+
+
+class IOCapture:
+    def __init__(self):
+        self._original_stdout = sys.stdout
+        self._original_stderr = sys.stderr
+        self._stdout = StringIO()
+        self._stderr = StringIO()
+
+    def __enter__(self):
+        sys.stdout = self._stdout
+        sys.stderr = self._stderr
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        sys.stdout = self._original_stdout
+        sys.stderr = self._original_stderr
+
+    @property
+    def stdout(self):
+        return self._stdout.getvalue()
+
+    @property
+    def stderr(self):
+        return self._stderr.getvalue()
+
+
+class TestFlowtyCli:
+    def test_prints_help_when_no_command_is_provided(self):
+        with IOCapture() as capture:
+            main([])
+            assert 'usage: flowty' in capture.stdout
+
+    def test_invokes_registered_command(self):
+        tvl1_parser = command_parsers.add_parser('tvl1')
+        tvl1_command = Mock()
+        tvl1_parser.set_defaults(command=tvl1_command)
+
+        main(['tvl1'])
+
+        tvl1_command.assert_called_once()