From 4d6a8480214ef4e9ff1bca5c6f5434cade14cac2 Mon Sep 17 00:00:00 2001 From: Test Improver Date: Sat, 21 Mar 2026 01:13:02 +0000 Subject: [PATCH] test: replace gc.collect+sleep teardown with shutil.rmtree(ignore_errors=True) Closes #185 Replace fragile tearDown patterns (gc.collect + time.sleep) used to work around Windows file-lock timing issues with TemporaryDirectory. Switch to tempfile.mkdtemp() + shutil.rmtree(ignore_errors=True) which is cleaner, faster, and avoids unnecessary test latency. Affected files: - tests/unit/test_deps.py - tests/unit/test_env_variables.py - tests/unit/test_helpers.py (removed empty tearDown) - tests/unit/test_vscode_adapter.py (4 test classes) - tests/unit/workflow/test_workflow.py (removed safe_rmdir helper) All 2261 tests pass. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- tests/unit/test_deps.py | 171 ++++++----- tests/unit/test_env_variables.py | 72 +++-- tests/unit/test_helpers.py | 66 ++-- tests/unit/test_vscode_adapter.py | 442 +++++++++++++++++---------- tests/unit/workflow/test_workflow.py | 140 +++------ 5 files changed, 483 insertions(+), 408 deletions(-) diff --git a/tests/unit/test_deps.py b/tests/unit/test_deps.py index 6e584c0d..a4faa06e 100644 --- a/tests/unit/test_deps.py +++ b/tests/unit/test_deps.py @@ -1,164 +1,177 @@ """Unit tests for the dependencies management module.""" import os +import shutil import tempfile import unittest -from unittest.mock import patch, mock_open -import yaml +from unittest.mock import mock_open, patch + import frontmatter +import yaml -from apm_cli.deps.aggregator import scan_workflows_for_dependencies, sync_workflow_dependencies -from apm_cli.deps.verifier import verify_dependencies, install_missing_dependencies, load_apm_config +from apm_cli.deps.aggregator import ( + scan_workflows_for_dependencies, + sync_workflow_dependencies, +) +from apm_cli.deps.verifier import ( + install_missing_dependencies, + load_apm_config, + verify_dependencies, +) class TestDependenciesAggregator(unittest.TestCase): """Test cases for the dependencies aggregator.""" - - @patch('glob.glob') - @patch('builtins.open', new_callable=mock_open) - @patch('frontmatter.load') - def test_scan_workflows_for_dependencies(self, mock_frontmatter_load, mock_file, mock_glob): + + @patch("glob.glob") + @patch("builtins.open", new_callable=mock_open) + @patch("frontmatter.load") + def test_scan_workflows_for_dependencies( + self, mock_frontmatter_load, mock_file, mock_glob + ): """Test scanning workflows for dependencies.""" # Mock glob to return workflow files # First call returns GitHub prompts, second call returns generic prompts mock_glob.side_effect = [ - ['.github/prompts/workflow1.prompt.md'], - ['.github/prompts/workflow2.prompt.md'] + [".github/prompts/workflow1.prompt.md"], + [".github/prompts/workflow2.prompt.md"], ] - + # Mock frontmatter.load to return content with mcp metadata mock_content1 = unittest.mock.MagicMock() - mock_content1.metadata = {'mcp': ['server1', 'server2']} - + mock_content1.metadata = {"mcp": ["server1", "server2"]} + mock_content2 = unittest.mock.MagicMock() - mock_content2.metadata = {'mcp': ['server2', 'server3']} - + mock_content2.metadata = {"mcp": ["server2", "server3"]} + mock_frontmatter_load.side_effect = [mock_content1, mock_content2] - + # Call the function result = scan_workflows_for_dependencies() - + # Verify the results self.assertIsInstance(result, set) - self.assertEqual(result, {'server1', 'server2', 'server3'}) - self.assertEqual(mock_glob.call_count, 2) # We now make two glob calls for different patterns + self.assertEqual(result, {"server1", "server2", "server3"}) + self.assertEqual( + mock_glob.call_count, 2 + ) # We now make two glob calls for different patterns self.assertEqual(mock_file.call_count, 2) self.assertEqual(mock_frontmatter_load.call_count, 2) - - @patch('apm_cli.deps.aggregator.scan_workflows_for_dependencies') - @patch('builtins.open', new_callable=mock_open) - @patch('yaml.dump') + + @patch("apm_cli.deps.aggregator.scan_workflows_for_dependencies") + @patch("builtins.open", new_callable=mock_open) + @patch("yaml.dump") def test_sync_workflow_dependencies(self, mock_yaml_dump, mock_file, mock_scan): """Test syncing workflow dependencies to apm.yml.""" # Mock scan_workflows_for_dependencies to return a set of servers - mock_scan.return_value = {'server1', 'server2', 'server3'} - + mock_scan.return_value = {"server1", "server2", "server3"} + # Call the function - success, servers = sync_workflow_dependencies('test.yml') - + success, servers = sync_workflow_dependencies("test.yml") + # Verify the results self.assertTrue(success) - self.assertEqual(set(servers), {'server1', 'server2', 'server3'}) + self.assertEqual(set(servers), {"server1", "server2", "server3"}) self.assertEqual(mock_scan.call_count, 1) - mock_file.assert_called_once_with('test.yml', 'w', encoding='utf-8') + mock_file.assert_called_once_with("test.yml", "w", encoding="utf-8") mock_yaml_dump.assert_called_once() class TestDependenciesVerifier(unittest.TestCase): """Test cases for the dependencies verifier.""" - + def setUp(self): """Set up test fixtures.""" - self.temp_dir = tempfile.TemporaryDirectory() - self.config_path = os.path.join(self.temp_dir.name, 'apm.yml') - + self.temp_dir = tempfile.mkdtemp() + self.config_path = os.path.join(self.temp_dir, "apm.yml") + # Create a test configuration file - config = { - 'version': '1.0', - 'servers': ['server1', 'server2', 'server3'] - } - - with open(self.config_path, 'w', encoding='utf-8') as f: + config = {"version": "1.0", "servers": ["server1", "server2", "server3"]} + + with open(self.config_path, "w", encoding="utf-8") as f: yaml.dump(config, f) - + def tearDown(self): """Tear down test fixtures.""" - # Force garbage collection to release file handles - import gc - gc.collect() - # Small delay to allow Windows to release locks - import time - time.sleep(0.1) - - self.temp_dir.cleanup() - + shutil.rmtree(self.temp_dir, ignore_errors=True) + def test_load_apm_config(self): """Test loading the APM configuration file.""" # Test with an existing file config = load_apm_config(self.config_path) self.assertIsInstance(config, dict) - self.assertEqual(config['version'], '1.0') - self.assertEqual(config['servers'], ['server1', 'server2', 'server3']) - + self.assertEqual(config["version"], "1.0") + self.assertEqual(config["servers"], ["server1", "server2", "server3"]) + # Test with a non-existent file - config = load_apm_config('nonexistent.yml') + config = load_apm_config("nonexistent.yml") self.assertIsNone(config) - - @patch('apm_cli.factory.PackageManagerFactory.create_package_manager') + + @patch("apm_cli.factory.PackageManagerFactory.create_package_manager") def test_verify_dependencies(self, mock_factory): """Test verifying dependencies.""" # Mock the package manager to return a list of installed packages mock_package_manager = unittest.mock.MagicMock() - mock_package_manager.list_installed.return_value = ['server1', 'server3'] + mock_package_manager.list_installed.return_value = ["server1", "server3"] mock_factory.return_value = mock_package_manager - + # Call the function all_installed, installed, missing = verify_dependencies(self.config_path) - + # Verify the results self.assertFalse(all_installed) - self.assertEqual(set(installed), {'server1', 'server3'}) - self.assertEqual(set(missing), {'server2'}) - + self.assertEqual(set(installed), {"server1", "server3"}) + self.assertEqual(set(missing), {"server2"}) + # Test with all packages installed - mock_package_manager.list_installed.return_value = ['server1', 'server2', 'server3'] + mock_package_manager.list_installed.return_value = [ + "server1", + "server2", + "server3", + ] all_installed, installed, missing = verify_dependencies(self.config_path) self.assertTrue(all_installed) - self.assertEqual(set(installed), {'server1', 'server2', 'server3'}) + self.assertEqual(set(installed), {"server1", "server2", "server3"}) self.assertEqual(missing, []) - - @patch('apm_cli.factory.ClientFactory.create_client') - @patch('apm_cli.factory.PackageManagerFactory.create_package_manager') - @patch('apm_cli.deps.verifier.verify_dependencies') - def test_install_missing_dependencies(self, mock_verify, mock_factory, mock_client_factory): + + @patch("apm_cli.factory.ClientFactory.create_client") + @patch("apm_cli.factory.PackageManagerFactory.create_package_manager") + @patch("apm_cli.deps.verifier.verify_dependencies") + def test_install_missing_dependencies( + self, mock_verify, mock_factory, mock_client_factory + ): """Test installing missing dependencies.""" # Mock verify_dependencies to return missing packages - mock_verify.return_value = (False, ['server1'], ['server2', 'server3']) - + mock_verify.return_value = (False, ["server1"], ["server2", "server3"]) + # Mock the package manager to install packages mock_package_manager = unittest.mock.MagicMock() mock_package_manager.install.return_value = True mock_factory.return_value = mock_package_manager - + # Mock the client adapter mock_client = unittest.mock.MagicMock() mock_client.configure_mcp_server.return_value = True mock_client_factory.return_value = mock_client - + # Call the function success, installed = install_missing_dependencies(self.config_path, "vscode") - + # Verify the results self.assertTrue(success) - self.assertEqual(set(installed), {'server2', 'server3'}) + self.assertEqual(set(installed), {"server2", "server3"}) self.assertEqual(mock_verify.call_count, 1) self.assertEqual(mock_package_manager.install.call_count, 2) self.assertEqual(mock_client.configure_mcp_server.call_count, 2) - + # Verify client was configured properly - mock_client.configure_mcp_server.assert_any_call('server2', server_name='server2') - mock_client.configure_mcp_server.assert_any_call('server3', server_name='server3') + mock_client.configure_mcp_server.assert_any_call( + "server2", server_name="server2" + ) + mock_client.configure_mcp_server.assert_any_call( + "server3", server_name="server3" + ) if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main() diff --git a/tests/unit/test_env_variables.py b/tests/unit/test_env_variables.py index 6b4a85fb..fce2b34b 100644 --- a/tests/unit/test_env_variables.py +++ b/tests/unit/test_env_variables.py @@ -1,51 +1,50 @@ """Tests for environment variables handling in VSCode adapter.""" -import os import json +import os +import shutil import tempfile import unittest -from unittest.mock import patch, MagicMock +from unittest.mock import MagicMock, patch + from apm_cli.adapters.client.vscode import VSCodeClientAdapter class TestEnvironmentVariablesHandling(unittest.TestCase): """Test cases for environment variables handling in VSCode adapter.""" - + def setUp(self): """Set up test fixtures.""" - self.temp_dir = tempfile.TemporaryDirectory() - self.vscode_dir = os.path.join(self.temp_dir.name, ".vscode") + self.temp_dir = tempfile.mkdtemp() + self.vscode_dir = os.path.join(self.temp_dir, ".vscode") os.makedirs(self.vscode_dir, exist_ok=True) self.temp_path = os.path.join(self.vscode_dir, "mcp.json") - + # Create a temporary MCP configuration file with open(self.temp_path, "w") as f: json.dump({"servers": {}}, f) - + # Create mock clients - self.mock_registry_patcher = patch('apm_cli.adapters.client.vscode.SimpleRegistryClient') + self.mock_registry_patcher = patch( + "apm_cli.adapters.client.vscode.SimpleRegistryClient" + ) self.mock_registry_class = self.mock_registry_patcher.start() self.mock_registry = MagicMock() self.mock_registry_class.return_value = self.mock_registry - - self.mock_integration_patcher = patch('apm_cli.adapters.client.vscode.RegistryIntegration') + + self.mock_integration_patcher = patch( + "apm_cli.adapters.client.vscode.RegistryIntegration" + ) self.mock_integration_class = self.mock_integration_patcher.start() self.mock_integration = MagicMock() self.mock_integration_class.return_value = self.mock_integration - + def tearDown(self): """Tear down test fixtures.""" - # Force garbage collection to release file handles - import gc - gc.collect() - # Small delay to allow Windows to release locks - import time - time.sleep(0.1) - self.mock_registry_patcher.stop() self.mock_integration_patcher.stop() - self.temp_dir.cleanup() - + shutil.rmtree(self.temp_dir, ignore_errors=True) + @patch("apm_cli.adapters.client.vscode.VSCodeClientAdapter.get_config_path") def test_configure_mcp_server_with_environment_variables(self, mock_get_path): """Test configuring an MCP server with environment variables.""" @@ -61,49 +60,48 @@ def test_configure_mcp_server_with_environment_variables(self, mock_get_path): "version": "1.0.0", "runtime_hint": "npx", "environment_variables": [ - { - "description": "YOUR_API_KEY", - "name": "AGENTQL_API_KEY" - } - ] + {"description": "YOUR_API_KEY", "name": "AGENTQL_API_KEY"} + ], } - ] + ], } - + # Set up the mock mock_get_path.return_value = self.temp_path self.mock_registry.find_server_by_reference.return_value = server_info - + # Create the adapter and configure the server adapter = VSCodeClientAdapter() result = adapter.configure_mcp_server( server_url="io.github.tinyfish-io/agentql-mcp", - server_name="io.github.tinyfish-io/agentql-mcp" + server_name="io.github.tinyfish-io/agentql-mcp", ) - + # Check the result self.assertTrue(result) - + # Read the config file and verify the content with open(self.temp_path, "r") as f: updated_config = json.load(f) - + # Check the server configuration server_config = updated_config["servers"]["io.github.tinyfish-io/agentql-mcp"] self.assertEqual(server_config["type"], "stdio") self.assertEqual(server_config["command"], "npx") self.assertEqual(server_config["args"], ["-y", "agentql-mcp"]) - + # Verify environment variables were added self.assertIn("env", server_config) self.assertIn("AGENTQL_API_KEY", server_config["env"]) - self.assertEqual(server_config["env"]["AGENTQL_API_KEY"], "${input:agentql-api-key}") - + self.assertEqual( + server_config["env"]["AGENTQL_API_KEY"], "${input:agentql-api-key}" + ) + # Verify input variables were added self.assertIn("inputs", updated_config) self.assertIsInstance(updated_config["inputs"], list) self.assertTrue(len(updated_config["inputs"]) > 0) - + # Check if the input variable for the API key is present input_var_found = False for input_var in updated_config["inputs"]: @@ -113,7 +111,7 @@ def test_configure_mcp_server_with_environment_variables(self, mock_get_path): self.assertTrue(input_var["password"]) self.assertIn("description", input_var) break - + self.assertTrue(input_var_found, "Input variable definition not found") diff --git a/tests/unit/test_helpers.py b/tests/unit/test_helpers.py index afe937f5..e581ec0d 100644 --- a/tests/unit/test_helpers.py +++ b/tests/unit/test_helpers.py @@ -1,42 +1,39 @@ """Tests for helper utility functions.""" import json -import unittest import sys +import unittest from pathlib import Path -from apm_cli.utils.helpers import is_tool_available, detect_platform, get_available_package_managers, find_plugin_json + +from apm_cli.utils.helpers import ( + detect_platform, + find_plugin_json, + get_available_package_managers, + is_tool_available, +) class TestHelpers(unittest.TestCase): """Test cases for helper utility functions.""" - - def tearDown(self): - """Tear down test fixtures.""" - # Force garbage collection to release file handles - import gc - gc.collect() - # Small delay to allow Windows to release locks - import time - time.sleep(0.1) - + def test_is_tool_available(self): """Test is_tool_available function with known commands.""" # Python should always be available in the test environment - self.assertTrue(is_tool_available('python')) - + self.assertTrue(is_tool_available("python")) + # Test a command that almost certainly doesn't exist - self.assertFalse(is_tool_available('this_command_does_not_exist_12345')) - + self.assertFalse(is_tool_available("this_command_does_not_exist_12345")) + def test_detect_platform(self): """Test detect_platform function.""" platform = detect_platform() - self.assertIn(platform, ['macos', 'linux', 'windows', 'unknown']) - + self.assertIn(platform, ["macos", "linux", "windows", "unknown"]) + def test_get_available_package_managers(self): """Test get_available_package_managers function.""" managers = get_available_package_managers() self.assertIsInstance(managers, dict) - + # The function should return a valid dict # If any managers are found, they should have valid string values for name, path in managers.items(): @@ -44,15 +41,19 @@ def test_get_available_package_managers(self): self.assertIsInstance(path, str) self.assertTrue(len(name) > 0) self.assertTrue(len(path) > 0) - + # On most Unix systems, at least one package manager should be available # This is a reasonable expectation but not guaranteed on minimal systems import sys - if sys.platform != 'win32': + + if sys.platform != "win32": # Skip this assertion on Windows since it might not have any # On Unix systems, we expect at least one package manager - self.assertGreater(len(managers), 0, - "Expected at least one package manager on Unix systems") + self.assertGreater( + len(managers), + 0, + "Expected at least one package manager on Unix systems", + ) class TestFindPluginJson(unittest.TestCase): @@ -61,6 +62,7 @@ class TestFindPluginJson(unittest.TestCase): def test_finds_root_plugin_json(self, tmp_path=None): """Root plugin.json is returned when present.""" import tempfile + with tempfile.TemporaryDirectory() as d: root = Path(d) pj = root / "plugin.json" @@ -70,6 +72,7 @@ def test_finds_root_plugin_json(self, tmp_path=None): def test_finds_github_plugin_json(self): """plugin.json under .github/plugin/ is found.""" import tempfile + with tempfile.TemporaryDirectory() as d: root = Path(d) target = root / ".github" / "plugin" / "plugin.json" @@ -80,6 +83,7 @@ def test_finds_github_plugin_json(self): def test_finds_claude_plugin_json(self): """plugin.json under .claude-plugin/ is found.""" import tempfile + with tempfile.TemporaryDirectory() as d: root = Path(d) target = root / ".claude-plugin" / "plugin.json" @@ -90,6 +94,7 @@ def test_finds_claude_plugin_json(self): def test_finds_cursor_plugin_json(self): """plugin.json under .cursor-plugin/ is found.""" import tempfile + with tempfile.TemporaryDirectory() as d: root = Path(d) target = root / ".cursor-plugin" / "plugin.json" @@ -100,9 +105,15 @@ def test_finds_cursor_plugin_json(self): def test_priority_order(self): """Root wins over .github/plugin/ which wins over .claude-plugin/ which wins over .cursor-plugin/.""" import tempfile + with tempfile.TemporaryDirectory() as d: root = Path(d) - for sub in ["plugin.json", ".github/plugin/plugin.json", ".claude-plugin/plugin.json", ".cursor-plugin/plugin.json"]: + for sub in [ + "plugin.json", + ".github/plugin/plugin.json", + ".claude-plugin/plugin.json", + ".cursor-plugin/plugin.json", + ]: p = root / sub p.parent.mkdir(parents=True, exist_ok=True) p.write_text(json.dumps({"name": sub})) @@ -111,6 +122,7 @@ def test_priority_order(self): def test_cursor_plugin_found_when_only_option(self): """When only .cursor-plugin/ has plugin.json, it is found.""" import tempfile + with tempfile.TemporaryDirectory() as d: root = Path(d) target = root / ".cursor-plugin" / "plugin.json" @@ -122,6 +134,7 @@ def test_cursor_plugin_found_when_only_option(self): def test_ignores_unrelated_locations(self): """plugin.json buried in node_modules or other dirs is NOT found.""" import tempfile + with tempfile.TemporaryDirectory() as d: root = Path(d) hidden = root / "node_modules" / "evil" / "plugin.json" @@ -132,9 +145,10 @@ def test_ignores_unrelated_locations(self): def test_returns_none_when_absent(self): """None is returned when no plugin.json exists anywhere.""" import tempfile + with tempfile.TemporaryDirectory() as d: assert find_plugin_json(Path(d)) is None -if __name__ == '__main__': - unittest.main() \ No newline at end of file +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_vscode_adapter.py b/tests/unit/test_vscode_adapter.py index 3acbde02..43798580 100644 --- a/tests/unit/test_vscode_adapter.py +++ b/tests/unit/test_vscode_adapter.py @@ -1,41 +1,46 @@ """Unit tests for the VSCode client adapter.""" -import os import json +import os +import shutil import tempfile import unittest from pathlib import Path +from unittest.mock import MagicMock, patch + import pytest -from unittest.mock import patch, MagicMock -from apm_cli.adapters.client.vscode import VSCodeClientAdapter + from apm_cli.adapters.client.base import MCPClientAdapter +from apm_cli.adapters.client.vscode import VSCodeClientAdapter class TestVSCodeClientAdapter(unittest.TestCase): """Test cases for the VSCode client adapter.""" - + def setUp(self): """Set up test fixtures.""" - self.temp_dir = tempfile.TemporaryDirectory() - self.vscode_dir = os.path.join(self.temp_dir.name, ".vscode") + self.temp_dir = tempfile.mkdtemp() + self.vscode_dir = os.path.join(self.temp_dir, ".vscode") os.makedirs(self.vscode_dir, exist_ok=True) self.temp_path = os.path.join(self.vscode_dir, "mcp.json") - - # Create a temporary MCP configuration file with open(self.temp_path, "w") as f: json.dump({"servers": {}}, f) - + # Create mock clients - self.mock_registry_patcher = patch('apm_cli.adapters.client.vscode.SimpleRegistryClient') + self.mock_registry_patcher = patch( + "apm_cli.adapters.client.vscode.SimpleRegistryClient" + ) self.mock_registry_class = self.mock_registry_patcher.start() self.mock_registry = MagicMock() self.mock_registry_class.return_value = self.mock_registry - - self.mock_integration_patcher = patch('apm_cli.adapters.client.vscode.RegistryIntegration') + + self.mock_integration_patcher = patch( + "apm_cli.adapters.client.vscode.RegistryIntegration" + ) self.mock_integration_class = self.mock_integration_patcher.start() self.mock_integration = MagicMock() self.mock_integration_class.return_value = self.mock_integration - + # Mock server details self.server_info = { "id": "12345", @@ -46,113 +51,105 @@ def setUp(self): "name": "@mcp/fetch", "version": "1.0.0", "registry_name": "npm", - "runtime_hint": "npx" + "runtime_hint": "npx", } - ] + ], } - + # Configure the mocks self.mock_registry.get_server_info.return_value = self.server_info self.mock_registry.get_server_by_name.return_value = self.server_info self.mock_registry.find_server_by_reference.return_value = self.server_info - + def tearDown(self): """Tear down test fixtures.""" - # Force garbage collection to release file handles - import gc - gc.collect() - # Small delay to allow Windows to release locks - import time - time.sleep(0.1) - self.mock_registry_patcher.stop() self.mock_integration_patcher.stop() - self.temp_dir.cleanup() - + shutil.rmtree(self.temp_dir, ignore_errors=True) + @patch("apm_cli.adapters.client.vscode.VSCodeClientAdapter.get_config_path") def test_get_current_config(self, mock_get_path): """Test getting the current configuration.""" mock_get_path.return_value = self.temp_path adapter = VSCodeClientAdapter() - + config = adapter.get_current_config() self.assertEqual(config, {"servers": {}}) - + @patch("apm_cli.adapters.client.vscode.VSCodeClientAdapter.get_config_path") def test_update_config(self, mock_get_path): """Test updating the configuration.""" mock_get_path.return_value = self.temp_path adapter = VSCodeClientAdapter() - + new_config = { "servers": { "test-server": { "type": "stdio", "command": "uvx", - "args": ["mcp-server-test"] + "args": ["mcp-server-test"], } } } - + result = adapter.update_config(new_config) - + with open(self.temp_path, "r") as f: updated_config = json.load(f) - + self.assertEqual(updated_config, new_config) self.assertTrue(result) - + @patch("apm_cli.adapters.client.vscode.VSCodeClientAdapter.get_config_path") def test_update_config_nonexistent_file(self, mock_get_path): """Test updating configuration when file doesn't exist.""" nonexistent_path = os.path.join(self.vscode_dir, "nonexistent.json") mock_get_path.return_value = nonexistent_path adapter = VSCodeClientAdapter() - + new_config = { "servers": { "test-server": { "type": "stdio", "command": "uvx", - "args": ["mcp-server-test"] + "args": ["mcp-server-test"], } } } - + result = adapter.update_config(new_config) - + with open(nonexistent_path, "r") as f: updated_config = json.load(f) - + self.assertEqual(updated_config, new_config) self.assertTrue(result) - + @patch("apm_cli.adapters.client.vscode.VSCodeClientAdapter.get_config_path") def test_configure_mcp_server(self, mock_get_path): """Test configuring an MCP server.""" mock_get_path.return_value = self.temp_path adapter = VSCodeClientAdapter() - - result = adapter.configure_mcp_server( - server_url="fetch", - server_name="fetch" - ) - + + result = adapter.configure_mcp_server(server_url="fetch", server_name="fetch") + with open(self.temp_path, "r") as f: updated_config = json.load(f) - + self.assertTrue(result) self.assertIn("servers", updated_config) self.assertIn("fetch", updated_config["servers"]) - + # Verify the registry client was called self.mock_registry.find_server_by_reference.assert_called_once_with("fetch") - + # Verify the server configuration self.assertEqual(updated_config["servers"]["fetch"]["type"], "stdio") self.assertEqual(updated_config["servers"]["fetch"]["command"], "npx") - self.assertEqual(updated_config["servers"]["fetch"]["args"], ["-y", "@mcp/fetch"]) - + self.assertEqual( + updated_config["servers"]["fetch"]["args"], ["-y", "@mcp/fetch"] + ) + @patch("apm_cli.adapters.client.vscode.VSCodeClientAdapter.get_config_path") def test_configure_mcp_server_update_existing(self, mock_get_path): """Test updating an existing MCP server.""" @@ -162,79 +159,79 @@ def test_configure_mcp_server_update_existing(self, mock_get_path): "fetch": { "type": "stdio", "command": "docker", - "args": ["run", "-i", "--rm", "mcp/fetch"] + "args": ["run", "-i", "--rm", "mcp/fetch"], } } } - + with open(self.temp_path, "w") as f: json.dump(existing_config, f) - + mock_get_path.return_value = self.temp_path adapter = VSCodeClientAdapter() - - result = adapter.configure_mcp_server( - server_url="fetch", - server_name="fetch" - ) - + + result = adapter.configure_mcp_server(server_url="fetch", server_name="fetch") + with open(self.temp_path, "r") as f: updated_config = json.load(f) - + self.assertTrue(result) self.assertIn("fetch", updated_config["servers"]) - + # Verify the registry client was called self.mock_registry.find_server_by_reference.assert_called_once_with("fetch") - + # Verify the server configuration self.assertEqual(updated_config["servers"]["fetch"]["type"], "stdio") self.assertEqual(updated_config["servers"]["fetch"]["command"], "npx") - self.assertEqual(updated_config["servers"]["fetch"]["args"], ["-y", "@mcp/fetch"]) - + self.assertEqual( + updated_config["servers"]["fetch"]["args"], ["-y", "@mcp/fetch"] + ) + @patch("apm_cli.adapters.client.vscode.VSCodeClientAdapter.get_config_path") def test_configure_mcp_server_empty_url(self, mock_get_path): """Test configuring an MCP server with empty URL.""" mock_get_path.return_value = self.temp_path adapter = VSCodeClientAdapter() - + result = adapter.configure_mcp_server( - server_url="", - server_name="Example Server" + server_url="", server_name="Example Server" ) - + self.assertFalse(result) - + @patch("apm_cli.adapters.client.vscode.VSCodeClientAdapter.get_config_path") def test_configure_mcp_server_registry_error(self, mock_get_path): """Test error behavior when registry doesn't have server details.""" # Configure the mock to return None when server is not found self.mock_registry.find_server_by_reference.return_value = None - + mock_get_path.return_value = self.temp_path adapter = VSCodeClientAdapter() - + # Test that ValueError is raised when server details can't be retrieved with self.assertRaises(ValueError) as context: adapter.configure_mcp_server( - server_url="unknown-server", - server_name="unknown-server" + server_url="unknown-server", server_name="unknown-server" ) - - self.assertIn("Failed to retrieve server details for 'unknown-server'. Server not found in registry.", str(context.exception)) - + + self.assertIn( + "Failed to retrieve server details for 'unknown-server'. Server not found in registry.", + str(context.exception), + ) + @patch("os.getcwd") def test_get_config_path_repository(self, mock_getcwd): """Test getting the config path in the repository.""" - mock_getcwd.return_value = self.temp_dir.name - + mock_getcwd.return_value = self.temp_dir + adapter = VSCodeClientAdapter() path = adapter.get_config_path() - + # Create Path objects for comparison to handle platform differences actual_path = Path(path) - expected_path = Path(self.temp_dir.name) / ".vscode" / "mcp.json" - + expected_path = Path(self.temp_dir) / ".vscode" / "mcp.json" + # Compare parts of the path to avoid string formatting issues self.assertEqual(actual_path.parent, expected_path.parent) self.assertEqual(actual_path.name, expected_path.name) @@ -264,7 +261,12 @@ def test_format_server_config_streamable_http_remote(self, mock_get_path): server_info = { "name": "streamable-server", - "remotes": [{"transport_type": "streamable-http", "url": "https://stream.example.com"}], + "remotes": [ + { + "transport_type": "streamable-http", + "url": "https://stream.example.com", + } + ], } config, inputs = adapter._format_server_config(server_info) @@ -279,22 +281,27 @@ def test_format_server_config_remote_with_list_headers(self, mock_get_path): server_info = { "name": "header-server", - "remotes": [{ - "transport_type": "http", - "url": "https://example.com", - "headers": [ - {"name": "Authorization", "value": "Bearer token123"}, - {"name": "X-Custom", "value": "val"}, - ], - }], + "remotes": [ + { + "transport_type": "http", + "url": "https://example.com", + "headers": [ + {"name": "Authorization", "value": "Bearer token123"}, + {"name": "X-Custom", "value": "val"}, + ], + } + ], } config, inputs = adapter._format_server_config(server_info) self.assertEqual(config["type"], "http") - self.assertEqual(config["headers"], { - "Authorization": "Bearer token123", - "X-Custom": "val", - }) + self.assertEqual( + config["headers"], + { + "Authorization": "Bearer token123", + "X-Custom": "val", + }, + ) @patch("apm_cli.adapters.client.vscode.VSCodeClientAdapter.get_config_path") def test_configure_self_defined_http_via_cache(self, mock_get_path): @@ -306,7 +313,9 @@ def test_configure_self_defined_http_via_cache(self, mock_get_path): cache = { "my-private-srv": { "name": "my-private-srv", - "remotes": [{"transport_type": "http", "url": "http://localhost:8787/"}], + "remotes": [ + {"transport_type": "http", "url": "http://localhost:8787/"} + ], } } @@ -322,7 +331,9 @@ def test_configure_self_defined_http_via_cache(self, mock_get_path): self.assertIn("my-private-srv", config["servers"]) self.assertEqual(config["servers"]["my-private-srv"]["type"], "http") - self.assertEqual(config["servers"]["my-private-srv"]["url"], "http://localhost:8787/") + self.assertEqual( + config["servers"]["my-private-srv"]["url"], "http://localhost:8787/" + ) class TestVSCodeSelectBestPackage(unittest.TestCase): @@ -330,9 +341,13 @@ class TestVSCodeSelectBestPackage(unittest.TestCase): def setUp(self): """Set up test fixtures.""" - self.mock_registry_patcher = patch('apm_cli.adapters.client.vscode.SimpleRegistryClient') + self.mock_registry_patcher = patch( + "apm_cli.adapters.client.vscode.SimpleRegistryClient" + ) self.mock_registry_patcher.start() - self.mock_integration_patcher = patch('apm_cli.adapters.client.vscode.RegistryIntegration') + self.mock_integration_patcher = patch( + "apm_cli.adapters.client.vscode.RegistryIntegration" + ) self.mock_integration_patcher.start() self.adapter = VSCodeClientAdapter() @@ -363,7 +378,11 @@ def test_falls_back_to_runtime_hint(self): """Falls back to any package with runtime_hint when no priority match.""" packages = [ {"name": "Azure.Mcp", "registry_name": "nuget", "runtime_hint": "dotnet"}, - {"name": "azure-mcp-linux-x64", "registry_name": "mcpb", "runtime_hint": ""}, + { + "name": "azure-mcp-linux-x64", + "registry_name": "mcpb", + "runtime_hint": "", + }, ] result = self.adapter._select_best_package(packages) self.assertEqual(result["name"], "Azure.Mcp") @@ -384,27 +403,29 @@ class TestVSCodeStdioRegistryPackages(unittest.TestCase): """Test that VS Code adapter correctly handles stdio-only registry servers.""" def setUp(self): - self.temp_dir = tempfile.TemporaryDirectory() - self.vscode_dir = os.path.join(self.temp_dir.name, ".vscode") + self.temp_dir = tempfile.mkdtemp() + self.vscode_dir = os.path.join(self.temp_dir, ".vscode") os.makedirs(self.vscode_dir, exist_ok=True) self.temp_path = os.path.join(self.vscode_dir, "mcp.json") with open(self.temp_path, "w") as f: json.dump({"servers": {}}, f) - self.mock_registry_patcher = patch('apm_cli.adapters.client.vscode.SimpleRegistryClient') + self.mock_registry_patcher = patch( + "apm_cli.adapters.client.vscode.SimpleRegistryClient" + ) self.mock_registry_class = self.mock_registry_patcher.start() self.mock_registry = MagicMock() self.mock_registry_class.return_value = self.mock_registry - self.mock_integration_patcher = patch('apm_cli.adapters.client.vscode.RegistryIntegration') + self.mock_integration_patcher = patch( + "apm_cli.adapters.client.vscode.RegistryIntegration" + ) self.mock_integration_patcher.start() def tearDown(self): - import gc; gc.collect() - import time; time.sleep(0.1) self.mock_registry_patcher.stop() self.mock_integration_patcher.stop() - self.temp_dir.cleanup() + shutil.rmtree(self.temp_dir, ignore_errors=True) @patch("apm_cli.adapters.client.vscode.VSCodeClientAdapter.get_config_path") def test_stdio_npm_selected_over_nuget(self, mock_get_path): @@ -417,21 +438,44 @@ def test_stdio_npm_selected_over_nuget(self, mock_get_path): "name": "azure", "description": "Azure MCP server", "packages": [ - {"name": "Azure.Mcp", "version": "2.0.0-beta.24", "registry_name": "nuget", - "runtime_hint": "dotnet", "runtime_arguments": [ - {"is_required": True, "value_hint": "server"}, - {"is_required": True, "value_hint": "start"}]}, - {"name": "@azure/mcp", "version": "2.0.0-beta.24", "registry_name": "npm", - "runtime_hint": "npx", "runtime_arguments": [ - {"is_required": True, "value_hint": "server"}, - {"is_required": True, "value_hint": "start"}]}, - {"name": "msmcp-azure", "version": "2.0.0-beta.24", "registry_name": "pypi", - "runtime_hint": "uvx", "runtime_arguments": [ - {"is_required": True, "value_hint": "server"}, - {"is_required": True, "value_hint": "start"}]}, - {"name": "azure-mcp-linux-x64", "version": "2.0.0-beta.24", "registry_name": "mcpb", - "runtime_hint": "", "runtime_arguments": []}, - ] + { + "name": "Azure.Mcp", + "version": "2.0.0-beta.24", + "registry_name": "nuget", + "runtime_hint": "dotnet", + "runtime_arguments": [ + {"is_required": True, "value_hint": "server"}, + {"is_required": True, "value_hint": "start"}, + ], + }, + { + "name": "@azure/mcp", + "version": "2.0.0-beta.24", + "registry_name": "npm", + "runtime_hint": "npx", + "runtime_arguments": [ + {"is_required": True, "value_hint": "server"}, + {"is_required": True, "value_hint": "start"}, + ], + }, + { + "name": "msmcp-azure", + "version": "2.0.0-beta.24", + "registry_name": "pypi", + "runtime_hint": "uvx", + "runtime_arguments": [ + {"is_required": True, "value_hint": "server"}, + {"is_required": True, "value_hint": "start"}, + ], + }, + { + "name": "azure-mcp-linux-x64", + "version": "2.0.0-beta.24", + "registry_name": "mcpb", + "runtime_hint": "", + "runtime_arguments": [], + }, + ], # No remotes key — server only provides stdio packages } @@ -461,12 +505,17 @@ def test_generic_runtime_hint_fallback(self, mock_get_path): "id": "nuget-only-id", "name": "nuget-server", "packages": [ - {"name": "MyServer.Mcp", "registry_name": "nuget", - "runtime_hint": "dotnet", "runtime_arguments": [ - {"is_required": True, "value_hint": "run"}, - {"is_required": True, "value_hint": "--project"}, - {"is_required": True, "value_hint": "MyServer.Mcp"}]} - ] + { + "name": "MyServer.Mcp", + "registry_name": "nuget", + "runtime_hint": "dotnet", + "runtime_arguments": [ + {"is_required": True, "value_hint": "run"}, + {"is_required": True, "value_hint": "--project"}, + {"is_required": True, "value_hint": "MyServer.Mcp"}, + ], + } + ], } self.mock_registry.find_server_by_reference.return_value = server_info adapter = VSCodeClientAdapter() @@ -497,7 +546,7 @@ def test_error_message_when_packages_exist_but_none_supported(self, mock_get_pat "packages": [ {"name": "binary-linux-x64", "registry_name": "mcpb"}, {"name": "binary-linux-arm64", "registry_name": "mcpb"}, - ] + ], } with self.assertRaises(ValueError) as ctx: @@ -523,9 +572,13 @@ class TestVSCodeInferRegistryName(unittest.TestCase): """Test _infer_registry_name with various package metadata patterns.""" def setUp(self): - self.mock_registry_patcher = patch('apm_cli.adapters.client.vscode.SimpleRegistryClient') + self.mock_registry_patcher = patch( + "apm_cli.adapters.client.vscode.SimpleRegistryClient" + ) self.mock_registry_patcher.start() - self.mock_integration_patcher = patch('apm_cli.adapters.client.vscode.RegistryIntegration') + self.mock_integration_patcher = patch( + "apm_cli.adapters.client.vscode.RegistryIntegration" + ) self.mock_integration_patcher.start() def tearDown(self): @@ -533,28 +586,68 @@ def tearDown(self): self.mock_integration_patcher.stop() def test_explicit_registry_name(self): - self.assertEqual(VSCodeClientAdapter._infer_registry_name({"name": "pkg", "registry_name": "npm"}), "npm") + self.assertEqual( + VSCodeClientAdapter._infer_registry_name( + {"name": "pkg", "registry_name": "npm"} + ), + "npm", + ) def test_empty_registry_name_scoped_npm(self): - self.assertEqual(VSCodeClientAdapter._infer_registry_name({"name": "@azure/mcp", "registry_name": ""}), "npm") + self.assertEqual( + VSCodeClientAdapter._infer_registry_name( + {"name": "@azure/mcp", "registry_name": ""} + ), + "npm", + ) def test_empty_registry_name_runtime_hint_npx(self): - self.assertEqual(VSCodeClientAdapter._infer_registry_name({"name": "some-pkg", "registry_name": "", "runtime_hint": "npx"}), "npm") + self.assertEqual( + VSCodeClientAdapter._infer_registry_name( + {"name": "some-pkg", "registry_name": "", "runtime_hint": "npx"} + ), + "npm", + ) def test_empty_registry_name_runtime_hint_uvx(self): - self.assertEqual(VSCodeClientAdapter._infer_registry_name({"name": "some-pkg", "registry_name": "", "runtime_hint": "uvx"}), "pypi") + self.assertEqual( + VSCodeClientAdapter._infer_registry_name( + {"name": "some-pkg", "registry_name": "", "runtime_hint": "uvx"} + ), + "pypi", + ) def test_empty_registry_name_docker_image(self): - self.assertEqual(VSCodeClientAdapter._infer_registry_name({"name": "ghcr.io/org/img", "registry_name": ""}), "docker") + self.assertEqual( + VSCodeClientAdapter._infer_registry_name( + {"name": "ghcr.io/org/img", "registry_name": ""} + ), + "docker", + ) def test_empty_registry_name_nuget_pascal_case(self): - self.assertEqual(VSCodeClientAdapter._infer_registry_name({"name": "Azure.Mcp", "registry_name": ""}), "nuget") + self.assertEqual( + VSCodeClientAdapter._infer_registry_name( + {"name": "Azure.Mcp", "registry_name": ""} + ), + "nuget", + ) def test_empty_registry_name_mcpb_url(self): - self.assertEqual(VSCodeClientAdapter._infer_registry_name({"name": "https://example.com/bin.mcpb", "registry_name": ""}), "mcpb") + self.assertEqual( + VSCodeClientAdapter._infer_registry_name( + {"name": "https://example.com/bin.mcpb", "registry_name": ""} + ), + "mcpb", + ) def test_unknown_returns_empty(self): - self.assertEqual(VSCodeClientAdapter._infer_registry_name({"name": "unknown-pkg", "registry_name": ""}), "") + self.assertEqual( + VSCodeClientAdapter._infer_registry_name( + {"name": "unknown-pkg", "registry_name": ""} + ), + "", + ) def test_none_package(self): self.assertEqual(VSCodeClientAdapter._infer_registry_name(None), "") @@ -564,9 +657,13 @@ class TestVSCodeExtractPackageArgs(unittest.TestCase): """Test _extract_package_args with both API formats.""" def setUp(self): - self.mock_registry_patcher = patch('apm_cli.adapters.client.vscode.SimpleRegistryClient') + self.mock_registry_patcher = patch( + "apm_cli.adapters.client.vscode.SimpleRegistryClient" + ) self.mock_registry_patcher.start() - self.mock_integration_patcher = patch('apm_cli.adapters.client.vscode.RegistryIntegration') + self.mock_integration_patcher = patch( + "apm_cli.adapters.client.vscode.RegistryIntegration" + ) self.mock_integration_patcher.start() def tearDown(self): @@ -581,7 +678,9 @@ def test_package_arguments_api_format(self): {"type": "positional", "value": "start"}, ], } - self.assertEqual(VSCodeClientAdapter._extract_package_args(pkg), ["server", "start"]) + self.assertEqual( + VSCodeClientAdapter._extract_package_args(pkg), ["server", "start"] + ) def test_runtime_arguments_legacy_format(self): pkg = { @@ -591,7 +690,9 @@ def test_runtime_arguments_legacy_format(self): {"is_required": True, "value_hint": "start"}, ], } - self.assertEqual(VSCodeClientAdapter._extract_package_args(pkg), ["server", "start"]) + self.assertEqual( + VSCodeClientAdapter._extract_package_args(pkg), ["server", "start"] + ) def test_prefers_package_arguments_over_runtime(self): pkg = { @@ -610,27 +711,29 @@ class TestVSCodeRealApiFormat(unittest.TestCase): """Test with the actual MCP registry API response format (empty registry_name, package_arguments).""" def setUp(self): - self.temp_dir = tempfile.TemporaryDirectory() - self.vscode_dir = os.path.join(self.temp_dir.name, ".vscode") + self.temp_dir = tempfile.mkdtemp() + self.vscode_dir = os.path.join(self.temp_dir, ".vscode") os.makedirs(self.vscode_dir, exist_ok=True) self.temp_path = os.path.join(self.vscode_dir, "mcp.json") with open(self.temp_path, "w") as f: json.dump({"servers": {}}, f) - self.mock_registry_patcher = patch('apm_cli.adapters.client.vscode.SimpleRegistryClient') + self.mock_registry_patcher = patch( + "apm_cli.adapters.client.vscode.SimpleRegistryClient" + ) self.mock_registry_class = self.mock_registry_patcher.start() self.mock_registry = MagicMock() self.mock_registry_class.return_value = self.mock_registry - self.mock_integration_patcher = patch('apm_cli.adapters.client.vscode.RegistryIntegration') + self.mock_integration_patcher = patch( + "apm_cli.adapters.client.vscode.RegistryIntegration" + ) self.mock_integration_patcher.start() def tearDown(self): - import gc; gc.collect() - import time; time.sleep(0.1) self.mock_registry_patcher.stop() self.mock_integration_patcher.stop() - self.temp_dir.cleanup() + shutil.rmtree(self.temp_dir, ignore_errors=True) @patch("apm_cli.adapters.client.vscode.VSCodeClientAdapter.get_config_path") def test_azure_mcp_real_api_format(self, mock_get_path): @@ -729,19 +832,23 @@ class TestExtractInputVariables(unittest.TestCase): """Tests for ${input:...} variable extraction in self-defined MCP servers.""" def setUp(self): - self.temp_dir = tempfile.TemporaryDirectory() - self.vscode_dir = os.path.join(self.temp_dir.name, ".vscode") + self.temp_dir = tempfile.mkdtemp() + self.vscode_dir = os.path.join(self.temp_dir, ".vscode") os.makedirs(self.vscode_dir, exist_ok=True) self.temp_path = os.path.join(self.vscode_dir, "mcp.json") with open(self.temp_path, "w") as f: json.dump({"servers": {}, "inputs": []}, f) - self.mock_registry_patcher = patch('apm_cli.adapters.client.vscode.SimpleRegistryClient') + self.mock_registry_patcher = patch( + "apm_cli.adapters.client.vscode.SimpleRegistryClient" + ) self.mock_registry_class = self.mock_registry_patcher.start() self.mock_registry = MagicMock() self.mock_registry_class.return_value = self.mock_registry - self.mock_integration_patcher = patch('apm_cli.adapters.client.vscode.RegistryIntegration') + self.mock_integration_patcher = patch( + "apm_cli.adapters.client.vscode.RegistryIntegration" + ) self.mock_integration_class = self.mock_integration_patcher.start() self.mock_integration = MagicMock() self.mock_integration_class.return_value = self.mock_integration @@ -749,7 +856,7 @@ def setUp(self): def tearDown(self): self.mock_registry_patcher.stop() self.mock_integration_patcher.stop() - self.temp_dir.cleanup() + shutil.rmtree(self.temp_dir, ignore_errors=True) def test_extract_single_input_variable(self): adapter = VSCodeClientAdapter() @@ -810,7 +917,10 @@ def test_self_defined_http_headers_generate_inputs(self, mock_get_path): "transport_type": "http", "url": "https://my-server.example.com/mcp/", "headers": [ - {"name": "Authorization", "value": "Bearer ${input:my-server-token}"}, + { + "name": "Authorization", + "value": "Bearer ${input:my-server-token}", + }, {"name": "X-Project", "value": "${input:my-server-project}"}, ], } @@ -851,9 +961,7 @@ def test_self_defined_stdio_env_generates_inputs(self, mock_get_path): self.mock_registry.find_server_by_reference.return_value = server_info adapter = VSCodeClientAdapter() - result = adapter.configure_mcp_server( - server_url="my-cli", server_name="my-cli" - ) + result = adapter.configure_mcp_server(server_url="my-cli", server_name="my-cli") assert result is True with open(self.temp_path, "r") as f: @@ -874,7 +982,12 @@ def test_input_variables_dedup_across_servers(self, mock_get_path): { "servers": {}, "inputs": [ - {"type": "promptString", "id": "my-server-token", "description": "existing", "password": True} + { + "type": "promptString", + "id": "my-server-token", + "description": "existing", + "password": True, + } ], }, f, @@ -887,7 +1000,10 @@ def test_input_variables_dedup_across_servers(self, mock_get_path): "transport_type": "http", "url": "https://example.com/mcp/", "headers": [ - {"name": "Authorization", "value": "Bearer ${input:my-server-token}"}, + { + "name": "Authorization", + "value": "Bearer ${input:my-server-token}", + }, ], } ], @@ -907,7 +1023,9 @@ def test_input_variables_dedup_across_servers(self, mock_get_path): class TestWarnInputVariables(unittest.TestCase): """Tests for _warn_input_variables on adapters that don't support input prompts.""" - def test_warning_emitted_for_input_reference(self, ): + def test_warning_emitted_for_input_reference( + self, + ): mapping = {"Authorization": "Bearer ${input:my-token}"} with patch("builtins.print") as mock_print: MCPClientAdapter._warn_input_variables(mapping, "my-server", "Copilot CLI") diff --git a/tests/unit/workflow/test_workflow.py b/tests/unit/workflow/test_workflow.py index 8179efb1..1b84bc5a 100644 --- a/tests/unit/workflow/test_workflow.py +++ b/tests/unit/workflow/test_workflow.py @@ -1,49 +1,26 @@ """Unit tests for workflow functionality.""" import os +import shutil import tempfile import unittest -import time -import shutil -import gc -import sys + +from apm_cli.workflow.discovery import create_workflow_template, discover_workflows from apm_cli.workflow.parser import WorkflowDefinition, parse_workflow_file -from apm_cli.workflow.runner import substitute_parameters, collect_parameters -from apm_cli.workflow.discovery import discover_workflows, create_workflow_template - - -def safe_rmdir(path): - """Safely remove a directory with retry logic for Windows. - - Args: - path (str): Path to directory to remove - """ - try: - shutil.rmtree(path) - except PermissionError: - # On Windows, give time for any lingering processes to release the lock - time.sleep(0.5) - gc.collect() # Force garbage collection to release file handles - try: - shutil.rmtree(path) - except PermissionError as e: - print(f"Failed to remove directory {path}: {e}") - # Continue without failing the test - pass +from apm_cli.workflow.runner import collect_parameters, substitute_parameters class TestWorkflowParser(unittest.TestCase): """Test cases for the workflow parser.""" - + def setUp(self): """Set up test fixtures.""" - self.temp_dir = tempfile.TemporaryDirectory() - self.temp_dir_path = self.temp_dir.name + self.temp_dir_path = tempfile.mkdtemp() # Create .github/prompts directory structure self.prompts_dir = os.path.join(self.temp_dir_path, ".github", "prompts") os.makedirs(self.prompts_dir, exist_ok=True) self.temp_path = os.path.join(self.prompts_dir, "test-workflow.prompt.md") - + # Create a test workflow file with open(self.temp_path, "w") as f: f.write("""--- @@ -61,70 +38,44 @@ def setUp(self): 1. Step One: ${input:param1} 2. Step Two: ${input:param2} """) - + def tearDown(self): """Tear down test fixtures.""" - # Force garbage collection to release file handles - gc.collect() - - # Give time for Windows to release locks - if sys.platform == 'win32': - time.sleep(0.1) - - # First, try the standard cleanup - try: - self.temp_dir.cleanup() - except PermissionError: - # If standard cleanup fails on Windows, use our safe_rmdir function - if hasattr(self, 'temp_dir_path') and os.path.exists(self.temp_dir_path): - safe_rmdir(self.temp_dir_path) - + shutil.rmtree(self.temp_dir_path, ignore_errors=True) + def test_parse_workflow_file(self): """Test parsing a workflow file.""" workflow = parse_workflow_file(self.temp_path) - + self.assertEqual(workflow.name, "test-workflow") self.assertEqual(workflow.description, "Test workflow") self.assertEqual(workflow.author, "Test Author") self.assertEqual(workflow.mcp_dependencies, ["test-package"]) self.assertEqual(workflow.input_parameters, ["param1", "param2"]) self.assertIn("# Test Workflow", workflow.content) - + def test_workflow_validation(self): """Test workflow validation.""" # Valid workflow workflow = WorkflowDefinition( "test", ".github/prompts/test.prompt.md", - { - "description": "Test", - "input": ["param1"] - }, - "content" + {"description": "Test", "input": ["param1"]}, + "content", ) self.assertEqual(workflow.validate(), []) - + # Invalid workflow - missing description workflow = WorkflowDefinition( - "test", - ".github/prompts/test.prompt.md", - { - "input": ["param1"] - }, - "content" + "test", ".github/prompts/test.prompt.md", {"input": ["param1"]}, "content" ) errors = workflow.validate() self.assertEqual(len(errors), 1) self.assertIn("description", errors[0]) - + # Input parameters are now optional, so this should not report an error workflow = WorkflowDefinition( - "test", - ".github/prompts/test.prompt.md", - { - "description": "Test" - }, - "content" + "test", ".github/prompts/test.prompt.md", {"description": "Test"}, "content" ) errors = workflow.validate() self.assertEqual(len(errors), 0) # Expecting 0 errors as input is optional @@ -132,41 +83,35 @@ def test_workflow_validation(self): class TestWorkflowRunner(unittest.TestCase): """Test cases for the workflow runner.""" - + def test_parameter_substitution(self): """Test parameter substitution.""" content = "This is a test with ${input:param1} and ${input:param2}." - params = { - "param1": "value1", - "param2": "value2" - } - + params = {"param1": "value1", "param2": "value2"} + result = substitute_parameters(content, params) self.assertEqual(result, "This is a test with value1 and value2.") - + def test_parameter_substitution_with_missing_params(self): """Test parameter substitution with missing parameters.""" content = "This is a test with ${input:param1} and ${input:param2}." - params = { - "param1": "value1" - } - + params = {"param1": "value1"} + result = substitute_parameters(content, params) self.assertEqual(result, "This is a test with value1 and ${input:param2}.") class TestWorkflowDiscovery(unittest.TestCase): """Test cases for workflow discovery.""" - + def setUp(self): """Set up test fixtures.""" - self.temp_dir = tempfile.TemporaryDirectory() - self.temp_dir_path = self.temp_dir.name - + self.temp_dir_path = tempfile.mkdtemp() + # Create .github/prompts directory structure self.prompts_dir = os.path.join(self.temp_dir_path, ".github", "prompts") os.makedirs(self.prompts_dir, exist_ok=True) - + # Create a few test workflow files self.workflow1_path = os.path.join(self.prompts_dir, "workflow1.prompt.md") with open(self.workflow1_path, "w") as f: @@ -177,7 +122,7 @@ def setUp(self): --- # Workflow 1 """) - + self.workflow2_path = os.path.join(self.prompts_dir, "workflow2.prompt.md") with open(self.workflow2_path, "w") as f: f.write("""--- @@ -187,36 +132,23 @@ def setUp(self): --- # Workflow 2 """) - + def tearDown(self): """Tear down test fixtures.""" - # Force garbage collection to release file handles - gc.collect() - - # Give time for Windows to release locks - if sys.platform == 'win32': - time.sleep(0.1) - - # First, try the standard cleanup - try: - self.temp_dir.cleanup() - except PermissionError: - # If standard cleanup fails on Windows, use our safe_rmdir function - if hasattr(self, 'temp_dir_path') and os.path.exists(self.temp_dir_path): - safe_rmdir(self.temp_dir_path) - + shutil.rmtree(self.temp_dir_path, ignore_errors=True) + def test_discover_workflows(self): """Test discovering workflows.""" workflows = discover_workflows(self.temp_dir_path) - + self.assertEqual(len(workflows), 2) self.assertIn("workflow1", [w.name for w in workflows]) self.assertIn("workflow2", [w.name for w in workflows]) - + def test_create_workflow_template(self): """Test creating a workflow template.""" template_path = create_workflow_template("test-template", self.temp_dir_path) - + self.assertTrue(os.path.exists(template_path)) with open(template_path, "r") as f: content = f.read() @@ -228,4 +160,4 @@ def test_create_workflow_template(self): if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main()