82 lines
2.9 KiB
Python
82 lines
2.9 KiB
Python
|
"""Integration test for DockerExecutor with real Docker."""
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
import sys
|
||
|
|
import tempfile
|
||
|
|
from pathlib import Path
|
||
|
|
from unittest.mock import MagicMock, patch
|
||
|
|
|
||
|
|
# Add parent directory to path for imports
|
||
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
# Create a real temp directory structure
|
||
|
|
with tempfile.TemporaryDirectory() as tmp:
|
||
|
|
stacks_root = Path(tmp) / "stacks"
|
||
|
|
stack_dir = stacks_root / "test-app"
|
||
|
|
stack_dir.mkdir(parents=True)
|
||
|
|
|
||
|
|
# Create a minimal compose file
|
||
|
|
compose_content = """services:
|
||
|
|
test:
|
||
|
|
image: alpine:latest
|
||
|
|
command: echo 'Hello from integration test'
|
||
|
|
"""
|
||
|
|
compose_file = stack_dir / "docker-compose.yml"
|
||
|
|
compose_file.write_text(compose_content)
|
||
|
|
|
||
|
|
print(f"Created stack at: {stack_dir}")
|
||
|
|
print(f"Compose file: {compose_file}")
|
||
|
|
|
||
|
|
# Import executor with mocked logger
|
||
|
|
with patch("app.executors.base.get_logger", return_value=MagicMock()):
|
||
|
|
from app.executors.docker_executor import DockerExecutor
|
||
|
|
executor = DockerExecutor()
|
||
|
|
|
||
|
|
# Mock settings to use our temp directory
|
||
|
|
mock_settings = MagicMock()
|
||
|
|
mock_settings.allowed_stacks_root = str(stacks_root)
|
||
|
|
|
||
|
|
async def run_test():
|
||
|
|
with patch("app.executors.docker_executor.get_settings", return_value=mock_settings):
|
||
|
|
# Test 1: Without pull
|
||
|
|
print("\n=== Test 1: pull=False ===")
|
||
|
|
result = await executor.execute({
|
||
|
|
"compose_dir": str(stack_dir),
|
||
|
|
"pull": False,
|
||
|
|
"timeout": 60,
|
||
|
|
})
|
||
|
|
print(f"Success: {result.success}")
|
||
|
|
print(f"compose_file: {result.data.get('compose_file')}")
|
||
|
|
print(f"pull_ran: {result.data.get('pull_ran')}")
|
||
|
|
if result.error:
|
||
|
|
print(f"Error: {result.error}")
|
||
|
|
up_logs = result.data.get("logs", {}).get("up", "")
|
||
|
|
print(f"Logs (up): {up_logs[:300] if up_logs else 'empty'}")
|
||
|
|
|
||
|
|
# Test 2: With pull
|
||
|
|
print("\n=== Test 2: pull=True ===")
|
||
|
|
result2 = await executor.execute({
|
||
|
|
"compose_dir": str(stack_dir),
|
||
|
|
"pull": True,
|
||
|
|
"timeout": 60,
|
||
|
|
})
|
||
|
|
print(f"Success: {result2.success}")
|
||
|
|
print(f"pull_ran: {result2.data.get('pull_ran')}")
|
||
|
|
pull_logs = result2.data.get("logs", {}).get("pull", "")
|
||
|
|
print(f"Logs (pull): {pull_logs[:300] if pull_logs else 'empty'}")
|
||
|
|
|
||
|
|
return result.success and result2.success
|
||
|
|
|
||
|
|
success = asyncio.run(run_test())
|
||
|
|
print(f"\n{'=' * 50}")
|
||
|
|
print(f"INTEGRATION TEST: {'PASSED' if success else 'FAILED'}")
|
||
|
|
print(f"{'=' * 50}")
|
||
|
|
return success
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
success = main()
|
||
|
|
sys.exit(0 if success else 1)
|