From 35ac9ea987b6a5dd9f7cf75b79ff650f5ae339d0 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 7 Jul 2026 21:08:45 +0000 Subject: [PATCH] test: Refactor collision avoidance tests to use a fixed timestamp and improve suffix collision checks. Introduced helper function for clearer logic in assertions regarding MKV file naming conventions. --- .../channels/tests/test_recording_pipeline.py | 37 ++++++++++++------- 1 file changed, 23 insertions(+), 14 deletions(-) diff --git a/apps/channels/tests/test_recording_pipeline.py b/apps/channels/tests/test_recording_pipeline.py index b1e2a45f..21254b30 100644 --- a/apps/channels/tests/test_recording_pipeline.py +++ b/apps/channels/tests/test_recording_pipeline.py @@ -8,6 +8,7 @@ Covers: 5. Recovery skip-list: "recording" status NOT in terminal skip list """ import os +import datetime as dt from datetime import timedelta from unittest.mock import MagicMock, patch @@ -17,6 +18,15 @@ from rest_framework.test import APIRequestFactory, force_authenticate from apps.channels.models import Channel, Recording +# Fixed wall time for collision tests: 10:30 avoids _2 appearing inside +# %Y%m%d_%H%M%S timestamps (e.g. hour 20 produces ..._205331 which contains "_2"). +COLLISION_TEST_START = timezone.make_aware(dt.datetime(2026, 1, 15, 10, 30, 0)) + + +def _path_has_collision_suffix(path, counter): + """True when the MKV basename ends with _.mkv (not timestamp digits).""" + return path.endswith(f"_{counter}.mkv") + # --------------------------------------------------------------------------- # Helpers @@ -71,7 +81,7 @@ class CollisionAvoidanceTests(TestCase): ch = MagicMock(name="TestCh") ch.name = "TestCh" program = {"title": "My Show"} - now = timezone.now() + now = COLLISION_TEST_START def mock_stat(path): raise OSError("No such file") @@ -80,8 +90,7 @@ class CollisionAvoidanceTests(TestCase): patch("os.makedirs"): final, ts, fname = self._call(ch, program, now, now + timedelta(hours=1)) - # Should NOT have a _2 suffix - self.assertNotIn("_2", final) + self.assertFalse(_path_has_collision_suffix(final, 2)) self.assertTrue(final.endswith(".mkv")) @patch("apps.channels.tasks.CoreSettings.get_dvr_tv_fallback_template", @@ -95,10 +104,10 @@ class CollisionAvoidanceTests(TestCase): ch = MagicMock(name="TestCh") ch.name = "TestCh" program = {"title": "My Show"} - now = timezone.now() + now = COLLISION_TEST_START def mock_stat(path): - if "_2" in path: + if _path_has_collision_suffix(path, 2): raise OSError("No such file") result = MagicMock() if path.endswith('.mkv'): @@ -113,7 +122,7 @@ class CollisionAvoidanceTests(TestCase): patch("os.makedirs"): final, hls_dir, fname = self._call(ch, program, now, now + timedelta(hours=1)) - self.assertNotIn("_2", final, "HLS path builder ignores legacy TS when MKV is empty") + self.assertFalse(_path_has_collision_suffix(final, 2), "HLS path builder ignores legacy TS when MKV is empty") @patch("apps.channels.tasks.CoreSettings.get_dvr_tv_fallback_template", return_value="TV/{show}/{start}.mkv") @@ -124,10 +133,10 @@ class CollisionAvoidanceTests(TestCase): ch = MagicMock(name="TestCh") ch.name = "TestCh" program = {"title": "My Show"} - now = timezone.now() + now = COLLISION_TEST_START def mock_stat(path): - if "_2" in path: + if _path_has_collision_suffix(path, 2): raise OSError("No such file") result = MagicMock() if path.endswith('.mkv'): @@ -140,7 +149,7 @@ class CollisionAvoidanceTests(TestCase): patch("os.makedirs"): final, ts, fname = self._call(ch, program, now, now + timedelta(hours=1)) - self.assertIn("_2", final, "Should increment counter when MKV file has data") + self.assertTrue(_path_has_collision_suffix(final, 2), "Should increment counter when MKV file has data") @patch("apps.channels.tasks.CoreSettings.get_dvr_tv_fallback_template", return_value="TV/{show}/{start}.mkv") @@ -151,7 +160,7 @@ class CollisionAvoidanceTests(TestCase): ch = MagicMock(name="TestCh") ch.name = "TestCh" program = {"title": "My Show"} - now = timezone.now() + now = COLLISION_TEST_START def mock_stat(path): result = MagicMock() @@ -162,7 +171,7 @@ class CollisionAvoidanceTests(TestCase): patch("os.makedirs"): final, ts, fname = self._call(ch, program, now, now + timedelta(hours=1)) - self.assertNotIn("_2", final, "Should NOT increment when all files are empty") + self.assertFalse(_path_has_collision_suffix(final, 2), "Should NOT increment when all files are empty") @patch("apps.channels.tasks.CoreSettings.get_dvr_tv_fallback_template", return_value="TV/{show}/{start}.mkv") @@ -173,10 +182,10 @@ class CollisionAvoidanceTests(TestCase): ch = MagicMock(name="TestCh") ch.name = "TestCh" program = {"title": "My Show"} - now = timezone.now() + now = COLLISION_TEST_START def mock_stat(path): - if "_3" in path: + if _path_has_collision_suffix(path, 3): raise OSError("No such file") result = MagicMock() if path.endswith('.mkv'): @@ -189,7 +198,7 @@ class CollisionAvoidanceTests(TestCase): patch("os.makedirs"): final, hls_dir, fname = self._call(ch, program, now, now + timedelta(hours=1)) - self.assertIn("_3", final, "Should increment to _3 when base and _2 MKVs are occupied") + self.assertTrue(_path_has_collision_suffix(final, 3), "Should increment to _3 when base and _2 MKVs are occupied") # =========================================================================