Add unit test for export for develop branch (#3212)

* implement unit test

* align with pre-commit

* avoid import error

* remove some tests

* align with pre-commit

* register demo_package in fixture

* change ctx manager to fixture

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,34 @@
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+import importlib
+import importlib.util
+import sys
+from pathlib import Path
+
+import pytest
+from otx.core.exporter.exportable_code import demo
+
+
+@pytest.fixture(scope="package", autouse=True)
+def fxt_fix_exportable_code_import_error():
+    # 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' and it makes an error as below.
+    #    import demo_package  # noqa: ERA001
+    # E  ModuleNotFoundError: No module named '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)
+
+    tmp_mod_demo_package = sys.modules.get("demo_package")
+    sys.modules["demo_package"] = module
+
+    yield
+
+    # Revert sys.modules["demo_package"] modification
+    if tmp_mod_demo_package is not None:
+        sys.modules["demo_package"] = tmp_mod_demo_package
+    else:
+        sys.modules.pop("demo_package")

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,91 @@
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+"""Test of AsyncExecutor in demo_package."""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+target_file = None
+AsyncExecutor = None
+
+
+@pytest.fixture(scope="module", autouse=True)
+def fxt_import_module():
+    global target_file  # noqa: PLW0603
+    global AsyncExecutor  # noqa: PLW0603
+    from otx.core.exporter.exportable_code.demo.demo_package.executors import asynchronous
+    from otx.core.exporter.exportable_code.demo.demo_package.executors.asynchronous import AsyncExecutor as Cls1
+
+    target_file = asynchronous
+    AsyncExecutor = Cls1
+
+
+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,66 @@
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+"""Test of AsyncExecutor in demo_package."""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+target_file = None
+SyncExecutor = None
+
+
+@pytest.fixture(scope="module", autouse=True)
+def fxt_import_module():
+    global target_file  # noqa: PLW0603
+    global SyncExecutor  # noqa: PLW0603
+    from otx.core.exporter.exportable_code.demo.demo_package.executors import synchronous
+    from otx.core.exporter.exportable_code.demo.demo_package.executors.synchronous import SyncExecutor as Cls1
+
+    target_file = synchronous
+    SyncExecutor = Cls1
+
+
+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