|
| 1 | +#!/usr/bin/env python3 |
| 2 | +# -*- coding: utf-8 -*- |
| 3 | + |
| 4 | +# Calibre-Web Automated – fork of Calibre-Web |
| 5 | +# Copyright (C) 2024-2025 Calibre-Web Automated contributors |
| 6 | +# SPDX-License-Identifier: GPL-3.0-or-later |
| 7 | + |
| 8 | +""" |
| 9 | +Test suite for auto_send_delay_minutes setting persistence and validation. |
| 10 | +Verifies that the delay setting is correctly validated and used. |
| 11 | +""" |
| 12 | + |
| 13 | +import pytest |
| 14 | +import sys |
| 15 | +import os |
| 16 | + |
| 17 | +# Mark all tests in this file as unit tests |
| 18 | +pytestmark = pytest.mark.unit |
| 19 | + |
| 20 | +# Add parent directory to path for imports |
| 21 | +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) |
| 22 | + |
| 23 | + |
| 24 | +class TestAutoSendDelayValidation: |
| 25 | + """Test auto-send delay setting validation and usage""" |
| 26 | + |
| 27 | + def test_schema_has_default(self): |
| 28 | + """Verify schema defines auto_send_delay_minutes with default""" |
| 29 | + schema_file = os.path.join( |
| 30 | + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), |
| 31 | + 'scripts', |
| 32 | + 'cwa_schema.sql' |
| 33 | + ) |
| 34 | + |
| 35 | + with open(schema_file, 'r', encoding='utf-8') as f: |
| 36 | + content = f.read() |
| 37 | + |
| 38 | + # Verify column exists with proper default |
| 39 | + assert 'auto_send_delay_minutes INTEGER DEFAULT 5 NOT NULL' in content |
| 40 | + |
| 41 | + def test_template_has_validation(self): |
| 42 | + """Verify settings template has min/max validation""" |
| 43 | + template_file = os.path.join( |
| 44 | + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), |
| 45 | + 'cps', |
| 46 | + 'templates', |
| 47 | + 'cwa_settings.html' |
| 48 | + ) |
| 49 | + |
| 50 | + with open(template_file, 'r', encoding='utf-8') as f: |
| 51 | + content = f.read() |
| 52 | + |
| 53 | + # Verify input has type=number with min/max |
| 54 | + assert 'name="auto_send_delay_minutes"' in content |
| 55 | + assert 'type="number"' in content |
| 56 | + assert 'min="1"' in content |
| 57 | + assert 'max="60"' in content |
| 58 | + |
| 59 | + def test_cwa_functions_validates_range(self): |
| 60 | + """Verify cwa_functions.py validates 1-60 range""" |
| 61 | + cwa_functions_file = os.path.join( |
| 62 | + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), |
| 63 | + 'cps', |
| 64 | + 'cwa_functions.py' |
| 65 | + ) |
| 66 | + |
| 67 | + with open(cwa_functions_file, 'r', encoding='utf-8') as f: |
| 68 | + content = f.read() |
| 69 | + |
| 70 | + # Verify validation logic clamps to 1-60 |
| 71 | + assert "'auto_send_delay_minutes'" in content |
| 72 | + assert "max(1, min(60, int_value))" in content |
| 73 | + |
| 74 | + def test_ingest_uses_delay_setting(self): |
| 75 | + """Verify ingest processor uses delay from CWA settings""" |
| 76 | + ingest_file = os.path.join( |
| 77 | + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), |
| 78 | + 'scripts', |
| 79 | + 'ingest_processor.py' |
| 80 | + ) |
| 81 | + |
| 82 | + with open(ingest_file, 'r', encoding='utf-8') as f: |
| 83 | + content = f.read() |
| 84 | + |
| 85 | + # Verify ingest reads from cwa_settings with fallback |
| 86 | + assert "self.cwa_settings.get('auto_send_delay_minutes', 5)" in content |
| 87 | + |
| 88 | + def test_internal_endpoint_validates_delay(self): |
| 89 | + """Verify /cwa-internal/schedule-auto-send validates delay_minutes""" |
| 90 | + cwa_functions_file = os.path.join( |
| 91 | + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), |
| 92 | + 'cps', |
| 93 | + 'cwa_functions.py' |
| 94 | + ) |
| 95 | + |
| 96 | + with open(cwa_functions_file, 'r', encoding='utf-8') as f: |
| 97 | + content = f.read() |
| 98 | + |
| 99 | + # Verify endpoint clamps delay_minutes to 0-60 range |
| 100 | + # Search for the schedule-auto-send endpoint function |
| 101 | + assert '/cwa-internal/schedule-auto-send' in content |
| 102 | + assert 'delay_minutes = int(data.get' in content |
| 103 | + assert 'max(0, min(60, delay_minutes))' in content |
| 104 | + |
| 105 | + def test_validation_range_boundary(self): |
| 106 | + """Test validation handles boundary values correctly""" |
| 107 | + # Mock test to verify logic (actual values tested at runtime) |
| 108 | + |
| 109 | + # Test lower bound: value < 1 should become 1 |
| 110 | + test_value = -5 |
| 111 | + clamped = max(1, min(60, test_value)) |
| 112 | + assert clamped == 1 |
| 113 | + |
| 114 | + # Test within range: should stay unchanged |
| 115 | + test_value = 30 |
| 116 | + clamped = max(1, min(60, test_value)) |
| 117 | + assert clamped == 30 |
| 118 | + |
| 119 | + # Test upper bound: value > 60 should become 60 |
| 120 | + test_value = 120 |
| 121 | + clamped = max(1, min(60, test_value)) |
| 122 | + assert clamped == 60 |
| 123 | + |
| 124 | + def test_default_fallback(self): |
| 125 | + """Test that missing/invalid values fall back to 5""" |
| 126 | + # Verify default logic |
| 127 | + default_value = 5 |
| 128 | + |
| 129 | + # Test None fallback |
| 130 | + value = None |
| 131 | + result = default_value if value is None else value |
| 132 | + assert result == 5 |
| 133 | + |
| 134 | + # Test empty string fallback in conversion |
| 135 | + try: |
| 136 | + value = "" |
| 137 | + result = int(value) if value else default_value |
| 138 | + except (ValueError, TypeError): |
| 139 | + result = default_value |
| 140 | + assert result == 5 |
| 141 | + |
| 142 | + |
| 143 | +class TestDelayUsageFlow: |
| 144 | + """Test that delay is properly used in scheduling flow""" |
| 145 | + |
| 146 | + def test_schedule_endpoint_uses_delay(self): |
| 147 | + """Verify internal schedule endpoint uses delay_minutes parameter""" |
| 148 | + cwa_functions_file = os.path.join( |
| 149 | + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), |
| 150 | + 'cps', |
| 151 | + 'cwa_functions.py' |
| 152 | + ) |
| 153 | + |
| 154 | + with open(cwa_functions_file, 'r', encoding='utf-8') as f: |
| 155 | + content = f.read() |
| 156 | + |
| 157 | + # Verify delay_minutes is extracted and used in timedelta calculation |
| 158 | + assert 'timedelta(minutes=delay_minutes)' in content |
| 159 | + |
| 160 | + def test_task_receives_delay_parameter(self): |
| 161 | + """Verify TaskAutoSend receives delay_minutes parameter""" |
| 162 | + cwa_functions_file = os.path.join( |
| 163 | + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), |
| 164 | + 'cps', |
| 165 | + 'cwa_functions.py' |
| 166 | + ) |
| 167 | + |
| 168 | + with open(cwa_functions_file, 'r', encoding='utf-8') as f: |
| 169 | + content = f.read() |
| 170 | + |
| 171 | + # Verify TaskAutoSend is called with delay_minutes |
| 172 | + assert 'TaskAutoSend(task_message, book_id, user_id, delay_minutes)' in content |
| 173 | + |
| 174 | + |
| 175 | +if __name__ == '__main__': |
| 176 | + pytest.main([__file__, '-v']) |
0 commit comments