404 lines
16 KiB
Python
404 lines
16 KiB
Python
"""Unit tests for EnvInspectExecutor."""
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
# Patch the logger before importing the executor
|
|
with patch("app.utils.logger.get_logger", return_value=MagicMock()):
|
|
from app.executors.env_inspect_executor import EnvInspectExecutor
|
|
|
|
|
|
class TestEnvInspectExecutor:
|
|
"""Test suite for EnvInspectExecutor."""
|
|
|
|
@pytest.fixture
|
|
def executor(self):
|
|
"""Create executor instance with mocked logger."""
|
|
with patch("app.executors.base.get_logger", return_value=MagicMock()):
|
|
return EnvInspectExecutor()
|
|
|
|
@pytest.fixture
|
|
def temp_env_root(self, tmp_path):
|
|
"""Create a temporary directory to act as /opt/letsbe/env."""
|
|
env_dir = tmp_path / "opt" / "letsbe" / "env"
|
|
env_dir.mkdir(parents=True)
|
|
return env_dir
|
|
|
|
@pytest.fixture
|
|
def mock_settings(self, temp_env_root):
|
|
"""Mock settings with temporary env root."""
|
|
settings = MagicMock()
|
|
settings.allowed_env_root = str(temp_env_root)
|
|
return settings
|
|
|
|
# ==================== Basic Inspection Tests ====================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inspect_all_keys(self, executor, temp_env_root, mock_settings):
|
|
"""Test reading all keys when no filter is provided."""
|
|
with patch("app.executors.env_inspect_executor.get_settings", return_value=mock_settings):
|
|
env_path = temp_env_root / "app.env"
|
|
env_path.write_text("KEY1=value1\nKEY2=value2\nKEY3=value3\n")
|
|
|
|
result = await executor.execute({
|
|
"path": str(env_path),
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["path"] == str(env_path)
|
|
assert result.data["keys"] == {
|
|
"KEY1": "value1",
|
|
"KEY2": "value2",
|
|
"KEY3": "value3",
|
|
}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inspect_selected_keys(self, executor, temp_env_root, mock_settings):
|
|
"""Test reading only selected keys."""
|
|
with patch("app.executors.env_inspect_executor.get_settings", return_value=mock_settings):
|
|
env_path = temp_env_root / "app.env"
|
|
env_path.write_text("KEY1=value1\nKEY2=value2\nKEY3=value3\n")
|
|
|
|
result = await executor.execute({
|
|
"path": str(env_path),
|
|
"keys": ["KEY1", "KEY3"],
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["keys"] == {
|
|
"KEY1": "value1",
|
|
"KEY3": "value3",
|
|
}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inspect_selected_keys_ignores_unknown(self, executor, temp_env_root, mock_settings):
|
|
"""Test that unknown keys in filter are silently ignored."""
|
|
with patch("app.executors.env_inspect_executor.get_settings", return_value=mock_settings):
|
|
env_path = temp_env_root / "app.env"
|
|
env_path.write_text("KEY1=value1\nKEY2=value2\n")
|
|
|
|
result = await executor.execute({
|
|
"path": str(env_path),
|
|
"keys": ["KEY1", "NONEXISTENT", "ALSO_MISSING"],
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["keys"] == {"KEY1": "value1"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inspect_empty_keys_filter(self, executor, temp_env_root, mock_settings):
|
|
"""Test with empty keys filter returns nothing."""
|
|
with patch("app.executors.env_inspect_executor.get_settings", return_value=mock_settings):
|
|
env_path = temp_env_root / "app.env"
|
|
env_path.write_text("KEY1=value1\n")
|
|
|
|
result = await executor.execute({
|
|
"path": str(env_path),
|
|
"keys": [],
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["keys"] == {}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inspect_with_keys_null(self, executor, temp_env_root, mock_settings):
|
|
"""Test that keys=null returns all keys (same as omitting keys)."""
|
|
with patch("app.executors.env_inspect_executor.get_settings", return_value=mock_settings):
|
|
env_path = temp_env_root / "app.env"
|
|
env_path.write_text("KEY1=value1\nKEY2=value2\n")
|
|
|
|
result = await executor.execute({
|
|
"path": str(env_path),
|
|
"keys": None,
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["keys"] == {"KEY1": "value1", "KEY2": "value2"}
|
|
|
|
# ==================== File Not Found Tests ====================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_file(self, executor, temp_env_root, mock_settings):
|
|
"""Test error when file does not exist."""
|
|
with patch("app.executors.env_inspect_executor.get_settings", return_value=mock_settings):
|
|
env_path = temp_env_root / "nonexistent.env"
|
|
|
|
result = await executor.execute({
|
|
"path": str(env_path),
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "does not exist" in result.error or "Path validation failed" in result.error
|
|
|
|
# ==================== Path Validation Tests ====================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_path_traversal_rejected(self, executor, temp_env_root, mock_settings):
|
|
"""Test rejection of directory traversal attempts."""
|
|
with patch("app.executors.env_inspect_executor.get_settings", return_value=mock_settings):
|
|
result = await executor.execute({
|
|
"path": str(temp_env_root / ".." / ".." / "etc" / "passwd"),
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "Path validation failed" in result.error or "traversal" in result.error.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_path_outside_allowed_root(self, executor, temp_env_root, mock_settings):
|
|
"""Test rejection of paths outside allowed root."""
|
|
with patch("app.executors.env_inspect_executor.get_settings", return_value=mock_settings):
|
|
result = await executor.execute({
|
|
"path": "/etc/passwd",
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "Path validation failed" in result.error
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_path_in_allowed_root_nested(self, executor, temp_env_root, mock_settings):
|
|
"""Test acceptance of valid nested path within allowed root."""
|
|
with patch("app.executors.env_inspect_executor.get_settings", return_value=mock_settings):
|
|
nested_dir = temp_env_root / "subdir"
|
|
nested_dir.mkdir()
|
|
env_path = nested_dir / "app.env"
|
|
env_path.write_text("KEY=value\n")
|
|
|
|
result = await executor.execute({
|
|
"path": str(env_path),
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["keys"] == {"KEY": "value"}
|
|
|
|
# ==================== Payload Validation Tests ====================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_path_payload(self, executor):
|
|
"""Test rejection of payload without path."""
|
|
with pytest.raises(ValueError, match="Missing required field: path"):
|
|
await executor.execute({})
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_path_payload_with_keys(self, executor):
|
|
"""Test rejection of payload with keys but no path."""
|
|
with pytest.raises(ValueError, match="Missing required field: path"):
|
|
await executor.execute({
|
|
"keys": ["KEY1"],
|
|
})
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reject_invalid_keys_type(self, executor, temp_env_root, mock_settings):
|
|
"""Test rejection when keys is not a list or null."""
|
|
with patch("app.executors.env_inspect_executor.get_settings", return_value=mock_settings):
|
|
env_path = temp_env_root / "app.env"
|
|
env_path.write_text("KEY=value\n")
|
|
|
|
result = await executor.execute({
|
|
"path": str(env_path),
|
|
"keys": {"not": "a_list"},
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "'keys' must be a list" in result.error
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reject_keys_as_string(self, executor, temp_env_root, mock_settings):
|
|
"""Test rejection when keys is a string instead of list."""
|
|
with patch("app.executors.env_inspect_executor.get_settings", return_value=mock_settings):
|
|
env_path = temp_env_root / "app.env"
|
|
env_path.write_text("KEY=value\n")
|
|
|
|
result = await executor.execute({
|
|
"path": str(env_path),
|
|
"keys": "KEY",
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "'keys' must be a list" in result.error
|
|
|
|
# ==================== Task Type Tests ====================
|
|
|
|
def test_task_type_property(self, executor):
|
|
"""Test that task_type returns ENV_INSPECT."""
|
|
assert executor.task_type == "ENV_INSPECT"
|
|
|
|
# ==================== Registry Integration Tests ====================
|
|
|
|
def test_registry_integration(self):
|
|
"""Test that ENV_INSPECT is registered in executor registry."""
|
|
from app.executors import get_executor
|
|
|
|
executor = get_executor("ENV_INSPECT")
|
|
assert executor is not None
|
|
assert executor.task_type == "ENV_INSPECT"
|
|
|
|
# ==================== Empty File Tests ====================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_file(self, executor, temp_env_root, mock_settings):
|
|
"""Test reading an empty ENV file returns empty keys dict."""
|
|
with patch("app.executors.env_inspect_executor.get_settings", return_value=mock_settings):
|
|
env_path = temp_env_root / "empty.env"
|
|
env_path.write_text("")
|
|
|
|
result = await executor.execute({
|
|
"path": str(env_path),
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["keys"] == {}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_with_only_comments(self, executor, temp_env_root, mock_settings):
|
|
"""Test reading file with only comments returns empty keys dict."""
|
|
with patch("app.executors.env_inspect_executor.get_settings", return_value=mock_settings):
|
|
env_path = temp_env_root / "comments.env"
|
|
env_path.write_text("# This is a comment\n# Another comment\n")
|
|
|
|
result = await executor.execute({
|
|
"path": str(env_path),
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["keys"] == {}
|
|
|
|
# ==================== Parsing Tests ====================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parses_quoted_values_double(self, executor, temp_env_root, mock_settings):
|
|
"""Test parsing of double-quoted values."""
|
|
with patch("app.executors.env_inspect_executor.get_settings", return_value=mock_settings):
|
|
env_path = temp_env_root / "app.env"
|
|
env_path.write_text('KEY="value with spaces"\n')
|
|
|
|
result = await executor.execute({
|
|
"path": str(env_path),
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["keys"]["KEY"] == "value with spaces"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parses_quoted_values_single(self, executor, temp_env_root, mock_settings):
|
|
"""Test parsing of single-quoted values."""
|
|
with patch("app.executors.env_inspect_executor.get_settings", return_value=mock_settings):
|
|
env_path = temp_env_root / "app.env"
|
|
env_path.write_text("KEY='single quoted'\n")
|
|
|
|
result = await executor.execute({
|
|
"path": str(env_path),
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["keys"]["KEY"] == "single quoted"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parses_value_with_equals(self, executor, temp_env_root, mock_settings):
|
|
"""Test parsing values containing equals signs."""
|
|
with patch("app.executors.env_inspect_executor.get_settings", return_value=mock_settings):
|
|
env_path = temp_env_root / "app.env"
|
|
env_path.write_text("URL=postgres://user:pass@host/db?opt=val\n")
|
|
|
|
result = await executor.execute({
|
|
"path": str(env_path),
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["keys"]["URL"] == "postgres://user:pass@host/db?opt=val"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ignores_comment_lines(self, executor, temp_env_root, mock_settings):
|
|
"""Test that comment lines are ignored."""
|
|
with patch("app.executors.env_inspect_executor.get_settings", return_value=mock_settings):
|
|
env_path = temp_env_root / "app.env"
|
|
env_path.write_text("# Comment\nKEY=value\n# Another comment\nKEY2=value2\n")
|
|
|
|
result = await executor.execute({
|
|
"path": str(env_path),
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["keys"] == {"KEY": "value", "KEY2": "value2"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ignores_empty_lines(self, executor, temp_env_root, mock_settings):
|
|
"""Test that empty lines are ignored."""
|
|
with patch("app.executors.env_inspect_executor.get_settings", return_value=mock_settings):
|
|
env_path = temp_env_root / "app.env"
|
|
env_path.write_text("KEY1=value1\n\n\nKEY2=value2\n")
|
|
|
|
result = await executor.execute({
|
|
"path": str(env_path),
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["keys"] == {"KEY1": "value1", "KEY2": "value2"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handles_whitespace_around_key_value(self, executor, temp_env_root, mock_settings):
|
|
"""Test handling of whitespace around keys and values."""
|
|
with patch("app.executors.env_inspect_executor.get_settings", return_value=mock_settings):
|
|
env_path = temp_env_root / "app.env"
|
|
env_path.write_text(" KEY1 = value1 \n")
|
|
|
|
result = await executor.execute({
|
|
"path": str(env_path),
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["keys"]["KEY1"] == "value1"
|
|
|
|
|
|
class TestEnvInspectExecutorInternal:
|
|
"""Tests for internal methods of EnvInspectExecutor."""
|
|
|
|
@pytest.fixture
|
|
def executor(self):
|
|
"""Create executor instance with mocked logger."""
|
|
with patch("app.executors.base.get_logger", return_value=MagicMock()):
|
|
return EnvInspectExecutor()
|
|
|
|
def test_parse_env_file_basic(self, executor):
|
|
"""Test basic ENV file parsing."""
|
|
content = "KEY1=value1\nKEY2=value2\n"
|
|
result = executor._parse_env_file(content)
|
|
assert result == {"KEY1": "value1", "KEY2": "value2"}
|
|
|
|
def test_parse_env_file_with_comments(self, executor):
|
|
"""Test parsing ignores comments."""
|
|
content = "# Comment\nKEY=value\n# Another comment\n"
|
|
result = executor._parse_env_file(content)
|
|
assert result == {"KEY": "value"}
|
|
|
|
def test_parse_env_file_with_empty_lines(self, executor):
|
|
"""Test parsing ignores empty lines."""
|
|
content = "KEY1=value1\n\n\nKEY2=value2\n"
|
|
result = executor._parse_env_file(content)
|
|
assert result == {"KEY1": "value1", "KEY2": "value2"}
|
|
|
|
def test_parse_env_file_with_quotes(self, executor):
|
|
"""Test parsing handles quoted values."""
|
|
content = 'KEY1="quoted value"\nKEY2=\'single quoted\'\n'
|
|
result = executor._parse_env_file(content)
|
|
assert result == {"KEY1": "quoted value", "KEY2": "single quoted"}
|
|
|
|
def test_parse_env_file_with_equals_in_value(self, executor):
|
|
"""Test parsing handles equals signs in values."""
|
|
content = "URL=postgres://user:pass@host/db?opt=val\n"
|
|
result = executor._parse_env_file(content)
|
|
assert result == {"URL": "postgres://user:pass@host/db?opt=val"}
|
|
|
|
def test_parse_env_file_empty(self, executor):
|
|
"""Test parsing empty content."""
|
|
content = ""
|
|
result = executor._parse_env_file(content)
|
|
assert result == {}
|
|
|
|
def test_parse_env_file_only_comments(self, executor):
|
|
"""Test parsing content with only comments."""
|
|
content = "# Comment 1\n# Comment 2\n"
|
|
result = executor._parse_env_file(content)
|
|
assert result == {}
|