From 94ad1f3552dab5d990af6e75780a9516ed6519a4 Mon Sep 17 00:00:00 2001 From: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com> Date: Mon, 18 Aug 2025 16:54:59 +0200 Subject: [PATCH 1/3] test(cli): add comprehensive CLI test suite and reorganize files --- tests/cli/test_chat.py | 315 +++++++++++++++ tests/cli/test_cli_main.py | 336 ++++++++++++++++ tests/cli/test_debugger.py | 233 +++++++++++ tests/cli/test_llm_providers.py | 300 ++++++++++++++ tests/cli/test_migration.py | 686 ++++++++++++++++++++++++++++++++ tests/test_cli.py | 34 -- tests/test_cli_migration.py | 282 ------------- 7 files changed, 1870 insertions(+), 316 deletions(-) create mode 100644 tests/cli/test_chat.py create mode 100644 tests/cli/test_cli_main.py create mode 100644 tests/cli/test_debugger.py create mode 100644 tests/cli/test_llm_providers.py create mode 100644 tests/cli/test_migration.py delete mode 100644 tests/test_cli.py delete mode 100644 tests/test_cli_migration.py diff --git a/tests/cli/test_chat.py b/tests/cli/test_chat.py new file mode 100644 index 000000000..341778609 --- /dev/null +++ b/tests/cli/test_chat.py @@ -0,0 +1,315 @@ +# SPDX-FileCopyrightText: Copyright (c) 2023 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on 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. + +import asyncio +import sys +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from nemoguardrails.cli.chat import ( + ChatState, + extract_scene_text_content, + parse_events_inputs, + run_chat, +) + +chat_module = sys.modules["nemoguardrails.cli.chat"] + + +class TestParseEventsInputs: + def test_parse_simple_event(self): + result = parse_events_inputs("UserAction") + assert result == {"type": "UserAction"} + + def test_parse_event_with_params(self): + result = parse_events_inputs('UserAction(name="test", value=123)') + assert result == {"type": "UserAction", "name": "test", "value": 123} + + def test_parse_event_with_string_params(self): + result = parse_events_inputs('UserAction(message="hello world")') + assert result == {"type": "UserAction", "message": "hello world"} + + def test_parse_nested_event(self): + result = parse_events_inputs("bot.UtteranceAction") + assert result == {"type": "botUtteranceAction"} + + def test_parse_event_with_nested_params(self): + result = parse_events_inputs('UserAction(data={"key": "value"})') + assert result == {"type": "UserAction", "data": {"key": "value"}} + + def test_parse_event_with_list_params(self): + result = parse_events_inputs("UserAction(items=[1, 2, 3])") + assert result == {"type": "UserAction", "items": [1, 2, 3]} + + def test_parse_invalid_event(self): + result = parse_events_inputs("Invalid.Event.Format.TooMany") + assert result is None + + def test_parse_event_missing_equals(self): + result = parse_events_inputs("UserAction(invalid_param)") + assert result is None + + +class TestExtractSceneTextContent: + def test_extract_empty_list(self): + result = extract_scene_text_content([]) + assert result == "" + + def test_extract_single_text(self): + content = [{"text": "Hello World"}] + result = extract_scene_text_content(content) + assert result == "\nHello World" + + def test_extract_multiple_texts(self): + content = [{"text": "Line 1"}, {"text": "Line 2"}, {"text": "Line 3"}] + result = extract_scene_text_content(content) + assert result == "\nLine 1\nLine 2\nLine 3" + + def test_extract_mixed_content(self): + content = [ + {"text": "Text 1"}, + {"image": "image.png"}, + {"text": "Text 2"}, + {"button": "Click Me"}, + ] + result = extract_scene_text_content(content) + assert result == "\nText 1\nText 2" + + def test_extract_no_text_content(self): + content = [{"image": "image.png"}, {"button": "Click Me"}] + result = extract_scene_text_content(content) + assert result == "" + + +class TestChatState: + def test_initial_state(self): + chat_state = ChatState() + assert chat_state.state is None + assert chat_state.waiting_user_input is False + assert chat_state.paused is False + assert chat_state.running_timer_tasks == {} + assert chat_state.input_events == [] + assert chat_state.output_events == [] + assert chat_state.output_state is None + assert chat_state.events_counter == 0 + assert chat_state.first_time is False + + +class TestRunChat: + @patch("asyncio.run") + @patch.object(chat_module, "LLMRails") + @patch.object(chat_module, "RailsConfig") + def test_run_chat_v1_0(self, mock_rails_config, mock_llm_rails, mock_asyncio_run): + mock_config = MagicMock() + mock_config.colang_version = "1.0" + mock_rails_config.from_path.return_value = mock_config + + run_chat(config_path="test_config") + + mock_rails_config.from_path.assert_called_once_with("test_config") + mock_asyncio_run.assert_called_once() + + @patch.object(chat_module, "get_or_create_event_loop") + @patch.object(chat_module, "LLMRails") + @patch.object(chat_module, "RailsConfig") + def test_run_chat_v2_x(self, mock_rails_config, mock_llm_rails, mock_get_loop): + mock_config = MagicMock() + mock_config.colang_version = "2.x" + mock_rails_config.from_path.return_value = mock_config + + mock_loop = MagicMock() + mock_get_loop.return_value = mock_loop + + run_chat(config_path="test_config") + + mock_rails_config.from_path.assert_called_once_with("test_config") + mock_llm_rails.assert_called_once_with(mock_config, verbose=False) + mock_loop.run_until_complete.assert_called_once() + + @patch.object(chat_module, "RailsConfig") + def test_run_chat_invalid_version(self, mock_rails_config): + mock_config = MagicMock() + mock_config.colang_version = "3.0" + mock_rails_config.from_path.return_value = mock_config + + with pytest.raises(Exception, match="Invalid colang version"): + run_chat(config_path="test_config") + + @patch.object(chat_module, "console") + @patch("asyncio.run") + @patch.object(chat_module, "RailsConfig") + def test_run_chat_verbose_with_llm_calls( + self, mock_rails_config, mock_asyncio_run, mock_console + ): + mock_config = MagicMock() + mock_config.colang_version = "1.0" + mock_rails_config.from_path.return_value = mock_config + + run_chat(config_path="test_config", verbose=True, verbose_llm_calls=True) + + mock_console.print.assert_any_call( + "NOTE: use the `--verbose-no-llm` option to exclude the LLM prompts " + "and completions from the log.\n" + ) + + +class TestRunChatV1Async: + @pytest.mark.asyncio + async def test_run_chat_v1_no_config_no_server(self): + from nemoguardrails.cli.chat import _run_chat_v1_0 + + with pytest.raises(RuntimeError, match="At least one of"): + await _run_chat_v1_0(config_path=None, server_url=None) + + @pytest.mark.asyncio + @patch("builtins.input") + @patch.object(chat_module, "LLMRails") + @patch.object(chat_module, "RailsConfig") + async def test_run_chat_v1_local_config( + self, mock_rails_config, mock_llm_rails, mock_input + ): + from nemoguardrails.cli.chat import _run_chat_v1_0 + + mock_config = MagicMock() + mock_config.streaming_supported = False + mock_rails_config.from_path.return_value = mock_config + + mock_rails = AsyncMock() + mock_rails.generate_async = AsyncMock( + return_value={"role": "assistant", "content": "Hello!"} + ) + mock_rails.main_llm_supports_streaming = False + mock_llm_rails.return_value = mock_rails + + mock_input.side_effect = ["test message", KeyboardInterrupt()] + + try: + await _run_chat_v1_0(config_path="test_config") + except KeyboardInterrupt: + pass + + mock_rails.generate_async.assert_called_once() + + @pytest.mark.asyncio + @patch("builtins.input") + @patch.object(chat_module, "console") + @patch.object(chat_module, "LLMRails") + @patch.object(chat_module, "RailsConfig") + async def test_run_chat_v1_streaming_not_supported( + self, mock_rails_config, mock_llm_rails, mock_console, mock_input + ): + from nemoguardrails.cli.chat import _run_chat_v1_0 + + mock_config = MagicMock() + mock_config.streaming_supported = False + mock_rails_config.from_path.return_value = mock_config + + mock_rails = AsyncMock() + mock_llm_rails.return_value = mock_rails + + mock_input.side_effect = [KeyboardInterrupt()] + + try: + await _run_chat_v1_0(config_path="test_config", streaming=True) + except KeyboardInterrupt: + pass + + mock_console.print.assert_any_call( + "WARNING: The config `test_config` does not support streaming. " + "Falling back to normal mode." + ) + + @pytest.mark.asyncio + @patch("aiohttp.ClientSession") + @patch("builtins.input") + async def test_run_chat_v1_server_mode(self, mock_input, mock_client_session): + from nemoguardrails.cli.chat import _run_chat_v1_0 + + mock_session = AsyncMock() + mock_response = AsyncMock() + mock_response.headers = {} + mock_response.json = AsyncMock( + return_value={ + "messages": [{"role": "assistant", "content": "Server response"}] + } + ) + mock_response.__aenter__ = AsyncMock(return_value=mock_response) + mock_response.__aexit__ = AsyncMock() + + mock_post_context = AsyncMock() + mock_post_context.__aenter__ = AsyncMock(return_value=mock_response) + mock_post_context.__aexit__ = AsyncMock() + mock_session.post = MagicMock(return_value=mock_post_context) + + mock_client_session.return_value.__aenter__ = AsyncMock( + return_value=mock_session + ) + mock_client_session.return_value.__aexit__ = AsyncMock() + + mock_input.side_effect = ["test message", KeyboardInterrupt()] + + try: + await _run_chat_v1_0( + server_url="http://localhost:8000", config_id="test_id" + ) + except KeyboardInterrupt: + pass + + assert mock_session.post.called + call_args = mock_session.post.call_args + assert call_args[0][0] == "http://localhost:8000/v1/chat/completions" + assert "config_id" in call_args[1]["json"] + assert call_args[1]["json"]["config_id"] == "test_id" + assert call_args[1]["json"]["stream"] is False + + @pytest.mark.asyncio + @patch("aiohttp.ClientSession") + @patch("builtins.input") + async def test_run_chat_v1_server_streaming(self, mock_input, mock_client_session): + from nemoguardrails.cli.chat import _run_chat_v1_0 + + mock_session = AsyncMock() + mock_response = AsyncMock() + mock_response.headers = {"Transfer-Encoding": "chunked"} + + async def mock_iter_any(): + yield b"Stream " + yield b"response" + + mock_response.content.iter_any = mock_iter_any + mock_response.__aenter__ = AsyncMock(return_value=mock_response) + mock_response.__aexit__ = AsyncMock() + + mock_post_context = AsyncMock() + mock_post_context.__aenter__ = AsyncMock(return_value=mock_response) + mock_post_context.__aexit__ = AsyncMock() + mock_session.post = MagicMock(return_value=mock_post_context) + + mock_client_session.return_value.__aenter__ = AsyncMock( + return_value=mock_session + ) + mock_client_session.return_value.__aexit__ = AsyncMock() + + mock_input.side_effect = ["test message", KeyboardInterrupt()] + + try: + await _run_chat_v1_0( + server_url="http://localhost:8000", config_id="test_id", streaming=True + ) + except KeyboardInterrupt: + pass + + assert mock_session.post.called diff --git a/tests/cli/test_cli_main.py b/tests/cli/test_cli_main.py new file mode 100644 index 000000000..bf379e3c2 --- /dev/null +++ b/tests/cli/test_cli_main.py @@ -0,0 +1,336 @@ +# SPDX-FileCopyrightText: Copyright (c) 2023 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on 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 MagicMock, patch + +import pytest +from typer.testing import CliRunner + +from nemoguardrails import __version__ +from nemoguardrails.cli import app + +runner = CliRunner() + + +def test_app(): + result = runner.invoke( + app, + [ + "chat", + "--config=examples/rails/benefits_co/config.yml", + "--config=examples/rails/benefits_co/general.co", + ], + ) + assert result.exit_code == 1 + assert "not supported" in result.stdout + assert "Please provide a single" in result.stdout + + +class TestCLIVersion: + def test_version_flag(self): + result = runner.invoke(app, ["--version"]) + assert result.exit_code == 0 + assert __version__ in result.stdout + + def test_version_flag_short(self): + result = runner.invoke(app, ["-v"]) + assert result.exit_code == 0 + assert __version__ in result.stdout + + +class TestChatCommand: + def test_chat_with_multiple_configs_fails(self): + result = runner.invoke( + app, + [ + "chat", + "--config=config1.yml", + "--config=config2.yml", + ], + ) + assert result.exit_code == 1 + assert "Multiple configurations are not supported" in result.stdout + + @patch("nemoguardrails.cli.run_chat") + @patch("os.path.exists") + def test_chat_with_single_config(self, mock_exists, mock_run_chat): + mock_exists.return_value = True + result = runner.invoke(app, ["chat", "--config=test_config"]) + assert result.exit_code == 0 + mock_run_chat.assert_called_once() + + @patch("nemoguardrails.cli.run_chat") + @patch("os.path.exists") + def test_chat_with_verbose(self, mock_exists, mock_run_chat): + mock_exists.return_value = True + result = runner.invoke(app, ["chat", "--config=test_config", "--verbose"]) + assert result.exit_code == 0 + mock_run_chat.assert_called_once_with( + config_path="test_config", + verbose=True, + verbose_llm_calls=True, + streaming=False, + server_url=None, + config_id=None, + ) + + @patch("nemoguardrails.cli.run_chat") + @patch("os.path.exists") + def test_chat_with_verbose_no_llm(self, mock_exists, mock_run_chat): + mock_exists.return_value = True + result = runner.invoke( + app, ["chat", "--config=test_config", "--verbose-no-llm"] + ) + assert result.exit_code == 0 + mock_run_chat.assert_called_once_with( + config_path="test_config", + verbose=True, + verbose_llm_calls=False, + streaming=False, + server_url=None, + config_id=None, + ) + + @patch("nemoguardrails.cli.run_chat") + @patch("os.path.exists") + def test_chat_with_streaming(self, mock_exists, mock_run_chat): + mock_exists.return_value = True + result = runner.invoke(app, ["chat", "--config=test_config", "--streaming"]) + assert result.exit_code == 0 + mock_run_chat.assert_called_once_with( + config_path="test_config", + verbose=False, + verbose_llm_calls=True, + streaming=True, + server_url=None, + config_id=None, + ) + + @patch("nemoguardrails.cli.run_chat") + def test_chat_with_server_url(self, mock_run_chat): + result = runner.invoke( + app, + [ + "chat", + "--server-url=http://localhost:8000", + "--config-id=test_id", + ], + ) + assert result.exit_code == 0 + mock_run_chat.assert_called_once_with( + config_path="config", + verbose=False, + verbose_llm_calls=True, + streaming=False, + server_url="http://localhost:8000", + config_id="test_id", + ) + + @patch("nemoguardrails.cli.run_chat") + @patch("nemoguardrails.cli.init_random_seed") + @patch("os.path.exists") + def test_chat_with_debug_level(self, mock_exists, mock_init_seed, mock_run_chat): + mock_exists.return_value = True + result = runner.invoke( + app, ["chat", "--config=test_config", "--debug-level=DEBUG"] + ) + assert result.exit_code == 0 + mock_init_seed.assert_called_once_with(0) + mock_run_chat.assert_called_once() + + +class TestServerCommand: + @patch("uvicorn.run") + @patch("nemoguardrails.server.api.app") + def test_server_default_port(self, mock_app, mock_uvicorn): + result = runner.invoke(app, ["server"]) + assert result.exit_code == 0 + mock_uvicorn.assert_called_once() + call_args = mock_uvicorn.call_args + assert call_args[1]["port"] == 8000 + assert call_args[1]["host"] == "0.0.0.0" + + @patch("uvicorn.run") + @patch("nemoguardrails.server.api.app") + def test_server_custom_port(self, mock_app, mock_uvicorn): + result = runner.invoke(app, ["server", "--port=9000"]) + assert result.exit_code == 0 + mock_uvicorn.assert_called_once() + call_args = mock_uvicorn.call_args + assert call_args[1]["port"] == 9000 + + @patch("uvicorn.run") + @patch("nemoguardrails.server.api.app") + @patch("os.path.exists") + @patch("os.path.expanduser") + def test_server_with_config( + self, mock_expanduser, mock_exists, mock_app, mock_uvicorn + ): + mock_expanduser.return_value = "/path/to/config" + mock_exists.return_value = True + result = runner.invoke(app, ["server", "--config=/path/to/config"]) + assert result.exit_code == 0 + assert mock_app.rails_config_path == "/path/to/config" + + @patch("uvicorn.run") + @patch("nemoguardrails.server.api.app") + @patch("os.path.exists") + @patch("os.getcwd") + def test_server_with_local_config( + self, mock_getcwd, mock_exists, mock_app, mock_uvicorn + ): + mock_getcwd.return_value = "/current/dir" + mock_exists.return_value = True + result = runner.invoke(app, ["server"]) + assert result.exit_code == 0 + assert mock_app.rails_config_path == "/current/dir/config" + + @patch("uvicorn.run") + @patch("nemoguardrails.server.api.app") + def test_server_with_disable_chat_ui(self, mock_app, mock_uvicorn): + result = runner.invoke(app, ["server", "--disable-chat-ui"]) + assert result.exit_code == 0 + assert mock_app.disable_chat_ui is True + + @patch("uvicorn.run") + @patch("nemoguardrails.server.api.app") + def test_server_with_auto_reload(self, mock_app, mock_uvicorn): + result = runner.invoke(app, ["server", "--auto-reload"]) + assert result.exit_code == 0 + assert mock_app.auto_reload is True + + @patch("uvicorn.run") + @patch("nemoguardrails.server.api.app") + @patch("nemoguardrails.server.api.set_default_config_id") + def test_server_with_default_config_id( + self, mock_set_default, mock_app, mock_uvicorn + ): + result = runner.invoke(app, ["server", "--default-config-id=test_config"]) + assert result.exit_code == 0 + mock_set_default.assert_called_once_with("test_config") + + @patch("uvicorn.run") + @patch("nemoguardrails.server.api.app") + def test_server_with_prefix(self, mock_app, mock_uvicorn): + from fastapi import FastAPI + + with patch.object(FastAPI, "mount") as mock_mount: + result = runner.invoke(app, ["server", "--prefix=/api/v1"]) + assert result.exit_code == 0 + mock_mount.assert_called_once() + + +class TestConvertCommand: + def test_convert_missing_path(self): + result = runner.invoke(app, ["convert"]) + assert result.exit_code != 0 + + @patch("os.path.abspath") + @patch("nemoguardrails.cli.migrate") + def test_convert_with_defaults(self, mock_migrate, mock_abspath): + mock_abspath.return_value = "/abs/path/to/file" + result = runner.invoke(app, ["convert", "test_file.co"]) + assert result.exit_code == 0 + mock_migrate.assert_called_once_with( + path="/abs/path/to/file", + include_main_flow=True, + use_active_decorator=True, + from_version="1.0", + validate=False, + ) + + @patch("os.path.abspath") + @patch("nemoguardrails.cli.migrate") + def test_convert_with_options(self, mock_migrate, mock_abspath): + mock_abspath.return_value = "/abs/path/to/file" + result = runner.invoke( + app, + [ + "convert", + "test_file.co", + "--from-version=2.0-alpha", + "--validate", + "--no-use-active-decorator", + "--no-include-main-flow", + ], + ) + assert result.exit_code == 0 + mock_migrate.assert_called_once_with( + path="/abs/path/to/file", + include_main_flow=False, + use_active_decorator=False, + from_version="2.0-alpha", + validate=True, + ) + + @patch("nemoguardrails.cli.migrate") + @patch("os.path.abspath") + @patch("logging.getLogger") + def test_convert_with_verbose(self, mock_logger, mock_abspath, mock_migrate): + mock_abspath.return_value = "/abs/path/to/file" + mock_logger_instance = MagicMock() + mock_logger.return_value = mock_logger_instance + result = runner.invoke(app, ["convert", "test_file.co", "--verbose"]) + assert result.exit_code == 0 + mock_logger_instance.setLevel.assert_called() + + +class TestActionsServerCommand: + @patch("uvicorn.run") + def test_actions_server_default_port(self, mock_uvicorn): + result = runner.invoke(app, ["actions-server"]) + assert result.exit_code == 0 + mock_uvicorn.assert_called_once() + call_args = mock_uvicorn.call_args + assert call_args[1]["port"] == 8001 + assert call_args[1]["host"] == "0.0.0.0" + + @patch("uvicorn.run") + def test_actions_server_custom_port(self, mock_uvicorn): + result = runner.invoke(app, ["actions-server", "--port=9001"]) + assert result.exit_code == 0 + mock_uvicorn.assert_called_once() + call_args = mock_uvicorn.call_args + assert call_args[1]["port"] == 9001 + + +class TestFindProvidersCommand: + @patch("nemoguardrails.cli._list_providers") + def test_find_providers_list_only(self, mock_list_providers): + result = runner.invoke(app, ["find-providers", "--list"]) + assert result.exit_code == 0 + mock_list_providers.assert_called_once() + + @patch("nemoguardrails.cli.select_provider_with_type") + def test_find_providers_interactive(self, mock_select): + mock_select.return_value = ("chat completion", "openai") + result = runner.invoke(app, ["find-providers"]) + assert result.exit_code == 0 + assert "Selected chat completion provider: openai" in result.stdout + + @patch("nemoguardrails.cli.select_provider_with_type") + def test_find_providers_no_selection(self, mock_select): + mock_select.return_value = None + result = runner.invoke(app, ["find-providers"]) + assert result.exit_code == 0 + assert "No provider selected" in result.stdout + + +class TestEvalCommand: + def test_eval_command_exists(self): + result = runner.invoke(app, ["eval", "--help"]) + assert result.exit_code == 0 + assert "run" in result.stdout + assert "check-compliance" in result.stdout diff --git a/tests/cli/test_debugger.py b/tests/cli/test_debugger.py new file mode 100644 index 000000000..b0b0bcc97 --- /dev/null +++ b/tests/cli/test_debugger.py @@ -0,0 +1,233 @@ +# SPDX-FileCopyrightText: Copyright (c) 2023 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on 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 MagicMock, patch + +import pytest +from typer.testing import CliRunner + +from nemoguardrails.cli import debugger +from nemoguardrails.cli.chat import ChatState +from nemoguardrails.colang.v2_x.runtime.flows import FlowConfig, FlowState, State + +runner = CliRunner() + + +class TestDebuggerSetup: + def test_set_chat_state(self): + chat_state = MagicMock() + debugger.set_chat_state(chat_state) + assert debugger.chat_state == chat_state + + def test_set_runtime(self): + runtime = MagicMock() + debugger.set_runtime(runtime) + assert debugger.runtime == runtime + + def test_set_output_state(self): + state = MagicMock() + debugger.set_output_state(state) + assert debugger.state == state + + +class TestDebuggerCommands: + def setup_method(self): + self.chat_state = ChatState() + self.state = MagicMock(spec=State) + self.runtime = MagicMock() + debugger.chat_state = self.chat_state + debugger.state = self.state + debugger.runtime = self.runtime + + def test_restart_command(self): + self.chat_state.state = MagicMock() + # FIXME: input_events: List[dict] = field(default_factory=list) + self.chat_state.input_events = ["event1", "event2"] + self.chat_state.first_time = False + + result = runner.invoke(debugger.app, ["restart"]) + + assert result.exit_code == 0 + assert self.chat_state.state is None + assert self.chat_state.input_events == [] + assert self.chat_state.first_time is True + + def test_pause_command(self): + self.chat_state.paused = False + + result = runner.invoke(debugger.app, ["pause"]) + + assert result.exit_code == 0 + assert self.chat_state.paused is True + + def test_resume_command(self): + self.chat_state.paused = True + + result = runner.invoke(debugger.app, ["resume"]) + + assert result.exit_code == 0 + assert self.chat_state.paused is False + + +class TestFlowCommand: + def setup_method(self): + self.state = MagicMock(spec=State) + self.chat_state = ChatState() + debugger.state = self.state + debugger.chat_state = self.chat_state + + @patch("nemoguardrails.cli.debugger.console") + def test_flow_command_with_flow_config(self, mock_console): + flow_config = MagicMock(spec=FlowConfig) + self.state.flow_configs = {"test_flow": flow_config} + self.state.flow_states = {} + + result = runner.invoke(debugger.app, ["flow", "test_flow"]) + + assert result.exit_code == 0 + mock_console.print.assert_called_once_with(flow_config) + + @patch("nemoguardrails.cli.debugger.console") + def test_flow_command_with_flow_instance(self, mock_console): + flow_instance = MagicMock(spec=FlowState) + flow_instance.__dict__ = {"uid": "flow_uid_123", "status": "active"} + self.state.flow_configs = {} + self.state.flow_states = {"flow_uid_123": flow_instance} + + result = runner.invoke(debugger.app, ["flow", "uid_123"]) + + assert result.exit_code == 0 + mock_console.print.assert_called_once_with(flow_instance.__dict__) + + @patch("nemoguardrails.cli.debugger.console") + def test_flow_command_not_found(self, mock_console): + self.state.flow_configs = {} + self.state.flow_states = {} + + result = runner.invoke(debugger.app, ["flow", "nonexistent"]) + + assert result.exit_code == 0 + mock_console.print.assert_called_once_with("Flow 'nonexistent' does not exist.") + + +class TestFlowsCommand: + def setup_method(self): + self.state = MagicMock(spec=State) + self.chat_state = ChatState() + debugger.state = self.state + debugger.chat_state = self.chat_state + + @patch("nemoguardrails.cli.debugger.console") + @patch("nemoguardrails.cli.debugger.is_active_flow") + def test_flows_command_active_only(self, mock_is_active, mock_console): + flow_config = MagicMock(spec=FlowConfig) + flow_config.loop_priority = 1 + flow_config.loop_type = MagicMock(value="default") + flow_config.source_file = "/path/to/nemoguardrails/flow.py" + + flow_instance = MagicMock(spec=FlowState) + flow_instance.uid = "(flow)12345" + + self.state.flow_configs = {"test_flow": flow_config} + self.state.flow_id_states = {"test_flow": [flow_instance]} + + mock_is_active.return_value = True + + result = runner.invoke(debugger.app, ["flows"]) + + assert result.exit_code == 0 + mock_console.print.assert_called() + + @patch("nemoguardrails.cli.debugger.console") + def test_flows_command_all_flows(self, mock_console): + flow_config = MagicMock(spec=FlowConfig) + flow_config.loop_priority = 1 + flow_config.loop_type = MagicMock(value="default") + flow_config.source_file = "/path/to/flow.py" + + self.state.flow_configs = {"test_flow": flow_config} + self.state.flow_id_states = {} + + result = runner.invoke(debugger.app, ["flows", "--all"]) + + assert result.exit_code == 0 + mock_console.print.assert_called() + + @patch("nemoguardrails.cli.debugger.console") + def test_flows_command_order_by_name(self, mock_console): + flow_config_a = MagicMock(spec=FlowConfig) + flow_config_a.loop_priority = 2 + flow_config_a.loop_type = MagicMock(value="default") + flow_config_a.source_file = None + + flow_config_b = MagicMock(spec=FlowConfig) + flow_config_b.loop_priority = 1 + flow_config_b.loop_type = MagicMock(value="default") + flow_config_b.source_file = None + + self.state.flow_configs = {"flow_b": flow_config_b, "flow_a": flow_config_a} + self.state.flow_id_states = {} + + result = runner.invoke(debugger.app, ["flows", "--all", "--order-by-name"]) + + assert result.exit_code == 0 + mock_console.print.assert_called() + + +class TestTreeCommand: + def setup_method(self): + self.state = MagicMock(spec=State) + self.runtime = MagicMock() + debugger.state = self.state + debugger.runtime = self.runtime + + @patch("nemoguardrails.cli.debugger.Tree") + @patch("nemoguardrails.cli.debugger.console") + @patch("nemoguardrails.cli.debugger.is_active_flow") + def test_tree_command(self, mock_is_active, mock_console, mock_tree): + # create a main flow (required by tree command) + main_flow = MagicMock(spec=FlowState) + main_flow.uid = "main_flow" + main_flow.flow_id = "main" + main_flow.head = MagicMock() + main_flow.head.elements = [] + main_flow.child_flow_uids = [] + + # create mock flow configs with proper elements + main_config = MagicMock() + main_config.elements = [] + + self.state.flow_states = {"main_flow": main_flow} + self.state.flow_id_states = {"main": [main_flow]} + self.state.flow_configs = {"main": main_config} + mock_is_active.return_value = True + + # create a mock tree object + mock_tree_instance = MagicMock() + mock_tree.return_value = mock_tree_instance + + result = runner.invoke(debugger.app, ["tree"]) + + assert result.exit_code == 0 + mock_tree.assert_called_with("main") + + +# TODO: addd tests +# class TestVarsCommand: +# def setup_method(self): +# self.state = MagicMock(spec=State) +# self.runtime = MagicMock() +# debugger.state = self.state +# debugger.runtime = self.runtime diff --git a/tests/cli/test_llm_providers.py b/tests/cli/test_llm_providers.py new file mode 100644 index 000000000..74ac6f278 --- /dev/null +++ b/tests/cli/test_llm_providers.py @@ -0,0 +1,300 @@ +# SPDX-FileCopyrightText: Copyright (c) 2023 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on 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 MagicMock, patch + +import pytest +from typer.testing import CliRunner + +from nemoguardrails.cli import app +from nemoguardrails.cli.providers import ( + _get_provider_completions, + _list_providers, + find_providers, + select_provider, + select_provider_type, + select_provider_with_type, +) + +runner = CliRunner() + + +class TestListProviders: + @patch("nemoguardrails.cli.providers.console") + @patch("nemoguardrails.cli.providers.get_chat_provider_names") + @patch("nemoguardrails.cli.providers.get_llm_provider_names") + def test_list_providers( + self, mock_llm_providers, mock_chat_providers, mock_console + ): + mock_llm_providers.return_value = ["llm_provider_1", "llm_provider_2"] + mock_chat_providers.return_value = ["chat_provider_1", "chat_provider_2"] + + _list_providers() + + assert mock_console.print.call_count >= 4 + mock_console.print.assert_any_call("\n[bold]Text Completion Providers:[/]") + mock_console.print.assert_any_call(" • llm_provider_1") + mock_console.print.assert_any_call(" • llm_provider_2") + mock_console.print.assert_any_call("\n[bold]Chat Completion Providers:[/]") + mock_console.print.assert_any_call(" • chat_provider_1") + mock_console.print.assert_any_call(" • chat_provider_2") + + +class TestGetProviderCompletions: + @patch("nemoguardrails.cli.providers.get_llm_provider_names") + def test_get_text_completion_providers(self, mock_llm_providers): + mock_llm_providers.return_value = ["provider_b", "provider_a"] + + result = _get_provider_completions("text completion") + + assert result == ["provider_a", "provider_b"] + mock_llm_providers.assert_called_once() + + @patch("nemoguardrails.cli.providers.get_chat_provider_names") + def test_get_chat_completion_providers(self, mock_chat_providers): + mock_chat_providers.return_value = ["chat_b", "chat_a"] + + result = _get_provider_completions("chat completion") + + assert result == ["chat_a", "chat_b"] + mock_chat_providers.assert_called_once() + + def test_get_providers_invalid_type(self): + result = _get_provider_completions("invalid_type") + assert result == [] + + def test_get_providers_no_type(self): + result = _get_provider_completions(None) + assert result == [] + + +class TestSelectProviderType: + @patch("nemoguardrails.cli.providers.console") + @patch("nemoguardrails.cli.providers.PromptSession") + def test_select_provider_type_exact_match(self, mock_prompt_session, mock_console): + mock_session = MagicMock() + mock_session.prompt.return_value = "chat completion" + mock_prompt_session.return_value = mock_session + + result = select_provider_type() + + assert result == "chat completion" + mock_session.prompt.assert_called_once() + + @patch("nemoguardrails.cli.providers.console") + @patch("nemoguardrails.cli.providers.PromptSession") + def test_select_provider_type_fuzzy_match(self, mock_prompt_session, mock_console): + mock_session = MagicMock() + mock_session.prompt.return_value = "chat" + mock_prompt_session.return_value = mock_session + + result = select_provider_type() + + assert result == "chat completion" + + @patch("nemoguardrails.cli.providers.console") + @patch("nemoguardrails.cli.providers.PromptSession") + def test_select_provider_type_empty_input(self, mock_prompt_session, mock_console): + mock_session = MagicMock() + mock_session.prompt.return_value = "" + mock_prompt_session.return_value = mock_session + + result = select_provider_type() + + assert result is None + + @patch("nemoguardrails.cli.providers.console") + @patch("nemoguardrails.cli.providers.PromptSession") + def test_select_provider_type_ambiguous_match( + self, mock_prompt_session, mock_console + ): + mock_session = MagicMock() + mock_session.prompt.return_value = "completion" + mock_prompt_session.return_value = mock_session + + result = select_provider_type() + + assert result is None + + @patch("nemoguardrails.cli.providers.console") + @patch("nemoguardrails.cli.providers.PromptSession") + def test_select_provider_type_keyboard_interrupt( + self, mock_prompt_session, mock_console + ): + mock_session = MagicMock() + mock_session.prompt.side_effect = KeyboardInterrupt() + mock_prompt_session.return_value = mock_session + + result = select_provider_type() + + assert result is None + + @patch("nemoguardrails.cli.providers.console") + @patch("nemoguardrails.cli.providers.PromptSession") + def test_select_provider_type_eof_error(self, mock_prompt_session, mock_console): + mock_session = MagicMock() + mock_session.prompt.side_effect = EOFError() + mock_prompt_session.return_value = mock_session + + result = select_provider_type() + + assert result is None + + +class TestSelectProvider: + @patch("nemoguardrails.cli.providers.console") + @patch("nemoguardrails.cli.providers.PromptSession") + @patch("nemoguardrails.cli.providers._get_provider_completions") + def test_select_provider_exact_match( + self, mock_get_completions, mock_prompt_session, mock_console + ): + mock_get_completions.return_value = ["openai", "anthropic", "azure"] + mock_session = MagicMock() + mock_session.prompt.return_value = "openai" + mock_prompt_session.return_value = mock_session + + result = select_provider("chat completion") + + assert result == "openai" + + @patch("nemoguardrails.cli.providers.console") + @patch("nemoguardrails.cli.providers.PromptSession") + @patch("nemoguardrails.cli.providers._get_provider_completions") + def test_select_provider_fuzzy_match( + self, mock_get_completions, mock_prompt_session, mock_console + ): + mock_get_completions.return_value = ["openai", "anthropic", "azure"] + mock_session = MagicMock() + mock_session.prompt.return_value = "open" + mock_prompt_session.return_value = mock_session + + result = select_provider("chat completion") + + assert result == "openai" + + @patch("nemoguardrails.cli.providers.console") + @patch("nemoguardrails.cli.providers.PromptSession") + @patch("nemoguardrails.cli.providers._get_provider_completions") + def test_select_provider_empty_input( + self, mock_get_completions, mock_prompt_session, mock_console + ): + mock_get_completions.return_value = ["openai", "anthropic"] + mock_session = MagicMock() + mock_session.prompt.return_value = "" + mock_prompt_session.return_value = mock_session + + result = select_provider("chat completion") + + assert result is None + + @patch("nemoguardrails.cli.providers.console") + @patch("nemoguardrails.cli.providers.PromptSession") + @patch("nemoguardrails.cli.providers._get_provider_completions") + def test_select_provider_no_match( + self, mock_get_completions, mock_prompt_session, mock_console + ): + mock_get_completions.return_value = ["openai", "anthropic"] + mock_session = MagicMock() + mock_session.prompt.return_value = "invalid_provider" + mock_prompt_session.return_value = mock_session + + result = select_provider("chat completion") + + assert result is None + + @patch("nemoguardrails.cli.providers.console") + @patch("nemoguardrails.cli.providers.PromptSession") + @patch("nemoguardrails.cli.providers._get_provider_completions") + def test_select_provider_keyboard_interrupt( + self, mock_get_completions, mock_prompt_session, mock_console + ): + mock_get_completions.return_value = ["openai"] + mock_session = MagicMock() + mock_session.prompt.side_effect = KeyboardInterrupt() + mock_prompt_session.return_value = mock_session + + result = select_provider("chat completion") + + assert result is None + + +class TestSelectProviderWithType: + @patch("nemoguardrails.cli.providers.select_provider") + @patch("nemoguardrails.cli.providers.select_provider_type") + def test_select_provider_with_type_success( + self, mock_select_type, mock_select_provider + ): + mock_select_type.return_value = "chat completion" + mock_select_provider.return_value = "openai" + + result = select_provider_with_type() + + assert result == ("chat completion", "openai") + mock_select_type.assert_called_once() + mock_select_provider.assert_called_once_with("chat completion") + + @patch("nemoguardrails.cli.providers.select_provider_type") + def test_select_provider_with_type_no_type(self, mock_select_type): + mock_select_type.return_value = None + + result = select_provider_with_type() + + assert result is None + + @patch("nemoguardrails.cli.providers.select_provider") + @patch("nemoguardrails.cli.providers.select_provider_type") + def test_select_provider_with_type_no_provider( + self, mock_select_type, mock_select_provider + ): + mock_select_type.return_value = "chat completion" + mock_select_provider.return_value = None + + result = select_provider_with_type() + + assert result is None + + +class TestFindProvidersCommand: + @patch("nemoguardrails.cli.providers._list_providers") + def test_find_providers_list_only_as_function(self, mock_list_providers): + find_providers(list_only=True) + mock_list_providers.assert_called_once() + + @patch("nemoguardrails.cli.providers.select_provider_with_type") + @patch("typer.echo") + def test_find_providers_interactive_success(self, mock_echo, mock_select): + mock_select.return_value = ("chat completion", "openai") + find_providers(list_only=False) + mock_echo.assert_called_with("\nSelected chat completion provider: openai") + + @patch("nemoguardrails.cli.providers.select_provider_with_type") + @patch("typer.echo") + def test_find_providers_interactive_no_selection(self, mock_echo, mock_select): + mock_select.return_value = None + find_providers(list_only=False) + mock_echo.assert_called_with("No provider selected.") + + def test_find_providers_cli_list(self): + with patch("nemoguardrails.cli._list_providers") as mock_list: + result = runner.invoke(app, ["find-providers", "--list"]) + assert result.exit_code == 0 + mock_list.assert_called_once() + + def test_find_providers_cli_interactive(self): + with patch("nemoguardrails.cli.select_provider_with_type") as mock_select: + mock_select.return_value = ("text completion", "llama") + result = runner.invoke(app, ["find-providers"]) + assert result.exit_code == 0 + assert "Selected text completion provider: llama" in result.stdout diff --git a/tests/cli/test_migration.py b/tests/cli/test_migration.py new file mode 100644 index 000000000..867698469 --- /dev/null +++ b/tests/cli/test_migration.py @@ -0,0 +1,686 @@ +# SPDX-FileCopyrightText: Copyright (c) 2023 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on 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. + +import os +import tempfile +import textwrap +from pathlib import Path +from unittest.mock import MagicMock, mock_open, patch + +import pytest + +from nemoguardrails.cli.migration import ( + _confirm_and_tag_replace, + _get_co_files_to_process, + _get_config_files_to_process, + _globalize_variable_assignment, + _is_anonymous_flow, + _is_flow, + _process_co_files, + _process_config_files, + _revise_anonymous_flow, + _write_to_file, + _write_transformed_content_and_rename_original, + convert_colang_1_syntax, + convert_colang_2alpha_syntax, + migrate, +) + + +class TestColang2AlphaSyntaxConversion: + def test_orwhen_replacement(self): + input_lines = ["orwhen condition met"] + expected_output = ["or when condition met"] + assert convert_colang_2alpha_syntax(input_lines) == expected_output + + def test_flow_start_uid_replacement(self): + input_lines = ["flow_start_uid: 12345"] + expected_output = ["flow_instance_uid: 12345"] + assert convert_colang_2alpha_syntax(input_lines) == expected_output + + def test_regex_replacement_1(self): + input_lines = ['r"(?i).*({{$text}})((\s*\w+\s*){0,2})\W*$"'] + expected_output = ['regex("((?i).*({{$text}})((\\s*\\w+\\s*){0,2})\\W*$)")'] + assert convert_colang_2alpha_syntax(input_lines) == expected_output + + def test_regex_replacement_2(self): + input_lines = ["r'(?i).*({{$text}})((\s*\w+\s*){0,2})\W*$'"] + expected_output = ["regex('((?i).*({{$text}})((\\s*\\w+\\s*){0,2})\\W*$)')"] + assert convert_colang_2alpha_syntax(input_lines) == expected_output + + def test_curly_braces_replacement(self): + input_lines = ['"{{variable}}"'] + expected_output = ['"{variable}"'] + assert convert_colang_2alpha_syntax(input_lines) == expected_output + + def test_findall_replacement(self): + input_lines = ["findall matches"] + expected_output = ["find_all matches"] + assert convert_colang_2alpha_syntax(input_lines) == expected_output + + def test_triple_quotes_replacement_1(self): + input_lines = ['$ """some text"""'] + expected_output = ['$ ..."some text"'] + assert convert_colang_2alpha_syntax(input_lines) == expected_output + + def test_triple_quotes_replacement_2(self): + input_lines = ["$ '''some text'''"] + expected_output = ["$ ...'some text'"] + assert convert_colang_2alpha_syntax(input_lines) == expected_output + + def test_specific_phrases_replacement(self): + input_lines = [ + "catch colang errors", + "catch undefined flows", + "catch unexpected user utterance", + "track bot talking state", + "track user talking state", + "track user utterance state", + "poll llm request response", + "trigger user intent for unhandled user utterance", + "generate then continue interaction", + "track unhandled user intent state", + "respond to unhandled user intent", + ] + expected_output = [ + "catch colang errors", + "notification of undefined flow start", + "notification of unexpected user utterance", + "tracking bot talking state", + "tracking user talking state", + "tracking user talking state", + "polling llm request response", + "generating user intent for unhandled user utterance", + "llm continue interaction", + "tracking unhandled user intent state", + "continuation on unhandled user intent", + ] + assert convert_colang_2alpha_syntax(input_lines) == expected_output + + def test_meta_decorator_replacement(self): + input_lines = [ + "flow some_flow:", + "# meta: loop_id=123", + "# meta: example meta", + "some action", + ] + expected_output = [ + '@loop("123")', + "@meta(example_meta=True)", + "flow some_flow:", + "some action", + ] + assert convert_colang_2alpha_syntax(input_lines) == expected_output + + def test_convert_flow_examples(self): + input_1 = """ + flow bot inform something like issue + # meta: bot intent + (bot inform "ABC" + or bot inform "DEFG" + or bot inform "HJKL") + and (bot gesture "abc def" or bot gesture "hij kl") + """ + input_lines = textwrap.dedent(input_1).strip().split("\n") + + output_1 = """ + @meta(bot_intent=True) + flow bot inform something like issue + (bot inform "ABC" + or bot inform "DEFG" + or bot inform "HJKL") + and (bot gesture "abc def" or bot gesture "hij kl") + """ + output_lines = textwrap.dedent(output_1).strip().split("\n") + + assert convert_colang_2alpha_syntax(input_lines) == output_lines + + +class TestColang1SyntaxConversion: + def test_define_flow_conversion(self): + input_lines = ["define flow express greeting"] + expected_output = ["flow express greeting"] + assert convert_colang_1_syntax(input_lines) == expected_output + + def test_define_subflow_conversion(self): + input_lines = ["define subflow my_subflow"] + expected_output = ["flow my_subflow"] + assert convert_colang_1_syntax(input_lines) == expected_output + + def test_execute_to_await_and_pascal_case_action(self): + input_lines = ["execute some_action"] + expected_output = ["await SomeAction"] + assert convert_colang_1_syntax(input_lines) == expected_output + + def test_stop_to_abort(self): + input_lines = ["stop"] + expected_output = ["abort"] + assert convert_colang_1_syntax(input_lines) == expected_output + + def test_anonymous_flow_revised(self): + input_lines = ["flow", "user said hello"] + # because the flow is anonymous and only 'flow' is given, it will be converted to 'flow said hello' based on the first message + expected_output = ["flow said hello", "user said hello"] + output = convert_colang_1_syntax(input_lines) + # strip newline characters from the strings in the output list + output = [line.rstrip("\n") for line in output] + assert output == expected_output + + def test_global_variable_assignment(self): + input_lines = ["$variable = value"] + expected_output = ["global $variable\n$variable = value"] + assert convert_colang_1_syntax(input_lines) == expected_output + + def test_variable_assignment_in_await(self): + input_lines = ["$result = await some_action"] + expected_output = ["$result = await SomeAction"] + assert convert_colang_1_syntax(input_lines) == expected_output + + def test_bot_say_conversion(self): + input_lines = ["define bot", '"Hello!"', '"How can I help you?"'] + expected_output = [ + "flow bot", + 'bot say "Hello!"', + 'or bot say "How can I help you?"', + ] + assert convert_colang_1_syntax(input_lines) == expected_output + + def test_user_said_conversion(self): + input_lines = ["define user", '"I need assistance."', '"Can you help me?"'] + expected_output = [ + "flow user", + 'user said "I need assistance."', + 'or user said "Can you help me?"', + ] + assert convert_colang_1_syntax(input_lines) == expected_output + + def test_create_event_to_send(self): + input_lines = [" create event user_asked_question"] + expected_output = [" send user_asked_question"] + assert convert_colang_1_syntax(input_lines) == expected_output + + def test_config_variable_replacement(self): + input_lines = ["$config.setting = true"] + expected_output = [ + "global $system.config.setting\n$system.config.setting = true" + ] + assert convert_colang_1_syntax(input_lines) == expected_output + + def test_flow_with_special_characters(self): + input_lines = ["define flow my-flow's_test"] + expected_output = ["flow my flow s_test"] + assert convert_colang_1_syntax(input_lines) == expected_output + + def test_ellipsis_variable_assignment(self): + input_lines = ["# User's name", "$name = ...", "await greet_user"] + expected_output = [ + "# User's name", + "global $name\n$name = ...", + "await GreetUserAction", + ] + + expected_output = [ + "# User's name", + 'global $name\n$name = ... "User\'s name"', + "await GreetUserAction", + ] + assert convert_colang_1_syntax(input_lines) == expected_output + + @pytest.mark.skip("not implemented conversion") + def test_complex_conversion(self): + # TODO: add bot $response to bot say $response conversion + input_script = """ + define flow greeting_flow + when user express greeting + $response = execute generate_greeting + bot $response + """ + expected_output_script = """ + flow greeting_flow + when user express greeting + $response = await GenerateGreetingAction + bot say $response + """ + input_lines = textwrap.dedent(input_script).strip().split("\n") + expected_output = textwrap.dedent(expected_output_script).strip().split("\n") + + print(convert_colang_1_syntax(input_lines)) + assert convert_colang_1_syntax(input_lines) == expected_output + + def test_flow_with_execute_and_stop(self): + input_lines = [ + "define flow sample_flow", + ' when user "Cancel"', + " execute cancel_operation", + " stop", + ] + expected_output = [ + "flow sample_flow", + ' when user "Cancel"', + " await CancelOperationAction", + " abort", + ] + assert convert_colang_1_syntax(input_lines) == expected_output + + def test_await_camelcase_conversion(self): + input_lines = ["await sample_action"] + expected_output = ["await SampleAction"] + assert convert_colang_1_syntax(input_lines) == expected_output + + def test_nested_flow_conversion(self): + input_script = """ + define flow outer_flow + when condition_met + define subflow inner_flow + execute inner_action + """ + expected_output_script = """ + flow outer_flow + when condition_met + flow inner_flow + await InnerAction + """ + input_lines = textwrap.dedent(input_script).strip().split("\n") + expected_output = textwrap.dedent(expected_output_script).strip().split("\n") + assert convert_colang_1_syntax(input_lines) == expected_output + + +class TestMigrateFunction: + @patch("nemoguardrails.cli.migration._process_config_files") + @patch("nemoguardrails.cli.migration._process_co_files") + @patch("nemoguardrails.cli.migration._get_config_files_to_process") + @patch("nemoguardrails.cli.migration._get_co_files_to_process") + @patch("nemoguardrails.cli.migration.console") + def test_migrate_with_defaults( + self, + mock_console, + mock_get_co_files, + mock_get_config_files, + mock_process_co, + mock_process_config, + ): + mock_get_co_files.return_value = ["file1.co", "file2.co"] + mock_get_config_files.return_value = ["config.yml"] + mock_process_co.return_value = 2 + mock_process_config.return_value = 1 + + migrate("/test/path") + + mock_console.print.assert_any_call( + "Starting migration for path: /test/path from version 1.0 to latest version." + ) + mock_process_co.assert_called_once_with( + ["file1.co", "file2.co"], "1.0", False, True, True + ) + mock_process_config.assert_called_once_with(["config.yml"]) + + @patch("nemoguardrails.cli.migration._process_config_files") + @patch("nemoguardrails.cli.migration._process_co_files") + @patch("nemoguardrails.cli.migration._get_config_files_to_process") + @patch("nemoguardrails.cli.migration._get_co_files_to_process") + @patch("nemoguardrails.cli.migration.console") + def test_migrate_from_2_0_alpha( + self, + mock_console, + mock_get_co_files, + mock_get_config_files, + mock_process_co, + mock_process_config, + ): + mock_get_co_files.return_value = [] + mock_get_config_files.return_value = [] + mock_process_co.return_value = 0 + mock_process_config.return_value = 0 + + migrate("/test/path", from_version="2.0-alpha", include_main_flow=True) + + mock_process_co.assert_called_once_with([], "2.0-alpha", True, True, True) + + +class TestGetFilesToProcess: + def test_get_co_files_from_directory(self): + with tempfile.TemporaryDirectory() as tmpdir: + Path(tmpdir, "file1.co").touch() + Path(tmpdir, "file2.co").touch() + Path(tmpdir, "file3.txt").touch() + Path(tmpdir, "subdir").mkdir() + Path(tmpdir, "subdir", "file4.co").touch() + + files = _get_co_files_to_process(tmpdir) + + assert len(files) == 3 + assert all(str(f).endswith(".co") for f in files) + + def test_get_co_files_single_file(self): + with tempfile.TemporaryDirectory() as tmpdir: + test_file = Path(tmpdir, "test.co") + test_file.touch() + + files = _get_co_files_to_process(str(test_file)) + + assert len(files) == 1 + assert str(files[0]) == str(test_file) + + def test_get_config_files_from_directory(self): + with tempfile.TemporaryDirectory() as tmpdir: + Path(tmpdir, "config.yml").touch() + Path(tmpdir, "config.yaml").touch() + Path(tmpdir, "other.txt").touch() + + files = _get_config_files_to_process(tmpdir) + + assert len(files) == 2 + assert all(str(f).endswith((".yml", ".yaml")) for f in files) + + def test_get_config_files_single_file(self): + with tempfile.TemporaryDirectory() as tmpdir: + test_file = Path(tmpdir, "config.yml") + test_file.touch() + + files = _get_config_files_to_process(tmpdir) + + assert len(files) == 1 + assert str(files[0]) == str(test_file) + + +class TestWriteFunctions: + def test_write_to_file_success(self): + with tempfile.NamedTemporaryFile(mode="w", delete=False) as f: + filename = f.name + + try: + lines = ["line 1\n", "line 2\n", "line 3\n"] + result = _write_to_file(filename, lines) + + assert result is True + + with open(filename, "r") as f: + content = f.readlines() + assert content == lines + finally: + os.unlink(filename) + + @patch("builtins.open", side_effect=IOError("Permission denied")) + @patch("logging.error") + def test_write_to_file_failure(self, mock_log_error, mock_open_func): + result = _write_to_file("/invalid/path/file.txt", ["test"]) + + assert result is False + mock_log_error.assert_called_once() + + def test_write_transformed_content_and_rename_original(self): + with tempfile.TemporaryDirectory() as tmpdir: + original_file = Path(tmpdir, "test.co") + original_file.write_text("original content") + + new_lines = ["new line 1\n", "new line 2\n"] + result = _write_transformed_content_and_rename_original( + str(original_file), new_lines + ) + + assert result is True + assert original_file.exists() + assert original_file.read_text() == "new line 1\nnew line 2\n" + + renamed_file = Path(tmpdir, "test.v1.co") + assert renamed_file.exists() + assert renamed_file.read_text() == "original content" + + +class TestHelperFunctions: + def test_is_anonymous_flow(self): + assert _is_anonymous_flow("flow ") is True + assert _is_anonymous_flow("flow") is True + assert _is_anonymous_flow("flow my_flow") is False + assert _is_anonymous_flow(" flow ") is False + assert _is_anonymous_flow("not_flow") is False + + def test_is_flow(self): + assert _is_flow("flow my_flow") is True + assert _is_flow("flow test_flow_123") is True + assert _is_flow("flow") is False + assert _is_flow(" flow my_flow") is False + assert _is_flow("not_flow test") is False + + def test_globalize_variable_assignment(self): + result = _globalize_variable_assignment("$var = value") + assert "global $var" in result + + result = _globalize_variable_assignment("$result = await action") + assert "global" not in result + + result = _globalize_variable_assignment("$result = execute action") + assert "global" not in result + + result = _globalize_variable_assignment("regular = value") + assert "global regular" in result + assert "regular = value" in result + + def test_revise_anonymous_flow(self): + result = _revise_anonymous_flow("flow", "user said hello") + assert result == "flow said hello" + + result = _revise_anonymous_flow("flow", 'bot say "hi"') + assert "flow say" in result + + result = _revise_anonymous_flow("flow", "other content") + assert "flow other content" in result + + def test_confirm_and_tag_replace(self): + from nemoguardrails.cli import migration + + migration._LIBS_USED.clear() + + _confirm_and_tag_replace("new line", "old line", "core") + assert "core" in migration._LIBS_USED + + _confirm_and_tag_replace("same line", "same line", "llm") + assert "llm" not in migration._LIBS_USED + + +class TestProcessFiles: + @patch( + "nemoguardrails.cli.migration._write_transformed_content_and_rename_original" + ) + @patch("builtins.open", new_callable=mock_open, read_data="flow test\n") + @patch("nemoguardrails.cli.migration.console") + def test_process_co_files_v1_to_v2(self, mock_console, mock_file, mock_write): + mock_write.return_value = True + + files = [Path("test.co")] + result = _process_co_files(files, "1.0", False, True, False) + + assert result == 1 + mock_write.assert_called_once() + + @patch( + "nemoguardrails.cli.migration._write_transformed_content_and_rename_original" + ) + @patch("builtins.open", new_callable=mock_open, read_data="orwhen test\n") + @patch("nemoguardrails.cli.migration.console") + def test_process_co_files_v2_alpha_to_v2_beta( + self, mock_console, mock_file, mock_write + ): + mock_write.return_value = True + + files = [Path("test.co")] + result = _process_co_files(files, "2.0-alpha", False, True, False) + + assert result >= 0 + + @patch("nemoguardrails.cli.migration.parse_colang_file") + @patch( + "nemoguardrails.cli.migration._write_transformed_content_and_rename_original" + ) + @patch("builtins.open", new_callable=mock_open, read_data="flow test") + @patch("nemoguardrails.cli.migration.console") + def test_process_co_files_with_validation( + self, mock_console, mock_file, mock_write, mock_parse + ): + mock_write.return_value = True + mock_parse.return_value = {"flows": []} + + files = [Path("test.co")] + result = _process_co_files(files, "1.0", False, True, True) + + assert result == 1 + mock_parse.assert_called() + + @patch("builtins.open", new_callable=MagicMock) + @patch("nemoguardrails.cli.migration._process_sample_conversation_in_config") + @patch("nemoguardrails.cli.migration._comment_rails_flows_in_config") + @patch("nemoguardrails.cli.migration._get_rails_flows") + @patch("nemoguardrails.cli.migration._get_raw_config") + @patch("nemoguardrails.cli.migration.console") + def test_process_config_files( + self, + mock_console, + mock_get_config, + mock_get_rails, + mock_comment_rails, + mock_process_sample, + mock_open, + ): + mock_get_config.return_value = {"colang_version": "1.0"} + mock_get_rails.return_value = {} # No rails flows to process + + # Mock the file operations for _set_colang_version + mock_file = MagicMock() + mock_file.readlines.return_value = ["some_key: value\n"] + mock_file.__enter__ = MagicMock(return_value=mock_file) + mock_file.__exit__ = MagicMock(return_value=None) + mock_open.return_value = mock_file + + files = ["config.yml"] + result = _process_config_files(files) + + # The function returns 0 because there are no rails flows to write + assert result == 0 + # Verify the colang version setting was attempted + assert mock_open.called + + +class TestColang2AlphaSyntaxAdvanced: + def test_regex_conversion_complex(self): + input_lines = [ + 'r"\\d{3}-\\d{3}-\\d{4}"', + "r'[a-zA-Z0-9]+@[a-zA-Z0-9]+\\.[a-zA-Z]{2,}'", + ] + expected_output = [ + 'regex("(\\d{3}-\\d{3}-\\d{4})")', + "regex('([a-zA-Z0-9]+@[a-zA-Z0-9]+\\.[a-zA-Z]{2,})')", + ] + assert convert_colang_2alpha_syntax(input_lines) == expected_output + + def test_multiple_meta_decorators(self): + input_lines = [ + "flow test_flow:", + "# meta: loop_id=test_loop", + "# meta: bot intent", + "# meta: user visible", + " some content", + ] + expected_output = [ + '@loop("test_loop")', + "@meta(bot_intent=True)", + "@meta(user_visible=True)", + "flow test_flow:", + " some content", + ] + assert convert_colang_2alpha_syntax(input_lines) == expected_output + + def test_deprecated_flows_conversion(self): + input_lines = [ + "manage attentive posture", + "track user presence state", + "user became no longer present", + ] + expected_output = [ + "!!!! DEPRECATED FLOW (please remove): manage attentive posture", + "!!!! DEPRECATED FLOW (please remove): track user presence state", + "!!!! DEPRECATED FLOW (please remove): user became no longer present", + ] + assert convert_colang_2alpha_syntax(input_lines) == expected_output + + +class TestColang1SyntaxAdvanced: + def test_complex_flow_conversion(self): + input_lines = [ + "define flow user-request's_flow", + ' when user "help"', + " $response = execute get_help", + " stop", + ] + expected_output = [ + "flow user request s_flow", + ' when user "help"', + " $response = await GetHelpAction", + " abort", + ] + assert convert_colang_1_syntax(input_lines) == expected_output + + def test_variable_globalization_with_config(self): + input_lines = [ + "$config.timeout = 30", + "$config.retry_count = 3", + ] + result = convert_colang_1_syntax(input_lines) + assert "global $system.config.timeout" in result[0] + assert "$system.config.timeout = 30" in result[0] + assert "global $system.config.retry_count" in result[1] + assert "$system.config.retry_count = 3" in result[1] + + def test_bot_and_user_ellipsis_conversion(self): + input_lines = [ + "bot ...", + "user ...", + ] + expected_output = [ + "bot said something", + "user said something", + ] + assert convert_colang_1_syntax(input_lines) == expected_output + + def test_mixed_define_statements(self): + input_lines = [ + "define bot greeting", + '"Hello there!"', + '"Hi!"', + "define user request", + '"I need help"', + '"Can you assist me?"', + ] + expected_output = [ + "flow bot greeting", + 'bot say "Hello there!"', + 'or bot say "Hi!"', + "flow user request", + 'user said "I need help"', + 'or user said "Can you assist me?"', + ] + assert convert_colang_1_syntax(input_lines) == expected_output + + def test_await_action_camelcase_conversion(self): + input_lines = [ + "await fetch_user_data", + "await process_payment_info", + "await send_email_notification", + ] + expected_output = [ + "await FetchUserDataAction", + "await ProcessPaymentInfoAction", + "await SendEmailNotificationAction", + ] + assert convert_colang_1_syntax(input_lines) == expected_output diff --git a/tests/test_cli.py b/tests/test_cli.py deleted file mode 100644 index 6e321052f..000000000 --- a/tests/test_cli.py +++ /dev/null @@ -1,34 +0,0 @@ -# SPDX-FileCopyrightText: Copyright (c) 2023 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# SPDX-License-Identifier: Apache-2.0 -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on 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 typer.testing import CliRunner - -from nemoguardrails.cli import app - -runner = CliRunner() - - -def test_app(): - result = runner.invoke( - app, - [ - "chat", - "--config=examples/rails/benefits_co/config.yml", - "--config=examples/rails/benefits_co/general.co", - ], - ) - assert result.exit_code == 1 - assert "not supported" in result.stdout - assert "Please provide a single" in result.stdout diff --git a/tests/test_cli_migration.py b/tests/test_cli_migration.py deleted file mode 100644 index 7acf05ab8..000000000 --- a/tests/test_cli_migration.py +++ /dev/null @@ -1,282 +0,0 @@ -# SPDX-FileCopyrightText: Copyright (c) 2023 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# SPDX-License-Identifier: Apache-2.0 -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on 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. - -import textwrap - -import pytest - -from nemoguardrails.cli.migration import ( - convert_colang_1_syntax, - convert_colang_2alpha_syntax, -) - - -class TestColang2AlphaSyntaxConversion: - def test_orwhen_replacement(self): - input_lines = ["orwhen condition met"] - expected_output = ["or when condition met"] - assert convert_colang_2alpha_syntax(input_lines) == expected_output - - def test_flow_start_uid_replacement(self): - input_lines = ["flow_start_uid: 12345"] - expected_output = ["flow_instance_uid: 12345"] - assert convert_colang_2alpha_syntax(input_lines) == expected_output - - def test_regex_replacement_1(self): - input_lines = ['r"(?i).*({{$text}})((\s*\w+\s*){0,2})\W*$"'] - expected_output = ['regex("((?i).*({{$text}})((\\s*\\w+\\s*){0,2})\\W*$)")'] - assert convert_colang_2alpha_syntax(input_lines) == expected_output - - def test_regex_replacement_2(self): - input_lines = ["r'(?i).*({{$text}})((\s*\w+\s*){0,2})\W*$'"] - expected_output = ["regex('((?i).*({{$text}})((\\s*\\w+\\s*){0,2})\\W*$)')"] - assert convert_colang_2alpha_syntax(input_lines) == expected_output - - def test_curly_braces_replacement(self): - input_lines = ['"{{variable}}"'] - expected_output = ['"{variable}"'] - assert convert_colang_2alpha_syntax(input_lines) == expected_output - - def test_findall_replacement(self): - input_lines = ["findall matches"] - expected_output = ["find_all matches"] - assert convert_colang_2alpha_syntax(input_lines) == expected_output - - def test_triple_quotes_replacement_1(self): - input_lines = ['$ """some text"""'] - expected_output = ['$ ..."some text"'] - assert convert_colang_2alpha_syntax(input_lines) == expected_output - - def test_triple_quotes_replacement_2(self): - input_lines = ["$ '''some text'''"] - expected_output = ["$ ...'some text'"] - assert convert_colang_2alpha_syntax(input_lines) == expected_output - - def test_specific_phrases_replacement(self): - input_lines = [ - "catch colang errors", - "catch undefined flows", - "catch unexpected user utterance", - "track bot talking state", - "track user talking state", - "track user utterance state", - "poll llm request response", - "trigger user intent for unhandled user utterance", - "generate then continue interaction", - "track unhandled user intent state", - "respond to unhandled user intent", - ] - expected_output = [ - "catch colang errors", - "notification of undefined flow start", - "notification of unexpected user utterance", - "tracking bot talking state", - "tracking user talking state", - "tracking user talking state", - "polling llm request response", - "generating user intent for unhandled user utterance", - "llm continue interaction", - "tracking unhandled user intent state", - "continuation on unhandled user intent", - ] - assert convert_colang_2alpha_syntax(input_lines) == expected_output - - def test_meta_decorator_replacement(self): - input_lines = [ - "flow some_flow:", - "# meta: loop_id=123", - "# meta: example meta", - "some action", - ] - expected_output = [ - '@loop("123")', - "@meta(example_meta=True)", - "flow some_flow:", - "some action", - ] - assert convert_colang_2alpha_syntax(input_lines) == expected_output - - def test_convert_flow_examples(self): - input_1 = """ - flow bot inform something like issue - # meta: bot intent - (bot inform "ABC" - or bot inform "DEFG" - or bot inform "HJKL") - and (bot gesture "abc def" or bot gesture "hij kl") - """ - input_lines = textwrap.dedent(input_1).strip().split("\n") - - output_1 = """ - @meta(bot_intent=True) - flow bot inform something like issue - (bot inform "ABC" - or bot inform "DEFG" - or bot inform "HJKL") - and (bot gesture "abc def" or bot gesture "hij kl") - """ - output_lines = textwrap.dedent(output_1).strip().split("\n") - - assert convert_colang_2alpha_syntax(input_lines) == output_lines - - -class TestColang1SyntaxConversion: - def test_define_flow_conversion(self): - input_lines = ["define flow express greeting"] - expected_output = ["flow express greeting"] - assert convert_colang_1_syntax(input_lines) == expected_output - - def test_define_subflow_conversion(self): - input_lines = ["define subflow my_subflow"] - expected_output = ["flow my_subflow"] - assert convert_colang_1_syntax(input_lines) == expected_output - - def test_execute_to_await_and_pascal_case_action(self): - input_lines = ["execute some_action"] - expected_output = ["await SomeAction"] - assert convert_colang_1_syntax(input_lines) == expected_output - - def test_stop_to_abort(self): - input_lines = ["stop"] - expected_output = ["abort"] - assert convert_colang_1_syntax(input_lines) == expected_output - - def test_anonymous_flow_revised(self): - input_lines = ["flow", "user said hello"] - # because the flow is anonymous and only 'flow' is given, it will be converted to 'flow said hello' based on the first message - expected_output = ["flow said hello", "user said hello"] - output = convert_colang_1_syntax(input_lines) - # strip newline characters from the strings in the output list - output = [line.rstrip("\n") for line in output] - assert output == expected_output - - def test_global_variable_assignment(self): - input_lines = ["$variable = value"] - expected_output = ["global $variable\n$variable = value"] - assert convert_colang_1_syntax(input_lines) == expected_output - - def test_variable_assignment_in_await(self): - input_lines = ["$result = await some_action"] - expected_output = ["$result = await SomeAction"] - assert convert_colang_1_syntax(input_lines) == expected_output - - def test_bot_say_conversion(self): - input_lines = ["define bot", '"Hello!"', '"How can I help you?"'] - expected_output = [ - "flow bot", - 'bot say "Hello!"', - 'or bot say "How can I help you?"', - ] - assert convert_colang_1_syntax(input_lines) == expected_output - - def test_user_said_conversion(self): - input_lines = ["define user", '"I need assistance."', '"Can you help me?"'] - expected_output = [ - "flow user", - 'user said "I need assistance."', - 'or user said "Can you help me?"', - ] - assert convert_colang_1_syntax(input_lines) == expected_output - - def test_create_event_to_send(self): - input_lines = [" create event user_asked_question"] - expected_output = [" send user_asked_question"] - assert convert_colang_1_syntax(input_lines) == expected_output - - def test_config_variable_replacement(self): - # TODO(Rdinu): Need to see if this conversion is correct - input_lines = ["$config.setting = true"] - expected_output = [ - "global $system.config.setting\n$system.config.setting = true" - ] - assert convert_colang_1_syntax(input_lines) == expected_output - - def test_flow_with_special_characters(self): - input_lines = ["define flow my-flow's_test"] - expected_output = ["flow my flow s_test"] - assert convert_colang_1_syntax(input_lines) == expected_output - - def test_ellipsis_variable_assignment(self): - input_lines = ["# User's name", "$name = ...", "await greet_user"] - expected_output = [ - "# User's name", - "global $name\n$name = ...", - "await GreetUserAction", - ] - - expected_output = [ - "# User's name", - 'global $name\n$name = ... "User\'s name"', - "await GreetUserAction", - ] - assert convert_colang_1_syntax(input_lines) == expected_output - - @pytest.mark.skip("not implemented conversion") - def test_complex_conversion(self): - # TODO: add bot $response to bot say $response conversion - input_script = """ - define flow greeting_flow - when user express greeting - $response = execute generate_greeting - bot $response - """ - expected_output_script = """ - flow greeting_flow - when user express greeting - $response = await GenerateGreetingAction - bot say $response - """ - input_lines = textwrap.dedent(input_script).strip().split("\n") - expected_output = textwrap.dedent(expected_output_script).strip().split("\n") - - print(convert_colang_1_syntax(input_lines)) - assert convert_colang_1_syntax(input_lines) == expected_output - - def test_flow_with_execute_and_stop(self): - input_lines = [ - "define flow sample_flow", - ' when user "Cancel"', - " execute cancel_operation", - " stop", - ] - expected_output = [ - "flow sample_flow", - ' when user "Cancel"', - " await CancelOperationAction", - " abort", - ] - assert convert_colang_1_syntax(input_lines) == expected_output - - def test_await_camelcase_conversion(self): - input_lines = ["await sample_action"] - expected_output = ["await SampleAction"] - assert convert_colang_1_syntax(input_lines) == expected_output - - def test_nested_flow_conversion(self): - input_script = """ - define flow outer_flow - when condition_met - define subflow inner_flow - execute inner_action - """ - expected_output_script = """ - flow outer_flow - when condition_met - flow inner_flow - await InnerAction - """ - input_lines = textwrap.dedent(input_script).strip().split("\n") - expected_output = textwrap.dedent(expected_output_script).strip().split("\n") - assert convert_colang_1_syntax(input_lines) == expected_output From cd357431c6bd9dd95b875cb9afda66ec68dacd23 Mon Sep 17 00:00:00 2001 From: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com> Date: Tue, 2 Sep 2025 12:30:30 +0200 Subject: [PATCH 2/3] fix failing test on windows --- tests/cli/test_cli_main.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/cli/test_cli_main.py b/tests/cli/test_cli_main.py index bf379e3c2..b0812d359 100644 --- a/tests/cli/test_cli_main.py +++ b/tests/cli/test_cli_main.py @@ -13,6 +13,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import os from unittest.mock import MagicMock, patch import pytest @@ -195,7 +196,8 @@ def test_server_with_local_config( mock_exists.return_value = True result = runner.invoke(app, ["server"]) assert result.exit_code == 0 - assert mock_app.rails_config_path == "/current/dir/config" + expected_path = os.path.join("/current/dir", "config") + assert mock_app.rails_config_path == expected_path @patch("uvicorn.run") @patch("nemoguardrails.server.api.app") From 0aee5a08f53c855f3123a5f8bfa45bdfce2c2489 Mon Sep 17 00:00:00 2001 From: Pouyanpi <13303554+Pouyanpi@users.noreply.github.com> Date: Wed, 17 Sep 2025 14:45:00 +0200 Subject: [PATCH 3/3] remove comments --- tests/cli/test_debugger.py | 9 --------- 1 file changed, 9 deletions(-) diff --git a/tests/cli/test_debugger.py b/tests/cli/test_debugger.py index b0b0bcc97..f6c90ca8a 100644 --- a/tests/cli/test_debugger.py +++ b/tests/cli/test_debugger.py @@ -222,12 +222,3 @@ def test_tree_command(self, mock_is_active, mock_console, mock_tree): assert result.exit_code == 0 mock_tree.assert_called_with("main") - - -# TODO: addd tests -# class TestVarsCommand: -# def setup_method(self): -# self.state = MagicMock(spec=State) -# self.runtime = MagicMock() -# debugger.state = self.state -# debugger.runtime = self.runtime