fix: mock sys.argv when running core service

tests/core/_backup/test_backup.py

@@ -10,6 +10,7 @@
 # specific language governing permissions and limitations under the License.
 
 import os
+from unittest.mock import patch
 
 import pytest
 
@@ -46,8 +47,9 @@ backup_file_path = ".taipy_backups"
 
 
 def test_backup_storage_folder_when_core_run():
-    core = Core()
-    core.run()
+    with patch("sys.argv", ["prog"]):
+        core = Core()
+        core.run()
     backup_files = read_backup_file(backup_file_path)
     assert backup_files == [f"{Config.core.storage_folder}\n"]
     core.stop()

tests/core/test_core.py

@@ -9,6 +9,8 @@
 # an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 # specific language governing permissions and limitations under the License.
 
+from unittest.mock import patch
+
 import pytest
 
 from src.taipy.core import Core
@@ -24,9 +26,10 @@ from taipy.config.exceptions.exceptions import ConfigurationUpdateBlocked
 class TestCore:
     def test_run_core_trigger_config_check(self, caplog):
         Config.configure_data_node(id="d0", storage_type="toto")
-        with pytest.raises(SystemExit):
-            core = Core()
-            core.run()
+        with patch("sys.argv", ["prog"]):
+            with pytest.raises(SystemExit):
+                core = Core()
+                core.run()
         expected_error_message = (
             "`storage_type` field of DataNodeConfig `d0` must be either csv, sql_table,"
             " sql, mongo_collection, pickle, excel, generic, json, parquet, or in_memory."
@@ -38,42 +41,44 @@ class TestCore:
     def test_run_core_as_a_service_development_mode(self):
         _OrchestratorFactory._dispatcher = None
 
-        core = Core()
-        assert core._orchestrator is None
-        assert core._dispatcher is None
-        assert _OrchestratorFactory._dispatcher is None
-
-        core.run()
-        assert core._orchestrator is not None
-        assert core._orchestrator == _Orchestrator
-        assert _OrchestratorFactory._orchestrator is not None
-        assert _OrchestratorFactory._orchestrator == _Orchestrator
-        assert core._dispatcher is not None
-        assert isinstance(core._dispatcher, _DevelopmentJobDispatcher)
-        assert isinstance(_OrchestratorFactory._dispatcher, _DevelopmentJobDispatcher)
-        core.stop()
+        with patch("sys.argv", ["prog"]):
+            core = Core()
+            assert core._orchestrator is None
+            assert core._dispatcher is None
+            assert _OrchestratorFactory._dispatcher is None
+
+            core.run()
+            assert core._orchestrator is not None
+            assert core._orchestrator == _Orchestrator
+            assert _OrchestratorFactory._orchestrator is not None
+            assert _OrchestratorFactory._orchestrator == _Orchestrator
+            assert core._dispatcher is not None
+            assert isinstance(core._dispatcher, _DevelopmentJobDispatcher)
+            assert isinstance(_OrchestratorFactory._dispatcher, _DevelopmentJobDispatcher)
+            core.stop()
 
     def test_run_core_as_a_service_standalone_mode(self):
         _OrchestratorFactory._dispatcher = None
 
-        core = Core()
-        assert core._orchestrator is None
-
-        assert core._dispatcher is None
-        assert _OrchestratorFactory._dispatcher is None
-
-        Config.configure_job_executions(mode=JobConfig._STANDALONE_MODE, max_nb_of_workers=2)
-        core.run()
-        assert core._orchestrator is not None
-        assert core._orchestrator == _Orchestrator
-        assert _OrchestratorFactory._orchestrator is not None
-        assert _OrchestratorFactory._orchestrator == _Orchestrator
-        assert core._dispatcher is not None
-        assert isinstance(core._dispatcher, _StandaloneJobDispatcher)
-        assert isinstance(_OrchestratorFactory._dispatcher, _StandaloneJobDispatcher)
-        assert core._dispatcher.is_running()
-        assert _OrchestratorFactory._dispatcher.is_running()
-        core.stop()
+        with patch("sys.argv", ["prog"]):
+            core = Core()
+            assert core._orchestrator is None
+
+            assert core._dispatcher is None
+            assert _OrchestratorFactory._dispatcher is None
+
+            Config.configure_job_executions(mode=JobConfig._STANDALONE_MODE, max_nb_of_workers=2)
+            core.run()
+            assert core._orchestrator is not None
+            assert core._orchestrator == _Orchestrator
+            assert _OrchestratorFactory._orchestrator is not None
+            assert _OrchestratorFactory._orchestrator == _Orchestrator
+            assert core._dispatcher is not None
+            assert isinstance(core._dispatcher, _StandaloneJobDispatcher)
+            assert isinstance(_OrchestratorFactory._dispatcher, _StandaloneJobDispatcher)
+            assert core._dispatcher.is_running()
+            assert _OrchestratorFactory._dispatcher.is_running()
+            core.stop()
 
     def test_core_service_can_only_be_run_once(self):
         core_instance_1 = Core()
@@ -97,18 +102,20 @@ class TestCore:
     def test_block_config_update_when_core_service_is_running_development_mode(self):
         _OrchestratorFactory._dispatcher = None
 
-        core = Core()
-        core.run()
-        with pytest.raises(ConfigurationUpdateBlocked):
-            Config.configure_data_node(id="i1")
-        core.stop()
+        with patch("sys.argv", ["prog"]):
+            core = Core()
+            core.run()
+            with pytest.raises(ConfigurationUpdateBlocked):
+                Config.configure_data_node(id="i1")
+            core.stop()
 
     def test_block_config_update_when_core_service_is_running_standalone_mode(self):
         _OrchestratorFactory._dispatcher = None
 
-        core = Core()
-        Config.configure_job_executions(mode=JobConfig._STANDALONE_MODE, max_nb_of_workers=2)
-        core.run()
-        with pytest.raises(ConfigurationUpdateBlocked):
-            Config.configure_data_node(id="i1")
-        core.stop()
+        with patch("sys.argv", ["prog"]):
+            core = Core()
+            Config.configure_job_executions(mode=JobConfig._STANDALONE_MODE, max_nb_of_workers=2)
+            core.run()
+            with pytest.raises(ConfigurationUpdateBlocked):
+                Config.configure_data_node(id="i1")
+            core.stop()

tests/core/test_taipy.py

@@ -551,8 +551,9 @@ class TestTaipy:
         task_cfg_1 = Config.configure_task("t1", print, input_cfg_1, output_cfg_1)
         scenario_cfg_1 = Config.configure_scenario("s1", [task_cfg_1], [], Frequency.DAILY)
 
-        core = Core()
-        core.run()
+        with mock.patch("sys.argv", ["prog"]):
+            core = Core()
+            core.run()
 
         scenario_1 = tp.create_scenario(scenario_cfg_1)
         tp.submit(scenario_1)
@@ -569,8 +570,9 @@ class TestTaipy:
         task_cfg_1 = Config.configure_task("t1", print, input_cfg_1, output_cfg_1)
         scenario_cfg_1 = Config.configure_scenario("s1", [task_cfg_1], [], Frequency.DAILY)
 
-        core = Core()
-        core.run()
+        with mock.patch("sys.argv", ["prog"]):
+            core = Core()
+            core.run()
 
         scenario_1 = tp.create_scenario(scenario_cfg_1)
         tp.submit(scenario_1, wait=True)