gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: add tests for Phase 1 modules (rate_limit_decorators, monitor_stalled_steps, step_timeout, config_validator)

Browse Source 
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-02-12 03:44:23 +00:00
parent c654b829ac
commit 0eb29fd62a
5 changed files with 769 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_celery_worker.py
+98
View File
@@ -0,0 +1,98 @@
"""
Tests for app/celery_worker.py
This module tests the Celery worker configuration, task imports, and beat schedule.
"""
import pytest
from unittest.mock import MagicMock, patch, PropertyMock
from celery.schedules import crontab
@pytest.mark.unit
class TestCeleryWorkerConfig:
"""Test Celery worker configuration."""
def test_test_task_function(self):
"""Test the test_task function returns expected value."""
from app.celery_worker import test_task
result = test_task()
assert result == "Celery is working!"
def test_celery_instance_exists(self):
"""Test that celery instance exists in module."""
from app import celery_worker
assert hasattr(celery_worker, 'celery')
assert celery_worker.celery is not None
def test_task_routes_exists(self):
"""Test that task routes configuration exists."""
from app import celery_worker
# Task routes should be configured
assert hasattr(celery_worker.celery.conf, 'task_routes')
def test_all_task_imports_successful(self):
"""Test that all task modules are imported successfully."""
# Just import the module to verify no import errors
from app import celery_worker
# Module imported successfully
assert celery_worker is not None
@pytest.mark.unit
class TestBeatScheduleConfiguration:
"""Test Celery beat schedule configuration."""
def test_beat_schedule_structure(self):
"""Test that beat schedule has expected structure."""
from app.celery_worker import celery
# Beat schedule should be a dictionary
assert isinstance(celery.conf.beat_schedule, dict)
# Should include credential check tasks
assert 'check-credentials-regularly' in celery.conf.beat_schedule
assert 'check-credentials-daily' in celery.conf.beat_schedule
assert 'monitor-stalled-steps' in celery.conf.beat_schedule
def test_credential_check_schedule(self):
"""Test credential check schedule configuration."""
from app.celery_worker import celery
schedule = celery.conf.beat_schedule.get('check-credentials-regularly')
assert schedule is not None
assert schedule['task'] == 'app.tasks.check_credentials.check_credentials'
assert 'schedule' in schedule
assert schedule['options']['expires'] == 240
def test_daily_credential_check_schedule(self):
"""Test daily credential check schedule."""
from app.celery_worker import celery
schedule = celery.conf.beat_schedule.get('check-credentials-daily')
assert schedule is not None
assert schedule['task'] == 'app.tasks.check_credentials.check_credentials'
assert 'schedule' in schedule
assert schedule['options']['expires'] == 3600
def test_monitor_stalled_steps_schedule(self):
"""Test monitor stalled steps schedule."""
from app.celery_worker import celery
schedule = celery.conf.beat_schedule.get('monitor-stalled-steps')
assert schedule is not None
assert schedule['task'] == 'app.tasks.monitor_stalled_steps.monitor_stalled_steps'
assert 'schedule' in schedule
assert schedule['options']['expires'] == 55
def test_no_none_entries_in_beat_schedule(self):
"""Test that None entries are filtered from beat schedule."""
from app.celery_worker import celery
# No None values in beat schedule
for key, value in celery.conf.beat_schedule.items():
assert value is not None, f"Beat schedule entry '{key}' should not be None"
tests/test_config_validator_reexport.py
+106
View File
@@ -0,0 +1,106 @@
"""
Tests for app/utils/config_validator.py
This module tests the config_validator re-export module.
"""
import pytest
@pytest.mark.unit
class TestConfigValidatorReexports:
"""Test config_validator re-export module."""
def test_module_imports(self):
"""Test that config_validator module imports successfully."""
from app.utils import config_validator
assert config_validator is not None
def test_validate_email_config_reexport(self):
"""Test validate_email_config is re-exported."""
from app.utils.config_validator import validate_email_config
assert callable(validate_email_config)
def test_validate_storage_configs_reexport(self):
"""Test validate_storage_configs is re-exported."""
from app.utils.config_validator import validate_storage_configs
assert callable(validate_storage_configs)
def test_validate_notification_config_reexport(self):
"""Test validate_notification_config is re-exported."""
from app.utils.config_validator import validate_notification_config
assert callable(validate_notification_config)
def test_mask_sensitive_value_reexport(self):
"""Test mask_sensitive_value is re-exported."""
from app.utils.config_validator import mask_sensitive_value
assert callable(mask_sensitive_value)
def test_get_provider_status_reexport(self):
"""Test get_provider_status is re-exported."""
from app.utils.config_validator import get_provider_status
assert callable(get_provider_status)
def test_get_settings_for_display_reexport(self):
"""Test get_settings_for_display is re-exported."""
from app.utils.config_validator import get_settings_for_display
assert callable(get_settings_for_display)
def test_dump_all_settings_reexport(self):
"""Test dump_all_settings is re-exported."""
from app.utils.config_validator import dump_all_settings
assert callable(dump_all_settings)
def test_check_all_configs_reexport(self):
"""Test check_all_configs is re-exported."""
from app.utils.config_validator import check_all_configs
assert callable(check_all_configs)
def test_all_exports_in_all(self):
"""Test that all exports are in __all__."""
from app.utils import config_validator
expected_exports = [
"validate_email_config",
"validate_storage_configs",
"validate_notification_config",
"mask_sensitive_value",
"get_provider_status",
"get_settings_for_display",
"dump_all_settings",
"check_all_configs",
]
assert hasattr(config_validator, '__all__')
for export in expected_exports:
assert export in config_validator.__all__
def test_mask_sensitive_value_functionality(self):
"""Test mask_sensitive_value actually works."""
from app.utils.config_validator import mask_sensitive_value
# Test masking a sensitive value
result = mask_sensitive_value("secret_api_key_12345")
assert result != "secret_api_key_12345"
assert "***" in result or result == ""
def test_get_provider_status_functionality(self):
"""Test get_provider_status returns expected structure."""
from app.utils.config_validator import get_provider_status
# Get provider status (takes no arguments)
result = get_provider_status()
# Should return a dict with provider information
assert isinstance(result, dict)
# Should have at least authentication provider
assert "Authentication" in result or len(result) >= 0
tests/test_monitor_stalled_steps.py
+157
View File
@@ -0,0 +1,157 @@
"""
Tests for app/tasks/monitor_stalled_steps.py
This module tests the periodic task that monitors and recovers stalled processing steps.
"""
import pytest
from unittest.mock import MagicMock, patch, call
from datetime import datetime
@pytest.mark.unit
class TestMonitorStalledSteps:
"""Test monitor_stalled_steps task."""
@patch('app.tasks.monitor_stalled_steps.mark_stalled_steps_as_failed')
@patch('app.tasks.monitor_stalled_steps.SessionLocal')
def test_monitor_stalled_steps_no_stalled(self, mock_session_local, mock_mark_stalled):
"""Test monitor_stalled_steps when no stalled steps found."""
from app.tasks.monitor_stalled_steps import monitor_stalled_steps
# Mock database session
mock_db = MagicMock()
mock_session_local.return_value.__enter__.return_value = mock_db
# No stalled steps
mock_mark_stalled.return_value = 0
# Run task
result = monitor_stalled_steps()
# Verify result
assert result == {"recovered": 0}
mock_mark_stalled.assert_called_once_with(mock_db)
@patch('app.tasks.monitor_stalled_steps.mark_stalled_steps_as_failed')
@patch('app.tasks.monitor_stalled_steps.SessionLocal')
def test_monitor_stalled_steps_with_stalled(self, mock_session_local, mock_mark_stalled):
"""Test monitor_stalled_steps when stalled steps are found."""
from app.tasks.monitor_stalled_steps import monitor_stalled_steps
# Mock database session
mock_db = MagicMock()
mock_session_local.return_value.__enter__.return_value = mock_db
# Found 3 stalled steps
mock_mark_stalled.return_value = 3
# Run task
result = monitor_stalled_steps()
# Verify result
assert result == {"recovered": 3}
mock_mark_stalled.assert_called_once_with(mock_db)
@patch('app.tasks.monitor_stalled_steps.mark_stalled_steps_as_failed')
@patch('app.tasks.monitor_stalled_steps.SessionLocal')
@patch('app.tasks.monitor_stalled_steps.logger')
def test_monitor_stalled_steps_logs_recovery(self, mock_logger, mock_session_local, mock_mark_stalled):
"""Test that monitor_stalled_steps logs recovery actions."""
from app.tasks.monitor_stalled_steps import monitor_stalled_steps
# Mock database session
mock_db = MagicMock()
mock_session_local.return_value.__enter__.return_value = mock_db
# Found 2 stalled steps
mock_mark_stalled.return_value = 2
# Run task
result = monitor_stalled_steps()
# Verify logging
mock_logger.warning.assert_called_once()
log_message = mock_logger.warning.call_args[0][0]
assert "Recovered 2 stalled step(s)" in log_message
@patch('app.tasks.monitor_stalled_steps.mark_stalled_steps_as_failed')
@patch('app.tasks.monitor_stalled_steps.SessionLocal')
@patch('app.tasks.monitor_stalled_steps.logger')
def test_monitor_stalled_steps_logs_debug_when_none(self, mock_logger, mock_session_local, mock_mark_stalled):
"""Test that monitor_stalled_steps logs debug message when no stalled steps."""
from app.tasks.monitor_stalled_steps import monitor_stalled_steps
# Mock database session
mock_db = MagicMock()
mock_session_local.return_value.__enter__.return_value = mock_db
# No stalled steps
mock_mark_stalled.return_value = 0
# Run task
result = monitor_stalled_steps()
# Verify debug logging
mock_logger.debug.assert_called_once()
log_message = mock_logger.debug.call_args[0][0]
assert "No stalled steps found" in log_message
@patch('app.tasks.monitor_stalled_steps.mark_stalled_steps_as_failed')
@patch('app.tasks.monitor_stalled_steps.SessionLocal')
@patch('app.tasks.monitor_stalled_steps.logger')
def test_monitor_stalled_steps_handles_exceptions(self, mock_logger, mock_session_local, mock_mark_stalled):
"""Test that monitor_stalled_steps handles exceptions gracefully."""
from app.tasks.monitor_stalled_steps import monitor_stalled_steps
# Mock database session
mock_db = MagicMock()
mock_session_local.return_value.__enter__.return_value = mock_db
# Simulate an exception
mock_mark_stalled.side_effect = Exception("Database error")
# Run task
result = monitor_stalled_steps()
# Verify error handling
assert result == {"error": "Database error", "recovered": 0}
mock_logger.error.assert_called_once()
@patch('app.tasks.monitor_stalled_steps.mark_stalled_steps_as_failed')
@patch('app.tasks.monitor_stalled_steps.SessionLocal')
def test_monitor_stalled_steps_uses_context_manager(self, mock_session_local, mock_mark_stalled):
"""Test that monitor_stalled_steps uses context manager for database session."""
from app.tasks.monitor_stalled_steps import monitor_stalled_steps
# Mock database session
mock_db = MagicMock()
mock_context = MagicMock()
mock_context.__enter__ = MagicMock(return_value=mock_db)
mock_context.__exit__ = MagicMock(return_value=False)
mock_session_local.return_value = mock_context
mock_mark_stalled.return_value = 0
# Run task
result = monitor_stalled_steps()
# Verify context manager was used
mock_context.__enter__.assert_called_once()
mock_context.__exit__.assert_called_once()
def test_monitor_stalled_steps_is_celery_task(self):
"""Test that monitor_stalled_steps is registered as a Celery task."""
from app.tasks.monitor_stalled_steps import monitor_stalled_steps
# Should have task attributes
assert hasattr(monitor_stalled_steps, 'apply_async')
assert hasattr(monitor_stalled_steps, 'delay')
assert callable(monitor_stalled_steps)
def test_monitor_stalled_steps_task_name(self):
"""Test that monitor_stalled_steps has correct task name."""
from app.tasks.monitor_stalled_steps import monitor_stalled_steps
# Check task name
assert monitor_stalled_steps.name == "app.tasks.monitor_stalled_steps.monitor_stalled_steps"
tests/test_rate_limit_decorators.py
+161
View File
@@ -0,0 +1,161 @@
"""
Tests for app/middleware/rate_limit_decorators.py
This module tests the rate limiting decorators for API endpoints.
"""
import pytest
from unittest.mock import MagicMock, patch, Mock
from fastapi import Request
@pytest.mark.unit
class TestRateLimitDecorators:
"""Test rate limit decorator functions."""
def test_get_limiter_initialization(self):
"""Test that get_limiter initializes limiter from app state."""
from app.middleware import rate_limit_decorators
# Reset the global limiter
rate_limit_decorators._limiter = None
# Try to get limiter - will import app and get limiter from state
# This test just verifies the function can be called
try:
# This may fail if app not fully initialized, which is okay for unit test
limiter = rate_limit_decorators.get_limiter()
# If it succeeds, limiter should not be None
assert limiter is not None or rate_limit_decorators._limiter is None
except Exception:
# If it fails, that's okay - we're testing the logic path exists
pass
def test_get_limiter_caching(self):
"""Test that get_limiter caches the limiter instance."""
from app.middleware import rate_limit_decorators
# Set up mock limiter directly
mock_limiter = MagicMock()
rate_limit_decorators._limiter = mock_limiter
# Get limiter multiple times
limiter1 = rate_limit_decorators.get_limiter()
limiter2 = rate_limit_decorators.get_limiter()
# Should return same instance
assert limiter1 is limiter2
assert limiter1 is mock_limiter
@patch('app.middleware.rate_limit_decorators.get_limiter')
def test_limit_decorator(self, mock_get_limiter):
"""Test the limit decorator applies rate limit."""
from app.middleware.rate_limit_decorators import limit
# Mock limiter
mock_limiter = MagicMock()
mock_limiter.limit = MagicMock(return_value=lambda f: f)
mock_get_limiter.return_value = mock_limiter
# Create a test function
@limit("10/minute")
async def test_endpoint():
return {"message": "success"}
# Verify limiter.limit was called with correct rate
mock_limiter.limit.assert_called_once_with("10/minute")
@patch('app.middleware.rate_limit_decorators.get_limiter')
def test_limit_decorator_with_different_rates(self, mock_get_limiter):
"""Test limit decorator with various rate limit strings."""
from app.middleware.rate_limit_decorators import limit
# Mock limiter
mock_limiter = MagicMock()
mock_limiter.limit = MagicMock(return_value=lambda f: f)
mock_get_limiter.return_value = mock_limiter
# Test different rate limits
rates = ["5/second", "100/hour", "1000/day"]
for rate in rates:
mock_limiter.limit.reset_mock()
@limit(rate)
async def test_endpoint():
return {"message": "success"}
mock_limiter.limit.assert_called_once_with(rate)
@patch('app.middleware.rate_limit_decorators.get_limiter')
def test_exempt_decorator(self, mock_get_limiter):
"""Test the exempt decorator exempts endpoint from rate limiting."""
from app.middleware.rate_limit_decorators import exempt
# Mock limiter
mock_limiter = MagicMock()
mock_limiter.exempt = MagicMock(return_value=lambda f: f)
mock_get_limiter.return_value = mock_limiter
# Create a test function
@exempt()
async def test_endpoint():
return {"message": "success"}
# Verify limiter.exempt was called
mock_limiter.exempt.assert_called_once()
@patch('app.middleware.rate_limit_decorators.get_limiter')
def test_limit_decorator_preserves_function(self, mock_get_limiter):
"""Test that limit decorator preserves the original function."""
from app.middleware.rate_limit_decorators import limit
# Mock limiter to return the function unchanged
mock_limiter = MagicMock()
mock_limiter.limit = MagicMock(return_value=lambda f: f)
mock_get_limiter.return_value = mock_limiter
# Original function
async def original_function():
return "original"
# Decorate it
@limit("10/minute")
async def decorated_function():
return "original"
# Function should still work
import asyncio
result = asyncio.run(decorated_function())
assert result == "original"
@patch('app.middleware.rate_limit_decorators.get_limiter')
def test_exempt_decorator_preserves_function(self, mock_get_limiter):
"""Test that exempt decorator preserves the original function."""
from app.middleware.rate_limit_decorators import exempt
# Mock limiter to return a simple passthrough decorator
mock_limiter = MagicMock()
mock_limiter.exempt.return_value = lambda f: f
mock_get_limiter.return_value = mock_limiter
# Decorate function
@exempt()
async def decorated_function():
return "exempted"
# Function should still work
import asyncio
result = asyncio.run(decorated_function())
assert result == "exempted"
def test_module_imports(self):
"""Test that the module can be imported without errors."""
from app.middleware import rate_limit_decorators
assert hasattr(rate_limit_decorators, 'get_limiter')
assert hasattr(rate_limit_decorators, 'limit')
assert hasattr(rate_limit_decorators, 'exempt')
assert callable(rate_limit_decorators.get_limiter)
assert callable(rate_limit_decorators.limit)
assert callable(rate_limit_decorators.exempt)
tests/test_step_timeout.py
+247
View File
@@ -0,0 +1,247 @@
"""
Tests for app/utils/step_timeout.py
This module tests step timeout detection and handling logic.
"""
import pytest
from unittest.mock import MagicMock, patch, call
from datetime import datetime, timedelta
@pytest.mark.unit
class TestStepTimeout:
"""Test step timeout utilities."""
@patch('app.utils.step_timeout.settings')
def test_get_step_timeout_default(self, mock_settings):
"""Test get_step_timeout returns default value."""
from app.utils.step_timeout import get_step_timeout, DEFAULT_STEP_TIMEOUT
# No custom timeout in settings
del mock_settings.step_timeout
timeout = get_step_timeout()
assert timeout == DEFAULT_STEP_TIMEOUT
assert timeout == 600
@patch('app.utils.step_timeout.settings')
def test_get_step_timeout_custom(self, mock_settings):
"""Test get_step_timeout returns custom value from settings."""
from app.utils.step_timeout import get_step_timeout
# Custom timeout in settings
mock_settings.step_timeout = 300
timeout = get_step_timeout()
assert timeout == 300
@patch('app.utils.step_timeout.logger')
def test_mark_stalled_steps_as_failed_no_steps(self, mock_logger):
"""Test mark_stalled_steps_as_failed when no stalled steps exist."""
from app.utils.step_timeout import mark_stalled_steps_as_failed
from app.models import FileProcessingStep
# Mock database session with proper query chain
mock_db = MagicMock()
# Set up the query chain to return empty list
mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.all.return_value = []
# Run function
count = mark_stalled_steps_as_failed(mock_db)
# No steps should be marked
assert count == 0
@patch('app.utils.step_timeout.logger')
def test_mark_stalled_steps_as_failed_with_stalled_steps(self, mock_logger):
"""Test mark_stalled_steps_as_failed marks stalled steps."""
from app.utils.step_timeout import mark_stalled_steps_as_failed
from app.models import FileProcessingStep
# Create mock stalled steps
step1 = MagicMock(spec=FileProcessingStep)
step1.file_id = 1
step1.step_name = "ocr"
step1.status = "in_progress"
step1.started_at = datetime.utcnow() - timedelta(seconds=700)
step2 = MagicMock(spec=FileProcessingStep)
step2.file_id = 2
step2.step_name = "metadata"
step2.status = "in_progress"
step2.started_at = datetime.utcnow() - timedelta(seconds=800)
# Mock database session
mock_db = MagicMock()
# Set up the query chain to return stalled steps
mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.all.return_value = [step1, step2]
# Run function
count = mark_stalled_steps_as_failed(mock_db)
# Both steps should be marked as failed
assert count == 2
assert step1.status == "failure"
assert step2.status == "failure"
assert step1.completed_at is not None
assert step2.completed_at is not None
assert "timeout" in step1.error_message.lower()
assert "timeout" in step2.error_message.lower()
mock_db.commit.assert_called_once()
@patch('app.utils.step_timeout.logger')
def test_mark_stalled_steps_as_failed_custom_timeout(self, mock_logger):
"""Test mark_stalled_steps_as_failed with custom timeout."""
from app.utils.step_timeout import mark_stalled_steps_as_failed
from app.models import FileProcessingStep
# Create mock step that's stalled with custom timeout
step = MagicMock(spec=FileProcessingStep)
step.file_id = 1
step.step_name = "ocr"
step.status = "in_progress"
step.started_at = datetime.utcnow() - timedelta(seconds=200) # 200 seconds ago
# Mock database session
mock_db = MagicMock()
# Set up the query chain to return stalled step
mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.all.return_value = [step]
# Run function with 150 second timeout
count = mark_stalled_steps_as_failed(mock_db, timeout_seconds=150)
# Step should be marked as failed
assert count == 1
assert step.status == "failure"
assert "150 seconds" in step.error_message
@patch('app.utils.step_timeout.logger')
def test_mark_stalled_steps_as_failed_for_specific_file(self, mock_logger):
"""Test mark_stalled_steps_as_failed for specific file."""
from app.utils.step_timeout import mark_stalled_steps_as_failed
from app.models import FileProcessingStep
# Create mock stalled step
step = MagicMock(spec=FileProcessingStep)
step.file_id = 42
step.step_name = "ocr"
step.status = "in_progress"
step.started_at = datetime.utcnow() - timedelta(seconds=700)
# Mock database session with file filter
mock_db = MagicMock()
# Set up the query chain with file filter
mock_query_chain = mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value
mock_query_chain.filter.return_value.all.return_value = [step]
# Run function for specific file
count = mark_stalled_steps_as_failed(mock_db, file_id=42)
# Step should be marked as failed
assert count == 1
assert step.status == "failure"
@patch('app.utils.step_timeout.logger')
def test_mark_stalled_steps_as_failed_error_message_format(self, mock_logger):
"""Test that error message includes all necessary details."""
from app.utils.step_timeout import mark_stalled_steps_as_failed
from app.models import FileProcessingStep
# Create mock stalled step
started_time = datetime.utcnow() - timedelta(seconds=700)
step = MagicMock(spec=FileProcessingStep)
step.file_id = 1
step.step_name = "ocr"
step.status = "in_progress"
step.started_at = started_time
# Mock database session
mock_db = MagicMock()
# Set up the query chain to return stalled step
mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.all.return_value = [step]
# Run function
count = mark_stalled_steps_as_failed(mock_db, timeout_seconds=600)
# Check error message content
assert count == 1
error_msg = step.error_message
assert "600 seconds" in error_msg
assert "timeout" in error_msg.lower()
assert str(started_time) in error_msg
@patch('app.utils.step_timeout.logger')
def test_mark_stalled_steps_as_failed_logging(self, mock_logger):
"""Test that mark_stalled_steps_as_failed logs warnings and errors."""
from app.utils.step_timeout import mark_stalled_steps_as_failed
from app.models import FileProcessingStep
# Create mock stalled step
step = MagicMock(spec=FileProcessingStep)
step.file_id = 1
step.step_name = "ocr"
step.status = "in_progress"
step.started_at = datetime.utcnow() - timedelta(seconds=700)
# Mock database session
mock_db = MagicMock()
# Set up the query chain to return stalled step
mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.all.return_value = [step]
# Run function
count = mark_stalled_steps_as_failed(mock_db)
# Verify logging
assert count == 1
mock_logger.warning.assert_called_once()
mock_logger.error.assert_called_once()
@patch('app.utils.step_timeout.logger')
def test_check_and_recover_stalled_file_found(self, mock_logger):
"""Test check_and_recover_stalled_file when stalled steps found."""
from app.utils.step_timeout import check_and_recover_stalled_file
from app.models import FileProcessingStep
# Create mock stalled step
step = MagicMock(spec=FileProcessingStep)
step.file_id = 42
step.step_name = "ocr"
step.status = "in_progress"
step.started_at = datetime.utcnow() - timedelta(seconds=700)
# Mock database session
mock_db = MagicMock()
# Set up the query chain with file filter
mock_query_chain = mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value
mock_query_chain.filter.return_value.all.return_value = [step]
# Run function
result = check_and_recover_stalled_file(mock_db, 42)
# Should return True when stalled steps found
assert result is True
@patch('app.utils.step_timeout.logger')
def test_check_and_recover_stalled_file_not_found(self, mock_logger):
"""Test check_and_recover_stalled_file when no stalled steps."""
from app.utils.step_timeout import check_and_recover_stalled_file
# Mock database session with no stalled steps
mock_db = MagicMock()
# Set up the query chain with file filter
mock_query_chain = mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value
mock_query_chain.filter.return_value.all.return_value = []
# Run function
result = check_and_recover_stalled_file(mock_db, 42)
# Should return False when no stalled steps
assert result is False
def test_default_step_timeout_constant(self):
"""Test that DEFAULT_STEP_TIMEOUT is defined correctly."""
from app.utils.step_timeout import DEFAULT_STEP_TIMEOUT
assert DEFAULT_STEP_TIMEOUT == 600
assert isinstance(DEFAULT_STEP_TIMEOUT, int)