mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-19 01:25:06 +00:00
7d6bef5507
- Improve DVR recording stability by ensuring FFmpeg restarts on mid-stream outages, maintaining segment continuity and proper numbering. - Update HLS→MKV concatenation to handle timestamp splices and corrupt segments, utilizing error-tolerant FFmpeg flags. - Introduce helper functions for managing HLS segment counts and retry logic, ensuring robust handling of recording interruptions. - Update tests to validate new retry logic and error handling in recording processes. (Closes #1170)
280 lines
11 KiB
Python
280 lines
11 KiB
Python
"""Tests for DVR retry logic.
|
|
|
|
Covers:
|
|
- _db_retry(): exponential backoff, max retries, connection reset
|
|
- Final metadata save retry in run_recording post-processing
|
|
- Initial TS proxy connection retry (per-base retry on retriable errors)
|
|
- recover_recordings_on_startup DB retry wrappers
|
|
"""
|
|
from datetime import timedelta
|
|
from unittest.mock import MagicMock, patch, call
|
|
|
|
from django.db import OperationalError
|
|
from django.test import TestCase
|
|
from django.utils import timezone
|
|
|
|
from apps.channels.models import Channel, Recording
|
|
from apps.channels.tasks import _db_retry
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _db_retry unit tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class DbRetryTests(TestCase):
|
|
"""Tests for the _db_retry() exponential backoff helper."""
|
|
|
|
@patch("apps.channels.tasks.time.sleep")
|
|
@patch("apps.channels.tasks.close_old_connections")
|
|
def test_succeeds_on_first_attempt(self, _close, _sleep):
|
|
"""No retry needed when fn succeeds immediately."""
|
|
result = _db_retry(lambda: "ok", max_retries=3)
|
|
self.assertEqual(result, "ok")
|
|
_sleep.assert_not_called()
|
|
|
|
@patch("apps.channels.tasks.time.sleep")
|
|
@patch("apps.channels.tasks.close_old_connections")
|
|
def test_retries_on_operational_error_then_succeeds(self, mock_close, mock_sleep):
|
|
"""Retry succeeds on second attempt after OperationalError."""
|
|
call_count = {"n": 0}
|
|
|
|
def flaky():
|
|
call_count["n"] += 1
|
|
if call_count["n"] == 1:
|
|
raise OperationalError("connection reset")
|
|
return "recovered"
|
|
|
|
result = _db_retry(flaky, max_retries=3, base_interval=1)
|
|
self.assertEqual(result, "recovered")
|
|
self.assertEqual(call_count["n"], 2)
|
|
|
|
@patch("apps.channels.tasks.time.sleep")
|
|
@patch("apps.channels.tasks.close_old_connections")
|
|
def test_raises_after_max_retries_exhausted(self, mock_close, mock_sleep):
|
|
"""Raises OperationalError after all retries fail."""
|
|
def always_fail():
|
|
raise OperationalError("db gone")
|
|
|
|
with self.assertRaises(OperationalError):
|
|
_db_retry(always_fail, max_retries=3, base_interval=1)
|
|
|
|
@patch("apps.channels.tasks.time.sleep")
|
|
@patch("apps.channels.tasks.close_old_connections")
|
|
def test_exponential_backoff_timing(self, mock_close, mock_sleep):
|
|
"""Sleep durations follow exponential backoff: 1s, 2s, 4s."""
|
|
call_count = {"n": 0}
|
|
|
|
def fail_twice():
|
|
call_count["n"] += 1
|
|
if call_count["n"] <= 2:
|
|
raise OperationalError("retry me")
|
|
return "done"
|
|
|
|
_db_retry(fail_twice, max_retries=3, base_interval=1)
|
|
mock_sleep.assert_has_calls([call(1), call(2)])
|
|
|
|
@patch("apps.channels.tasks.time.sleep")
|
|
@patch("apps.channels.tasks.close_old_connections")
|
|
def test_close_old_connections_called_between_retries(self, mock_close, mock_sleep):
|
|
"""Stale DB connections are reset before each retry attempt."""
|
|
call_count = {"n": 0}
|
|
|
|
def fail_once():
|
|
call_count["n"] += 1
|
|
if call_count["n"] == 1:
|
|
raise OperationalError("stale conn")
|
|
return "ok"
|
|
|
|
_db_retry(fail_once, max_retries=3)
|
|
mock_close.assert_called_once()
|
|
|
|
@patch("apps.channels.tasks.time.sleep")
|
|
@patch("apps.channels.tasks.close_old_connections")
|
|
def test_non_operational_error_not_retried(self, mock_close, mock_sleep):
|
|
"""Non-OperationalError exceptions propagate immediately."""
|
|
def raise_value_error():
|
|
raise ValueError("not a DB error")
|
|
|
|
with self.assertRaises(ValueError):
|
|
_db_retry(raise_value_error, max_retries=3)
|
|
mock_sleep.assert_not_called()
|
|
|
|
@patch("apps.channels.tasks.time.sleep")
|
|
@patch("apps.channels.tasks.close_old_connections")
|
|
def test_returns_fn_return_value(self, mock_close, mock_sleep):
|
|
"""Return value of fn() is passed through."""
|
|
result = _db_retry(lambda: {"key": "value"}, max_retries=3)
|
|
self.assertEqual(result, {"key": "value"})
|
|
|
|
@patch("apps.channels.tasks.time.sleep")
|
|
@patch("apps.channels.tasks.close_old_connections")
|
|
def test_single_retry_allowed(self, mock_close, mock_sleep):
|
|
"""max_retries=1 means no retry — fail immediately."""
|
|
with self.assertRaises(OperationalError):
|
|
_db_retry(
|
|
lambda: (_ for _ in ()).throw(OperationalError("fail")),
|
|
max_retries=1,
|
|
)
|
|
mock_sleep.assert_not_called()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Final metadata save retry integration tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class FinalMetadataSaveRetryTests(TestCase):
|
|
"""The final recording metadata save must retry on transient DB errors."""
|
|
|
|
def setUp(self):
|
|
self.channel = Channel.objects.create(
|
|
channel_number=95, name="Retry Test Channel"
|
|
)
|
|
|
|
@patch("core.utils.send_websocket_update", side_effect=lambda *a, **kw: None)
|
|
def test_metadata_save_uses_db_retry(self, _ws):
|
|
"""Verify recording metadata is saved via _db_retry (retries on OperationalError)."""
|
|
now = timezone.now()
|
|
rec = Recording.objects.create(
|
|
channel=self.channel,
|
|
start_time=now - timedelta(hours=1),
|
|
end_time=now + timedelta(hours=1),
|
|
custom_properties={"status": "recording"},
|
|
)
|
|
# Directly call _db_retry to save metadata as run_recording does
|
|
cp = rec.custom_properties.copy()
|
|
cp["status"] = "completed"
|
|
cp["ended_at"] = str(now)
|
|
cp["bytes_written"] = 1024
|
|
|
|
def _save():
|
|
rec.custom_properties = cp
|
|
rec.save(update_fields=["custom_properties"])
|
|
|
|
_db_retry(_save, max_retries=3, base_interval=1, label="test save")
|
|
rec.refresh_from_db()
|
|
self.assertEqual(rec.custom_properties["status"], "completed")
|
|
self.assertEqual(rec.custom_properties["bytes_written"], 1024)
|
|
|
|
@patch("core.utils.send_websocket_update", side_effect=lambda *a, **kw: None)
|
|
def test_metadata_survives_transient_save_failure(self, _ws):
|
|
"""Simulate OperationalError on first save, success on retry."""
|
|
now = timezone.now()
|
|
rec = Recording.objects.create(
|
|
channel=self.channel,
|
|
start_time=now - timedelta(hours=1),
|
|
end_time=now + timedelta(hours=1),
|
|
custom_properties={"status": "recording"},
|
|
)
|
|
cp = {"status": "completed", "bytes_written": 2048}
|
|
call_count = {"n": 0}
|
|
_real_save = rec.save
|
|
|
|
def patched_save(**kwargs):
|
|
call_count["n"] += 1
|
|
if call_count["n"] == 1:
|
|
raise OperationalError("connection reset by peer")
|
|
return _real_save(**kwargs)
|
|
|
|
with patch.object(rec, "save", side_effect=patched_save):
|
|
with patch("apps.channels.tasks.time.sleep"):
|
|
with patch("apps.channels.tasks.close_old_connections"):
|
|
def _save():
|
|
rec.custom_properties = cp
|
|
rec.save(update_fields=["custom_properties"])
|
|
_db_retry(_save, max_retries=3, base_interval=1, label="test")
|
|
|
|
rec.refresh_from_db()
|
|
self.assertEqual(rec.custom_properties["status"], "completed")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Initial connection retry tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class InitialConnectionRetryTests(TestCase):
|
|
"""Verify that the DVR task's reconnection logic retries the same
|
|
base URL before falling back to the next candidate."""
|
|
|
|
def test_reconnect_max_constant_exists_in_run_recording(self):
|
|
"""run_recording must use a time-bounded FFmpeg outage window to prevent
|
|
infinite restarts when the source stream is permanently down."""
|
|
import inspect
|
|
from apps.channels.tasks import run_recording, _dvr_ffmpeg_retry_window_seconds
|
|
source = inspect.getsource(run_recording)
|
|
|
|
self.assertGreater(_dvr_ffmpeg_retry_window_seconds(), 0)
|
|
self.assertIn("reconnect", source.lower(),
|
|
"run_recording must contain input reconnection flags")
|
|
self.assertIn("_ffmpeg_outage_started", source)
|
|
self.assertIn("_ffmpeg_retry_window", source)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# recover_recordings_on_startup retry tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class RecoveryRetryTests(TestCase):
|
|
"""DB operations in recover_recordings_on_startup must use _db_retry."""
|
|
|
|
def setUp(self):
|
|
self.channel = Channel.objects.create(
|
|
channel_number=97, name="Recovery Retry Channel"
|
|
)
|
|
|
|
@patch("apps.channels.tasks.run_recording.apply_async")
|
|
@patch("core.utils.send_websocket_update", side_effect=lambda *a, **kw: None)
|
|
def test_recovery_save_retries_on_operational_error(self, _ws, mock_async):
|
|
"""Recovery status update uses _db_retry — survives one OperationalError."""
|
|
now = timezone.now()
|
|
rec = Recording.objects.create(
|
|
channel=self.channel,
|
|
start_time=now - timedelta(minutes=30),
|
|
end_time=now + timedelta(minutes=30),
|
|
custom_properties={},
|
|
)
|
|
# Simulate what recovery does: mark interrupted, then save with retry
|
|
cp = rec.custom_properties or {}
|
|
cp["status"] = "interrupted"
|
|
cp["interrupted_reason"] = "server_restarted"
|
|
rec.custom_properties = cp
|
|
|
|
call_count = {"n": 0}
|
|
_real_save = Recording.save
|
|
|
|
def patched_save(self_rec, **kwargs):
|
|
call_count["n"] += 1
|
|
if call_count["n"] == 1:
|
|
raise OperationalError("db temporarily unavailable")
|
|
return _real_save(self_rec, **kwargs)
|
|
|
|
with patch.object(Recording, "save", patched_save):
|
|
with patch("apps.channels.tasks.time.sleep"):
|
|
with patch("apps.channels.tasks.close_old_connections"):
|
|
_db_retry(
|
|
lambda: rec.save(update_fields=["custom_properties"]),
|
|
max_retries=3,
|
|
label="test recovery",
|
|
)
|
|
|
|
rec.refresh_from_db()
|
|
self.assertEqual(rec.custom_properties.get("status"), "interrupted")
|
|
self.assertEqual(rec.custom_properties.get("interrupted_reason"), "server_restarted")
|
|
|
|
def test_db_retry_fetches_recording_list(self):
|
|
"""_db_retry correctly returns query results for recording list fetch."""
|
|
now = timezone.now()
|
|
rec = Recording.objects.create(
|
|
channel=self.channel,
|
|
start_time=now - timedelta(minutes=30),
|
|
end_time=now + timedelta(minutes=30),
|
|
custom_properties={},
|
|
)
|
|
result = _db_retry(
|
|
lambda: list(Recording.objects.filter(
|
|
start_time__lte=now, end_time__gt=now
|
|
)),
|
|
label="test query",
|
|
)
|
|
self.assertGreaterEqual(len(result), 1)
|
|
ids = [r.id for r in result]
|
|
self.assertIn(rec.id, ids)
|