216 lines
9.0 KiB
Python
216 lines
9.0 KiB
Python
import pytest
|
|
from unittest.mock import AsyncMock
|
|
from datetime import datetime
|
|
from app.backend.services.deployment_status_update_service import DeploymentStatusUpdateService
|
|
from app.common.models.deployment.deployment import Deployment
|
|
|
|
|
|
@pytest.fixture
|
|
def status_update_service():
|
|
return DeploymentStatusUpdateService()
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_heartbeat_message():
|
|
return {
|
|
"event_type": "DevOpsReconcileJobHeartbeat",
|
|
"payload": {
|
|
"operation": "heartbeat",
|
|
"id": "deployment-123-abc",
|
|
"status": "running",
|
|
"phase": "building",
|
|
"phase_message": "Building container image",
|
|
"error": None,
|
|
"url": None
|
|
}
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_success_message():
|
|
return {
|
|
"event_type": "DevOpsReconcileJobHeartbeat",
|
|
"payload": {
|
|
"operation": "heartbeat",
|
|
"id": "deployment-789-ghi",
|
|
"status": "success",
|
|
"phase": "finished",
|
|
"phase_message": "Deployment completed successfully",
|
|
"error": None,
|
|
"url": "https://my-app-alpha.freeleaps.com"
|
|
}
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_failed_message():
|
|
return {
|
|
"event_type": "DevOpsReconcileJobHeartbeat",
|
|
"payload": {
|
|
"operation": "heartbeat",
|
|
"id": "deployment-456-def",
|
|
"status": "failed",
|
|
"phase": "jenkins_build",
|
|
"phase_message": "Build failed due to compilation errors",
|
|
"error": "Build step 'Invoke top-level Maven targets' marked build as failure",
|
|
"url": None
|
|
}
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_deployment():
|
|
from unittest.mock import AsyncMock
|
|
|
|
class MockDeployment:
|
|
def __init__(self):
|
|
self.deployment_id = "deployment-123-abc"
|
|
self.deployment_status = "started"
|
|
self.deployment_stage = "initialization"
|
|
self.deployment_app_url = ""
|
|
self.updated_at = datetime.now()
|
|
self.save = AsyncMock()
|
|
|
|
return MockDeployment()
|
|
|
|
|
|
class TestDeploymentStatusUpdateService:
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_mapping(self, status_update_service):
|
|
"""Test that status mapping works correctly"""
|
|
assert status_update_service.status_mapping["running"] == "started"
|
|
assert status_update_service.status_mapping["success"] == "succeeded"
|
|
assert status_update_service.status_mapping["failed"] == "failed"
|
|
assert status_update_service.status_mapping["terminated"] == "aborted"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_phase_to_stage_mapping(self, status_update_service):
|
|
"""Test that phase to stage mapping works correctly"""
|
|
assert status_update_service.phase_to_stage_mapping["initializing"] == "initialization"
|
|
assert status_update_service.phase_to_stage_mapping["jenkins_build"] == "build"
|
|
assert status_update_service.phase_to_stage_mapping["building"] == "build"
|
|
assert status_update_service.phase_to_stage_mapping["deploying"] == "deployment"
|
|
assert status_update_service.phase_to_stage_mapping["finished"] == "completed"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_process_running_heartbeat_message(self, status_update_service, sample_heartbeat_message, mock_deployment, monkeypatch):
|
|
"""Test processing a running status heartbeat"""
|
|
# Mock Deployment.find_one to return our mock deployment
|
|
async def mock_find_one(query):
|
|
_ = query # Parameter required by interface but not used in mock
|
|
return mock_deployment
|
|
|
|
# Mock the logger methods to avoid actual logging during tests
|
|
status_update_service.module_logger.log_info = AsyncMock()
|
|
status_update_service.module_logger.log_warning = AsyncMock()
|
|
status_update_service.module_logger.log_error = AsyncMock()
|
|
status_update_service.module_logger.log_exception = AsyncMock()
|
|
|
|
# Mock the Beanie query mechanism properly
|
|
mock_deployment_class = AsyncMock()
|
|
mock_deployment_class.find_one = mock_find_one
|
|
monkeypatch.setattr("app.backend.services.deployment_status_update_service.Deployment", mock_deployment_class)
|
|
|
|
await status_update_service.process_heartbeat_message(
|
|
"test_key", sample_heartbeat_message, {}
|
|
)
|
|
|
|
# Verify the deployment was updated correctly
|
|
assert mock_deployment.deployment_status == "started"
|
|
assert mock_deployment.deployment_stage == "build"
|
|
mock_deployment.save.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_process_success_heartbeat_message(self, status_update_service, sample_success_message, mock_deployment, monkeypatch):
|
|
"""Test processing a success status heartbeat with URL"""
|
|
async def mock_find_one(query):
|
|
_ = query # Parameter required by interface but not used in mock
|
|
return mock_deployment
|
|
|
|
# Mock the logger methods
|
|
status_update_service.module_logger.log_info = AsyncMock()
|
|
status_update_service.module_logger.log_warning = AsyncMock()
|
|
status_update_service.module_logger.log_error = AsyncMock()
|
|
status_update_service.module_logger.log_exception = AsyncMock()
|
|
|
|
# Mock the Beanie query mechanism properly
|
|
mock_deployment_class = AsyncMock()
|
|
mock_deployment_class.find_one = mock_find_one
|
|
monkeypatch.setattr("app.backend.services.deployment_status_update_service.Deployment", mock_deployment_class)
|
|
|
|
await status_update_service.process_heartbeat_message(
|
|
"test_key", sample_success_message, {}
|
|
)
|
|
|
|
# Verify the deployment was updated correctly
|
|
assert mock_deployment.deployment_status == "succeeded"
|
|
assert mock_deployment.deployment_stage == "completed"
|
|
assert mock_deployment.deployment_app_url == "https://my-app-alpha.freeleaps.com"
|
|
mock_deployment.save.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_process_failed_heartbeat_message(self, status_update_service, sample_failed_message, mock_deployment, monkeypatch):
|
|
"""Test processing a failed status heartbeat"""
|
|
async def mock_find_one(query):
|
|
_ = query # Parameter required by interface but not used in mock
|
|
return mock_deployment
|
|
|
|
# Mock the logger methods
|
|
status_update_service.module_logger.log_info = AsyncMock()
|
|
status_update_service.module_logger.log_warning = AsyncMock()
|
|
status_update_service.module_logger.log_error = AsyncMock()
|
|
status_update_service.module_logger.log_exception = AsyncMock()
|
|
|
|
# Mock the Beanie query mechanism properly
|
|
mock_deployment_class = AsyncMock()
|
|
mock_deployment_class.find_one = mock_find_one
|
|
monkeypatch.setattr("app.backend.services.deployment_status_update_service.Deployment", mock_deployment_class)
|
|
|
|
await status_update_service.process_heartbeat_message(
|
|
"test_key", sample_failed_message, {}
|
|
)
|
|
|
|
# Verify the deployment was updated correctly
|
|
assert mock_deployment.deployment_status == "failed"
|
|
assert mock_deployment.deployment_stage == "build"
|
|
mock_deployment.save.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deployment_not_found(self, status_update_service, sample_heartbeat_message, monkeypatch):
|
|
"""Test handling when deployment is not found"""
|
|
async def mock_find_one(query):
|
|
_ = query # Parameter required by interface but not used in mock
|
|
return None
|
|
|
|
# Mock the logger methods
|
|
status_update_service.module_logger.log_info = AsyncMock()
|
|
status_update_service.module_logger.log_warning = AsyncMock()
|
|
status_update_service.module_logger.log_error = AsyncMock()
|
|
status_update_service.module_logger.log_exception = AsyncMock()
|
|
|
|
# Mock the Beanie query mechanism properly
|
|
mock_deployment_class = AsyncMock()
|
|
mock_deployment_class.find_one = mock_find_one
|
|
monkeypatch.setattr("app.backend.services.deployment_status_update_service.Deployment", mock_deployment_class)
|
|
|
|
# Should not raise an exception
|
|
await status_update_service.process_heartbeat_message(
|
|
"test_key", sample_heartbeat_message, {}
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_message_format(self, status_update_service):
|
|
"""Test handling invalid message format"""
|
|
invalid_message = {"invalid": "format"}
|
|
|
|
# Mock the logger methods
|
|
status_update_service.module_logger.log_info = AsyncMock()
|
|
status_update_service.module_logger.log_warning = AsyncMock()
|
|
status_update_service.module_logger.log_error = AsyncMock()
|
|
status_update_service.module_logger.log_exception = AsyncMock()
|
|
|
|
# Should not raise an exception due to try/catch in the method
|
|
await status_update_service.process_heartbeat_message(
|
|
"test_key", invalid_message, {}
|
|
) |