Add unit test for export (#3211)

* implement unit test

* align with pre-commit

* avoid import error

* remove some tests

* align with pre-commit

tests/test_helpers.py

@@ -4,6 +4,12 @@
 
 """Helper functions for tests."""
 
+import contextlib
+import tempfile
+from pathlib import Path
+from typing import Iterator
+
+import cv2
 import numpy as np
 
 
@@ -40,3 +46,122 @@ def generate_random_bboxes(
     areas = (x_max - x_min) * (y_max - y_min)
     bboxes = np.column_stack((x_min, y_min, x_max, y_max))
     return bboxes[areas > 0]
+
+
+@contextlib.contextmanager
+def generate_random_image_folder(width: int = 480, height: int = 360, number_of_images: int = 10) -> Iterator[str]:
+    """
+    Generates a folder with random images, cleans up automatically if used in a `with` statement
+
+    Parameters:
+        width (int): height of the images. Defaults to 480.
+        height (int): width of the images. Defaults to 360.
+        number_of_images (int): number of generated images. Defaults to 10.
+
+    Returns:
+        Iterator[str]: The temporary directory
+    """
+    temp_dir = tempfile.TemporaryDirectory()
+
+    for n in range(number_of_images):
+        temp_file = str(Path(temp_dir.name) / f"{n}.jpg")
+        _write_random_image(width, height, temp_file)
+
+    try:
+        yield temp_dir.name
+    finally:
+        temp_dir.cleanup()
+
+
+@contextlib.contextmanager
+def generate_random_video_folder(
+    width: int = 480,
+    height: int = 360,
+    number_of_videos: int = 10,
+    number_of_frames: int = 150,
+) -> Iterator[str]:
+    """
+    Generates a folder with random videos, cleans up automatically if used in a `with` statement
+
+    Parameters:
+        width (int): Width of the video. Defaults to 480.
+        height (int): Height of the video. Defaults to 360.
+        number_of_videos (int): Number of videos to generate. Defaults to 10.
+        number_of_frames (int): Number of frames in each video. Defaults to 150.
+
+    Returns:
+        Iterator[str]: A temporary directory with videos
+    """
+    temp_dir = tempfile.TemporaryDirectory()
+
+    for n in range(number_of_videos):
+        temp_file = str(Path(temp_dir.name) / f"{n}.mp4")
+        _write_random_video(width, height, number_of_frames, temp_file)
+
+    try:
+        yield temp_dir.name
+    finally:
+        temp_dir.cleanup()
+
+
+@contextlib.contextmanager
+def generate_random_single_image(width: int = 480, height: int = 360) -> Iterator[str]:
+    """
+    Generates a random image, cleans up automatically if used in a `with` statement
+
+    Parameters:
+        width (int): Width of the image. Defaults to 480.
+        height (int): Height of the image. Defaults to 360.
+
+    Returns:
+        Iterator[str]: Path to an image file
+    """
+
+    temp_dir = tempfile.TemporaryDirectory()
+    temp_file = str(Path(temp_dir.name) / "temp_image.jpg")
+    _write_random_image(width, height, temp_file)
+
+    try:
+        yield temp_file
+    finally:
+        temp_dir.cleanup()
+
+
+@contextlib.contextmanager
+def generate_random_single_video(width: int = 480, height: int = 360, number_of_frames: int = 150) -> Iterator[str]:
+    """
+    Generates a random video, cleans up automatically if used in a `with` statement
+
+    Parameters:
+        width (int): Width of the video. Defaults to 480.
+        height (int): Height of the video. Defaults to 360.
+        number_of_frames (int): Number of frames in the video. Defaults to 150.
+
+    Returns:
+        Iterator[str]: Path to a video file
+    """
+    temp_dir = tempfile.TemporaryDirectory()
+    temp_file = str(Path(temp_dir.name) / "temp_video.mp4")
+    _write_random_video(width, height, number_of_frames, temp_file)
+
+    try:
+        yield temp_file
+    finally:
+        temp_dir.cleanup()
+
+
+def _write_random_image(width: int, height: int, filename: str) -> None:
+    img = np.uint8(np.random.random((height, width, 3)) * 255)
+    cv2.imwrite(filename, img)
+
+
+def _write_random_video(width: int, height: int, number_of_frames: int, filename: str) -> None:
+    fourcc = cv2.VideoWriter_fourcc(*"mp4v")
+    f = filename
+    videowriter = cv2.VideoWriter(f, fourcc, 30, (width, height))
+
+    for _ in range(number_of_frames):
+        img = np.uint8(np.random.random((height, width, 3)) * 255)
+        videowriter.write(img)
+
+    videowriter.release()

tests/unit/core/exporter/exportable_code/__init__.py

@@ -0,0 +1,2 @@
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0

tests/unit/core/exporter/exportable_code/conftest.py

@@ -0,0 +1,17 @@
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+import importlib
+import importlib.util
+import sys
+from pathlib import Path
+
+from otx.core.exporter.exportable_code import demo
+
+# exportable_code is standalone package and files in exportable_code import 'demo_package' which is same as
+# 'otx.core.exporter.exportable_code.demo.demo_package'. To avoid import error while running test, need to
+# import 'otx.core.exporter.exportable_code.demo.demo_package' and register it as 'demo_package'.
+demo_package_file = Path(demo.__file__).parent / "demo_package" / "__init__.py"
+spec = importlib.util.spec_from_file_location("demo_package", demo_package_file)
+module = importlib.util.module_from_spec(spec)
+sys.modules["demo_package"] = module

tests/unit/core/exporter/exportable_code/demo/__init__.py

@@ -0,0 +1,2 @@
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0

tests/unit/core/exporter/exportable_code/demo/demo_package/__init__.py

@@ -0,0 +1,2 @@
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0

tests/unit/core/exporter/exportable_code/demo/demo_package/executors/__init__.py

@@ -0,0 +1,2 @@
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0

tests/unit/core/exporter/exportable_code/demo/demo_package/executors/test_asynchronous.py

@@ -0,0 +1,79 @@
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+"""Test of AsyncExecutor in demo_package."""
+
+from unittest.mock import MagicMock
+
+import pytest
+from otx.core.exporter.exportable_code.demo.demo_package.executors import asynchronous as target_file
+from otx.core.exporter.exportable_code.demo.demo_package.executors.asynchronous import AsyncExecutor
+
+
+class MockAsyncPipeline:
+    def __init__(self, *args, **kwargs):
+        self.arr = []
+        self.idx = 0
+
+    def get_result(self, *args, **kwawrgs):
+        if self.idx >= len(self.arr):
+            return None
+        ret = self.arr[self.idx]
+        self.idx += 1
+        return ret
+
+    def submit_data(self, frame, *args, **kwawrgs):
+        self.arr.append(frame)
+
+    def await_all(self):
+        pass
+
+
+class TestAsyncExecutor:
+    @pytest.fixture(autouse=True)
+    def setup(self, mocker):
+        mocker.patch.object(target_file, "AsyncPipeline", side_effect=MockAsyncPipeline)
+
+    @pytest.fixture()
+    def mock_model(self):
+        return MagicMock()
+
+    @pytest.fixture()
+    def mock_visualizer(self):
+        visualizer = MagicMock()
+        visualizer.is_quit.return_value = False
+        visualizer.draw.side_effect = lambda x, _: x
+        return visualizer
+
+    def test_init(self, mock_model, mock_visualizer):
+        AsyncExecutor(mock_model, mock_visualizer)
+
+    @pytest.fixture()
+    def mock_streamer(self, mocker):
+        return mocker.patch.object(target_file, "get_streamer", return_value=range(1, 4))
+
+    @pytest.fixture()
+    def mock_dump_frames(self, mocker):
+        return mocker.patch.object(target_file, "dump_frames")
+
+    def test_run(self, mocker, mock_model, mock_visualizer, mock_streamer, mock_dump_frames):
+        mock_render_result = mocker.patch.object(AsyncExecutor, "render_result", side_effect=lambda x: x)
+        executor = AsyncExecutor(mock_model, mock_visualizer)
+        executor.run(MagicMock())
+
+        mock_render_result.assert_called()
+        for i in range(1, 4):
+            assert mock_render_result.call_args_list[i - 1].args == (i,)
+        mock_visualizer.show.assert_called()
+        for i in range(1, 4):
+            assert mock_visualizer.show.call_args_list[i - 1].args == (i,)
+        mock_dump_frames.assert_called()
+
+    def test_render_result(self, mock_model, mock_visualizer):
+        executor = AsyncExecutor(mock_model, mock_visualizer)
+        mock_pred = MagicMock()
+        cur_frame = MagicMock()
+        frame_meta = {"frame": cur_frame}
+        fake_results = (mock_pred, frame_meta)
+        executor.render_result(fake_results)
+
+        mock_visualizer.draw.assert_called_once_with(cur_frame, mock_pred)

tests/unit/core/exporter/exportable_code/demo/demo_package/executors/test_synchronous.py

@@ -0,0 +1,54 @@
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+"""Test of AsyncExecutor in demo_package."""
+
+from unittest.mock import MagicMock
+
+import pytest
+from otx.core.exporter.exportable_code.demo.demo_package.executors import synchronous as target_file
+from otx.core.exporter.exportable_code.demo.demo_package.executors.synchronous import SyncExecutor
+
+
+class TestSyncExecutor:
+    @pytest.fixture()
+    def mock_model(self):
+        return MagicMock(side_effect=lambda x: (x, x))
+
+    @pytest.fixture()
+    def mock_visualizer(self):
+        visualizer = MagicMock()
+        visualizer.is_quit.return_value = False
+        visualizer.draw.side_effect = lambda x, _: x
+        return visualizer
+
+    def test_init(self, mock_model, mock_visualizer):
+        SyncExecutor(mock_model, mock_visualizer)
+
+    @pytest.fixture()
+    def mock_streamer(self, mocker):
+        return mocker.patch.object(target_file, "get_streamer", return_value=range(3))
+
+    @pytest.fixture()
+    def mock_dump_frames(self, mocker):
+        return mocker.patch.object(target_file, "dump_frames")
+
+    def test_run(self, mock_model, mock_visualizer, mock_streamer, mock_dump_frames):
+        executor = SyncExecutor(mock_model, mock_visualizer)
+        mock_input_stream = MagicMock()
+        executor.run(mock_input_stream, MagicMock())
+
+        mock_model.assert_called()
+        for i in range(3):
+            assert mock_model.call_args_list[i] == ((i,),)
+        mock_visualizer.draw.assert_called()
+        for i in range(3):
+            assert mock_visualizer.draw.call_args_list[i] == ((i, i),)
+        mock_visualizer.show.assert_called()
+        for i in range(3):
+            assert mock_visualizer.show.call_args_list[i] == ((i,),)
+        mock_dump_frames.assert_called_once_with(
+            list(range(3)),
+            mock_visualizer.output,
+            mock_input_stream,
+            range(3),
+        )

tests/unit/core/exporter/exportable_code/demo/demo_package/streamer/__init__.py

@@ -0,0 +1,2 @@
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0