swf-testbed/tests/test_main.py

0001 import pytest
0002 from typer.testing import CliRunner
0003 from pathlib import Path
0004 import shutil
0005 from unittest.mock import patch
0006
0007 from swf_testbed_cli.main import app
0008
0009 runner = CliRunner()
0010
0011 @pytest.fixture(scope="function")
0012 def test_environment(tmp_path):
0013     """Create a temporary directory for testing."""
0014     # tmp_path is a Path object provided by pytest
0015     # Change the current working directory to the temporary directory
0016     # so that the CLI commands run in a clean environment.
0017     # This is important because the CLI app creates files/dirs in the CWD.
0018     import os
0019     cwd = os.getcwd()
0020     os.chdir(tmp_path)
0021     yield tmp_path
0022     # Teardown: clean up by changing back to the original directory
0023     os.chdir(cwd)
0024
0025 @patch('swf_testbed_cli.main._check_supervisord_running', return_value=False)
0026 @patch('subprocess.run')
0027 def test_start(mock_run, mock_check_supervisord, test_environment):
0028     """Test the start command."""
0029     # Arrange
0030     (test_environment / "supervisord.conf").touch()
0031     (test_environment / "docker-compose.yml").touch()
0032     mock_run.return_value.returncode = 0
0033
0034     # Act
0035     result = runner.invoke(app, ["start"])
0036
0037     # Assert
0038     assert result.exit_code == 0
0039     assert mock_run.call_count == 3  # docker compose up, supervisord start, supervisorctl start
0040     mock_run.assert_any_call(["docker", "compose", "up", "-d"])
0041     mock_run.assert_any_call(["supervisord", "-c", "supervisord.conf"])
0042     mock_run.assert_any_call(["supervisorctl", "-c", "supervisord.conf", "start", "all"])
0043     assert "Starting testbed services..." in result.stdout
0044
0045 @patch('subprocess.run')
0046 def test_stop(mock_run, test_environment):
0047     """Test the stop command."""
0048     # Arrange
0049     (test_environment / "supervisord.conf").touch()
0050
0051     # Act
0052     result = runner.invoke(app, ["stop"])
0053
0054     # Assert
0055     assert result.exit_code == 0
0056     assert mock_run.call_count == 2
0057     mock_run.assert_any_call(["supervisorctl", "-c", "supervisord.conf", "stop", "all"])
0058     mock_run.assert_any_call(["docker", "compose", "down"])
0059     assert "Stopping testbed services..." in result.stdout
0060
0061 @patch('subprocess.run')
0062 def test_status(mock_run, test_environment):
0063     """Test the status command."""
0064     # Arrange
0065     (test_environment / "supervisord.conf").touch()
0066
0067     # Act
0068     result = runner.invoke(app, ["status"])
0069
0070     # Assert
0071     assert result.exit_code == 0
0072     assert mock_run.call_count == 2
0073     mock_run.assert_any_call(["docker", "compose", "ps"])
0074     mock_run.assert_any_call(["supervisorctl", "-c", "supervisord.conf", "status"])
0075     assert "--- Docker services status ---" in result.stdout
0076     assert "--- supervisord services status ---" in result.stdout
0077
0078
0079
0080 @patch('swf_testbed_cli.main._check_activemq_connection', return_value=False)
0081 @patch('swf_testbed_cli.main._check_postgres_connection', return_value=True)
0082 @patch('subprocess.run')
0083 def test_start_local_failure(mock_run, mock_check_postgres, mock_check_activemq, test_environment):
0084     """Test the start-local command when a service is not running."""
0085     # Arrange
0086     (test_environment / "supervisord.conf").touch()
0087
0088     # Act
0089     result = runner.invoke(app, ["start-local"])
0090
0091     # Assert
0092     assert result.exit_code != 0
0093     mock_check_postgres.assert_called_once()
0094     mock_check_activemq.assert_called_once()
0095     mock_run.assert_not_called()
0096     assert "Error: One or more background services are not available. Aborting." in result.stdout
0097
0098 @patch('subprocess.run')
0099 def test_stop_local(mock_run, test_environment):
0100     """Test the stop-local command."""
0101     # Arrange
0102     (test_environment / "supervisord.conf").touch()
0103
0104     # Act
0105     result = runner.invoke(app, ["stop-local"])
0106
0107     # Assert
0108     assert result.exit_code == 0
0109     mock_run.assert_called_once_with(["supervisorctl", "-c", "supervisord.conf", "stop", "all"])
0110     assert "--- Stopping local supervisord services ---" in result.stdout
0111
0112