Move test_telemetry.py to mpa tests in external (#1580)

Signed-off-by: Yunchu Lee <[email protected]>

Author: yunchu

.pre-commit-config.yaml

@@ -4,7 +4,7 @@ default_language_version:
 
 repos:
   - repo: https://github.com/PyCQA/isort
-    rev: "5.10.1"
+    rev: "5.12.0"
     hooks:
       - id: isort
         alias: isort_ote_sdk

tests/ote_cli/test_telemetry.py renamed to external/model-preparation-algorithm/tests/ote_cli/test_telemetry.py

@@ -1,70 +1,74 @@
-import sys
-import subprocess
+import os
 import pytest
-import unittest
-from unittest.mock import MagicMock, patch
+import sys
+from unittest.mock import patch
+
+from ote_cli.registry import Registry
+from ote_cli.utils.tests import (
+    create_venv,
+    get_some_vars,
+)
 
 from ote_sdk.test_suite.e2e_test_system import e2e_pytest_component
 
-from ote_cli.tools import ote
 
-from ote_cli.tools.ote import (
-    ote_demo,
-    ote_deploy,
-    ote_eval,
-    ote_export,
-    ote_find,
-    ote_optimize,
-    ote_train
-)
+TT_STABILITY_TESTS = os.environ.get("TT_STABILITY_TESTS", False)
+root = "/tmp/ote_cli/"
+templates = Registry("external/model-preparation-algorithm").filter(task_type="CLASSIFICATION").templates
+templates_ids = [template.model_template_id for template in templates]
 
 
-class TestTelemetry(unittest.TestCase):
+class TestToolsMPATelemetry:
+    @e2e_pytest_component
+    def test_create_venv(self):
+        work_dir, _, algo_backend_dir = get_some_vars(templates[0], root)
+        create_venv(algo_backend_dir, work_dir)
 
     @e2e_pytest_component
+    @pytest.mark.skipif(TT_STABILITY_TESTS, reason="This is TT_STABILITY_TESTS")
     @patch("ote_cli.tools.ote.ote_demo", return_value=None)
     @patch("ote_cli.utils.telemetry.init_telemetry_session", return_value=None)
     @patch("ote_cli.utils.telemetry.close_telemetry_session", return_value=None)
     @patch("ote_cli.utils.telemetry.send_version", return_value=None)
     @patch("ote_cli.utils.telemetry.send_cmd_results", return_value=None)
-    def test_tm_integration_exit_0(self,
-        mock_send_cmd,
-        mock_send_version,
-        mock_close_tm,
-        mock_init_tm,
-        mock_demo
-    ):
+    def test_tm_integration_exit_0(self, mock_send_cmd, mock_send_version, mock_close_tm, mock_init_tm, mock_demo):
+        from ote_cli.tools import ote
+
         backup_argv = sys.argv
         sys.argv = ["ote", "demo"]
         ret = ote.main()
         sys.argv = backup_argv
 
-        self.assertEqual(ret, 0)
+        assert ret == 0
         mock_demo.assert_called_once()
         mock_init_tm.assert_called_once()
         mock_close_tm.assert_called_once()
         mock_send_cmd.assert_called_with(None, "demo", {"retcode": 0})
 
     @e2e_pytest_component
+    @pytest.mark.skipif(TT_STABILITY_TESTS, reason="This is TT_STABILITY_TESTS")
     @patch("ote_cli.tools.ote.ote_demo", side_effect=Exception())
     @patch("ote_cli.utils.telemetry.init_telemetry_session", return_value=None)
     @patch("ote_cli.utils.telemetry.close_telemetry_session", return_value=None)
     @patch("ote_cli.utils.telemetry.send_version", return_value=None)
     @patch("ote_cli.utils.telemetry.send_cmd_results", return_value=None)
-    def test_tm_integration_exit_exception(self,
+    def test_tm_integration_exit_exception(
+        self,
         mock_send_cmd,
         mock_send_version,
         mock_close_tm,
         mock_init_tm,
         mock_demo,
     ):
+        from ote_cli.tools import ote
+
         backup_argv = sys.argv
         sys.argv = ["ote", "demo"]
         with pytest.raises(Exception) as e:
-            ret = ote.main()
+            ote.main()
         sys.argv = backup_argv
 
-        self.assertEqual(e.type, Exception, f"{e}")
+        assert e.type == Exception, f"{e}"
         mock_init_tm.assert_called_once()
         mock_close_tm.assert_called_once()
         mock_send_cmd.assert_called_with(None, "demo", {"retcode": -1, "exception": repr(Exception())})

tests/ote_cli/conftest.py

@@ -36,7 +36,13 @@ def manage_tm_config_for_testing():
     isip_exist = os.path.exists(isip_path)
     otm_exist = os.path.exists(otm_path)
 
+    if not os.path.exists(cfg_dir):
+        created_cfg_dir = True
+        os.makedirs(cfg_dir)
+
     isip_backup = None
+    is_cfg_dir_existed = True
+
     if not isip_exist:
         with open(isip_path, "w") as f:
             f.write("0")
@@ -68,3 +74,6 @@ def manage_tm_config_for_testing():
         if otm_backup is not None:
             with open(otm_path, "w") as f:
                 f.write(otm_backup)
+
+    if created_cfg_dir:
+        os.rmdir(cfg_dir)