From bae6c0046ade4fe73cc02cf4080f963ac80ed6a5 Mon Sep 17 00:00:00 2001 From: sumitdipsite2005 Date: Fri, 9 Oct 2026 19:38:47 -0500 Subject: [PATCH 1/5] Issue 3: add bitrate-only upgrade guard without altering quality ranking --- recorder_source/selection.py | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/recorder_source/selection.py b/recorder_source/selection.py index ce69526..a8cbd7f 100644 --- a/recorder_source/selection.py +++ b/recorder_source/selection.py @@ -115,6 +115,34 @@ def video_quality_rank( ) + +def bitrate_only_upgrade_below_minimum( + running_source: Union[Mapping[str, object], SourceCandidate], + alternative: Union[Mapping[str, object], SourceCandidate], + *, + motion_cap_fps: float, + minimum_percent: int = 10, +) -> bool: + """Guard only upgrades where bitrate is the sole video-rank difference. + + Unknown bitrate leaves the mature selection behavior unchanged. Comparing + integer products avoids rounding at the exact percentage boundary. + """ + running = _as_candidate(running_source) + proposed = _as_candidate(alternative) + current_bitrate = int(running.video_bitrate_bps or 0) + proposed_bitrate = int(proposed.video_bitrate_bps or 0) + if current_bitrate <= 0 or proposed_bitrate <= 0: + return False + + current_rank = video_quality_rank(running, motion_cap_fps=motion_cap_fps) + proposed_rank = video_quality_rank(proposed, motion_cap_fps=motion_cap_fps) + return ( + current_rank[:-1] == proposed_rank[:-1] + and proposed_bitrate > current_bitrate + and proposed_bitrate * 100 < current_bitrate * (100 + minimum_percent) + ) + def candidate_quality_rank( candidate: Union[Mapping[str, object], SourceCandidate], policy: SelectionPolicy, From f9a6b6e3386f6b4f51f508a37d95ff8ae519c656 Mon Sep 17 00:00:00 2001 From: sumitdipsite2005 Date: Fri, 9 Oct 2026 19:39:16 -0500 Subject: [PATCH 2/5] Issue 3: require fresh IDET confirmation, preserve renewal, guard bitrate-only rollovers --- record_dynamic.py | 225 ++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 186 insertions(+), 39 deletions(-) diff --git a/record_dynamic.py b/record_dynamic.py index 2d8e1f7..c8302d0 100644 --- a/record_dynamic.py +++ b/record_dynamic.py @@ -8188,6 +8188,104 @@ def run_idet_probe() -> tuple: scan_type_cache["results"][cache_key] = scan_type return scan_type + +def _confirm_nm3u8dl_quality_upgrade_idet( + candidate: dict, + stop_requested: Optional[Callable[[], bool]] = None, +) -> str: + """Confirm an IDET upgrade on the exact selected media, never its cache. + + Returns not_required, confirmed, failed, inconclusive, or cancelled. + This is a quality-upgrade guard, not an access-block classifier. + """ + if str(candidate.get("video_scan_type_source") or "").strip().casefold() != "idet": + return "not_required" + + expected = _normalize_nm3u8dl_video_scan_type( + candidate.get("video_scan_type") + ) + if not expected: + return "inconclusive" + if stop_requested is not None and stop_requested(): + return "cancelled" + + headers = candidate.get("effective_headers") + if not isinstance(headers, dict): + try: + headers = get_nm3u8dl_effective_headers( + candidate.get("headers") or {}, emit_logs=False + ) + except Exception: + return "inconclusive" + + stream_type = str(candidate.get("stream_type") or "").upper() + temp_path = "" + if stream_type == "HLS": + probe_url = str( + candidate.get("manifest_variant_url") + or candidate.get("stream_url") + or "" + ).strip() + sample_headers = headers + sample_range = "" + effective_url = str( + candidate.get("selected_media_final_url") + or candidate.get("manifest_final_url") + or "" + ).strip() + elif stream_type == "DASH": + try: + probe_url, sample_headers, sample_range, temp_path = ( + _prepare_nm3u8dl_selected_dash_media_sample(candidate, headers) + ) + except Exception: + return "inconclusive" + # Mirror the original DASH IDET probe's selected-representation route. + effective_url = str( + candidate.get("manifest_final_url") + or candidate.get("stream_url") + or "" + ).strip() + else: + return "inconclusive" + + try: + if stop_requested is not None and stop_requested(): + return "cancelled" + if not probe_url: + return "inconclusive" + keys = _nm3u8dl_known_media_probe_keys(candidate, candidate, headers) + if not keys: + return "inconclusive" + + for key in keys: + if stop_requested is not None and stop_requested(): + return "cancelled" + observed = _detect_nm3u8dl_stream_scan_type_with_idet( + probe_url, + sample_headers, + scan_type_cache=None, + timeout_route=_format_candidate_timeout_route( + [str(candidate.get("playlist_url") or "")], + candidate.get("manifest_final_url") or probe_url, + ), + effective_stream_url=effective_url, + decryption_key=key, + ) + if stop_requested is not None and stop_requested(): + return "cancelled" + if observed: + return "confirmed" if observed == expected else "failed" + return "inconclusive" + except Exception: + return "inconclusive" + finally: + if temp_path: + try: + os.remove(temp_path) + except Exception: + pass + def _ffprobe_nm3u8dl_stream_quality( stream_url: str, headers: dict, @@ -15299,26 +15397,6 @@ def record_runtime_scan_history( ) ) - targeted_upgrade_candidate = None - - if targeted_access_scan and quality_upgrade_enabled: - targeted_upgrade_candidate = ( - get_nm3u8dl_quality_upgrade_candidate( - running_source, - candidate_pool, - quality_upgrade_target_fps, - quality_min_remaining_min, - allow_unknown_expiry=allow_unknown_expiry, - now_ts=check_time, - ) - ) - - if targeted_upgrade_candidate is not None: - log_quality_upgrade_found( - targeted_upgrade_candidate, - check_time, - ) - if targeted_access_scan: access_history_changed = process_nm3u8dl_targeted_access_verification( state, @@ -15334,7 +15412,7 @@ def record_runtime_scan_history( ), running_source=running_source, access_check_interval_sec=access_check_interval_sec, - selected_candidate=targeted_upgrade_candidate, + selected_candidate=None, ) else: confirmed_blocked_urls = ( @@ -15457,21 +15535,79 @@ def record_runtime_scan_history( ) if quality_evaluation_due: - upgrade_candidate = targeted_upgrade_candidate - - if upgrade_candidate is None: - upgrade_candidate = ( - get_nm3u8dl_quality_upgrade_candidate( - running_source, - candidate_pool, - quality_upgrade_target_fps, - quality_min_remaining_min, - allow_unknown_expiry=allow_unknown_expiry, - now_ts=check_time, + # Reuse the mature selector on the current scan pool. Rejected + # quality candidates are removed only from this upgrade decision. + remaining_upgrade_candidates = list(candidate_pool) + rejected_upgrade_count = 0 + upgrade_candidate = None + + def quality_upgrade_cancelled(): + return ( + stop_event.is_set() + or state.stop_flag + or state.nm3u8dl_running_source is not running_source + ) + + while remaining_upgrade_candidates: + if quality_upgrade_cancelled(): + return + + upgrade_selection = get_nm3u8dl_quality_upgrade_selection_decision( + running_source, + remaining_upgrade_candidates, + quality_upgrade_target_fps, + quality_min_remaining_min, + allow_unknown_expiry=allow_unknown_expiry, + now_ts=time.time(), + ) + if upgrade_selection.selected_index is None: + break + + candidate_index = upgrade_selection.selected_index + proposed_upgrade = remaining_upgrade_candidates[candidate_index] + + if source_selection.bitrate_only_upgrade_below_minimum( + running_source, + proposed_upgrade, + motion_cap_fps=NM3U8DL_QUALITY_RANKING_MOTION_CAP_FPS, + minimum_percent=10, + ): + log( + "Quality upgrade bitrate safeguard → REJECTED; " + "bitrate is the only quality difference and the increase " + "is less than 10%; checking other candidates." ) + rejected_upgrade_count += 1 + del remaining_upgrade_candidates[candidate_index] + continue + + idet_confirmation = _confirm_nm3u8dl_quality_upgrade_idet( + proposed_upgrade, + stop_requested=quality_upgrade_cancelled, ) + if idet_confirmation == "cancelled" or quality_upgrade_cancelled(): + return + + if idet_confirmation in ("failed", "inconclusive"): + log( + f"Quality upgrade IDET confirmation → " + f"{idet_confirmation.upper()}; candidate excluded " + "for this scan; checking other candidates.", + level="WARN", + ) + rejected_upgrade_count += 1 + del remaining_upgrade_candidates[candidate_index] + continue + + if idet_confirmation == "confirmed": + log("Quality upgrade IDET confirmation → CONFIRMED.") + + upgrade_candidate = dict(proposed_upgrade) + break if upgrade_candidate is not None: + if quality_upgrade_cancelled(): + return for metadata_key in ( "candidate_count", "playlist_source_count", @@ -15480,16 +15616,19 @@ def record_runtime_scan_history( "source_errors", "source_results", ): - upgrade_candidate[metadata_key] = ( - resolution.get(metadata_key) - ) + upgrade_candidate[metadata_key] = resolution.get(metadata_key) - if targeted_upgrade_candidate is None: - log_quality_upgrade_found( - upgrade_candidate, - check_time, + if rejected_upgrade_count: + log( + "Quality upgrade decision → alternative candidate " + f"selected after rejecting {rejected_upgrade_count} " + "candidate(s) in this scan." ) + # This is the first point an upgrade can be reported approved. + log_quality_upgrade_found(upgrade_candidate, check_time) + if quality_upgrade_cancelled(): + return state.nm3u8dl_pending_source = upgrade_candidate state.nm3u8dl_rollover_reason = "quality_upgrade" state.nm3u8dl_renewal_rollover_requested = True @@ -15566,6 +15705,14 @@ def record_runtime_scan_history( ) quality_decision_level = "WARN" + if rejected_upgrade_count: + quality_decision = ( + f"{rejected_upgrade_count} quality-upgrade candidate(s) " + "rejected by IDET confirmation or the bitrate safeguard; " + "current recording continues" + ) + quality_decision_level = "WARN" + if authorization_check_due: replacement_candidates = [ pool_candidate From aa3ace803356da5e31d52e7cd3b2bd5b78453e21 Mon Sep 17 00:00:00 2001 From: sumitdipsite2005 Date: Fri, 9 Oct 2026 19:40:08 -0500 Subject: [PATCH 3/5] Issue 3: regression tests for independent IDET confirmation and bitrate threshold --- tests/test_issue3_upgrade_guards.py | 194 ++++++++++++++++++++++++++++ 1 file changed, 194 insertions(+) create mode 100644 tests/test_issue3_upgrade_guards.py diff --git a/tests/test_issue3_upgrade_guards.py b/tests/test_issue3_upgrade_guards.py new file mode 100644 index 0000000..9f0f75f --- /dev/null +++ b/tests/test_issue3_upgrade_guards.py @@ -0,0 +1,194 @@ +"""Regression checks for Recorder Issue 3 quality-upgrade safeguards. + +The main recorder reads machine-specific OneDrive configuration at import, so +its narrow IDET helper is exercised from its AST without starting a recorder. +""" +from __future__ import annotations + +import ast +from pathlib import Path +from types import SimpleNamespace +from typing import Callable, Optional +import unittest +from unittest.mock import Mock + +from recorder_source.models import SelectionPolicy, SourceCandidate +from recorder_source.selection import ( + bitrate_only_upgrade_below_minimum, + select_quality_upgrade, +) + + +def source(bitrate: int, **changes) -> SourceCandidate: + values = dict( + entry_title="stream", + quality_known=True, + video_width=1920, + video_height=1080, + video_fps=50.0, + video_scan_type="progressive", + video_bitrate_bps=bitrate, + expiry=100000.0, + launchable=True, + ) + values.update(changes) + return SourceCandidate(**values) + + +class BitrateUpgradeSafeguardTests(unittest.TestCase): + def test_below_ten_percent_is_rejected(self): + self.assertTrue( + bitrate_only_upgrade_below_minimum( + source(4_380_000), source(4_478_000), + motion_cap_fps=50, + ) + ) + + def test_exact_ten_percent_passes_without_rounding(self): + self.assertFalse( + bitrate_only_upgrade_below_minimum( + source(4_380_000), source(4_818_000), + motion_cap_fps=50, + ) + ) + + def test_above_ten_percent_passes(self): + self.assertFalse( + bitrate_only_upgrade_below_minimum( + source(4_380_000), source(5_100_000), + motion_cap_fps=50, + ) + ) + + def test_unknown_current_or_alternative_bitrate_preserves_behavior(self): + for current, alternative in ((0, 5_000_000), (4_000_000, 0)): + with self.subTest(current=current, alternative=alternative): + self.assertFalse( + bitrate_only_upgrade_below_minimum( + source(current), source(alternative), + motion_cap_fps=50, + ) + ) + + def test_other_video_quality_improvements_are_not_throttled(self): + cases = [ + (source(4_000_000, video_fps=25), source(4_100_000)), + (source(4_000_000, video_width=1280, video_height=720), + source(4_100_000)), + (source(4_000_000, video_scan_type="interlaced"), + source(4_100_000)), + ] + for current, alternative in cases: + with self.subTest(current=current, alternative=alternative): + self.assertFalse( + bitrate_only_upgrade_below_minimum( + current, alternative, motion_cap_fps=50, + ) + ) + + def test_original_selection_still_ranks_eligible_candidates(self): + policy = SelectionPolicy(900, 900, allow_unknown_expiry=False) + current = source(4_000_000) + small = source(4_100_000, entry_title="small") + meaningful = source(4_500_000, entry_title="meaningful") + selection = select_quality_upgrade( + current, [small, meaningful], 50, policy, now_ts=1000, + ) + self.assertEqual(selection.selected.entry_title, "meaningful") + # The mature selector is untouched; the guard acts only on the chosen + # bitrate-only upgrade. + self.assertTrue( + bitrate_only_upgrade_below_minimum( + current, small, motion_cap_fps=50, + ) + ) + + +class IDETConfirmationTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + root = Path(__file__).resolve().parents[1] + tree = ast.parse((root / "record_dynamic.py").read_text(encoding="utf-8")) + declaration = next( + node for node in tree.body + if isinstance(node, ast.FunctionDef) + and node.name == "_confirm_nm3u8dl_quality_upgrade_idet" + ) + cls.compiled_helper = compile( + ast.Module(body=[declaration], type_ignores=[]), + str(root / "record_dynamic.py"), + "exec", + ) + + def make_helper(self, observed="progressive", sample=None): + self.detect = Mock(return_value=observed) + self.prepare = Mock(return_value=sample or ( + "dash-selected-sample.mp4", {}, "", "dash-selected-sample.mp4" + )) + self.remove = Mock() + env = { + "Optional": Optional, + "Callable": Callable, + "os": SimpleNamespace(remove=self.remove), + "_normalize_nm3u8dl_video_scan_type": lambda x: x or "", + "get_nm3u8dl_effective_headers": lambda h, emit_logs=False: h, + "_nm3u8dl_known_media_probe_keys": lambda c, q, h: [""], + "_detect_nm3u8dl_stream_scan_type_with_idet": self.detect, + "_prepare_nm3u8dl_selected_dash_media_sample": self.prepare, + "_format_candidate_timeout_route": lambda p, u: "test-route", + } + exec(self.compiled_helper, env) + return env["_confirm_nm3u8dl_quality_upgrade_idet"] + + def candidate(self, kind="HLS", **changes): + data = { + "stream_type": kind, + "video_scan_type": "progressive", + "video_scan_type_source": "idet", + "stream_url": "https://test.invalid/master.m3u8", + "manifest_variant_url": "https://test.invalid/variant-1080.m3u8", + "manifest_final_url": "https://test.invalid/master.m3u8", + "effective_headers": {}, + "playlist_url": "https://test.invalid/playlist.m3u", + } + data.update(changes) + return data + + def test_hls_agrees_on_same_selected_variant_without_cache(self): + confirm = self.make_helper() + self.assertEqual(confirm(self.candidate()), "confirmed") + args, kwargs = self.detect.call_args + self.assertEqual(args[0], "https://test.invalid/variant-1080.m3u8") + self.assertIsNone(kwargs["scan_type_cache"]) + + def test_idet_disagreement_rejects_and_inconclusive_rejects(self): + self.assertEqual( + self.make_helper(observed="interlaced")(self.candidate()), "failed" + ) + self.assertEqual( + self.make_helper(observed="")(self.candidate()), "inconclusive" + ) + + def test_non_idet_sources_never_reprobe(self): + confirm = self.make_helper() + result = confirm(self.candidate(video_scan_type_source="manifest")) + self.assertEqual(result, "not_required") + self.detect.assert_not_called() + + def test_dash_rechecks_selected_sample_not_whole_mpd_and_cleans_up(self): + confirm = self.make_helper() + self.assertEqual(confirm(self.candidate(kind="DASH")), "confirmed") + self.assertEqual(self.detect.call_args.args[0], "dash-selected-sample.mp4") + self.assertIsNone(self.detect.call_args.kwargs["scan_type_cache"]) + self.remove.assert_called_once_with("dash-selected-sample.mp4") + + def test_cancelled_confirmation_cannot_approve_upgrade(self): + confirm = self.make_helper() + self.assertEqual( + confirm(self.candidate(), stop_requested=lambda: True), "cancelled" + ) + self.detect.assert_not_called() + + +if __name__ == "__main__": + unittest.main() From f065ce372926b3926bc7cf514d84a015d1eb915b Mon Sep 17 00:00:00 2001 From: sumitdipsite2005 Date: Fri, 9 Oct 2026 19:41:10 -0500 Subject: [PATCH 4/5] Issue 3: confirm selected IDET results before bitrate eligibility safeguard --- record_dynamic.py | 30 +++++++++++++++--------------- 1 file changed, 15 insertions(+), 15 deletions(-) diff --git a/record_dynamic.py b/record_dynamic.py index c8302d0..5bb3693 100644 --- a/record_dynamic.py +++ b/record_dynamic.py @@ -15566,21 +15566,6 @@ def quality_upgrade_cancelled(): candidate_index = upgrade_selection.selected_index proposed_upgrade = remaining_upgrade_candidates[candidate_index] - if source_selection.bitrate_only_upgrade_below_minimum( - running_source, - proposed_upgrade, - motion_cap_fps=NM3U8DL_QUALITY_RANKING_MOTION_CAP_FPS, - minimum_percent=10, - ): - log( - "Quality upgrade bitrate safeguard → REJECTED; " - "bitrate is the only quality difference and the increase " - "is less than 10%; checking other candidates." - ) - rejected_upgrade_count += 1 - del remaining_upgrade_candidates[candidate_index] - continue - idet_confirmation = _confirm_nm3u8dl_quality_upgrade_idet( proposed_upgrade, stop_requested=quality_upgrade_cancelled, @@ -15602,6 +15587,21 @@ def quality_upgrade_cancelled(): if idet_confirmation == "confirmed": log("Quality upgrade IDET confirmation → CONFIRMED.") + if source_selection.bitrate_only_upgrade_below_minimum( + running_source, + proposed_upgrade, + motion_cap_fps=NM3U8DL_QUALITY_RANKING_MOTION_CAP_FPS, + minimum_percent=10, + ): + log( + "Quality upgrade bitrate safeguard → REJECTED; " + "bitrate is the only quality difference and the increase " + "is less than 10%; checking other candidates." + ) + rejected_upgrade_count += 1 + del remaining_upgrade_candidates[candidate_index] + continue + upgrade_candidate = dict(proposed_upgrade) break From 0cb9206c05d6c3617203c2d00cb165cf21c50e4b Mon Sep 17 00:00:00 2001 From: sumitdipsite2005 Date: Fri, 9 Oct 2026 19:42:11 -0500 Subject: [PATCH 5/5] Issue 3: exercise actual quality-upgrade fallback and cancellation decision path --- tests/test_issue3_upgrade_guards.py | 153 ++++++++++++++++++++++++++++ 1 file changed, 153 insertions(+) diff --git a/tests/test_issue3_upgrade_guards.py b/tests/test_issue3_upgrade_guards.py index 9f0f75f..40f02d7 100644 --- a/tests/test_issue3_upgrade_guards.py +++ b/tests/test_issue3_upgrade_guards.py @@ -190,5 +190,158 @@ def test_cancelled_confirmation_cannot_approve_upgrade(self): self.detect.assert_not_called() +class UpgradeDecisionIntegrationTests(unittest.TestCase): + """Exercise the actual upgrade-decision block without recorder startup.""" + + @classmethod + def setUpClass(cls): + root = Path(__file__).resolve().parents[1] + tree = ast.parse((root / "record_dynamic.py").read_text(encoding="utf-8")) + monitor = next( + node for node in tree.body + if isinstance(node, ast.FunctionDef) + and node.name == "monitor_nm3u8dl_playlist_renewal" + ) + decision_if = next( + node for node in ast.walk(monitor) + if isinstance(node, ast.If) + and isinstance(node.test, ast.Name) + and node.test.id == "quality_evaluation_due" + ) + harness = ast.parse("def run_decision():\\n pass\\n".replace("\\n", "\n")) + harness.body[0].body = decision_if.body + cls.decision_code = compile( + ast.fix_missing_locations(harness), + str(root / "record_dynamic.py"), + "exec", + ) + + def evaluate(self, current, alternatives, confirm=None, stopped=False): + import time as time_module + from recorder_source import selection as selection_module + + running = current.to_mapping() + candidate_pool = [x.to_mapping() for x in alternatives] + logs = [] + approvals = [] + state = SimpleNamespace( + nm3u8dl_running_source=running, + nm3u8dl_pending_source=None, + nm3u8dl_rollover_reason=None, + nm3u8dl_renewal_rollover_requested=False, + stop_flag=False, + ) + + def pick(running_source, pool, target_fps, min_remaining, + allow_unknown_expiry=False, now_ts=None): + return select_quality_upgrade( + SourceCandidate.from_mapping(running_source), + [SourceCandidate.from_mapping(item) for item in pool], + target_fps, + SelectionPolicy( + mandatory_min_remaining_sec=900, + upgrade_min_remaining_sec=900, + allow_unknown_expiry=allow_unknown_expiry, + ), + now_ts=1000, + ) + + def choose_confirmation(candidate, stop_requested=None): + if confirm is None: + return "not_required" + result = confirm(candidate) + if stopped: + state.stop_flag = True + return result + + env = { + "stop_event": SimpleNamespace(is_set=lambda: False), + "state": state, + "running_source": running, + "candidate_pool": candidate_pool, + "resolution": {}, + "check_time": 1000.0, + "quality_upgrade_target_fps": 50.0, + "quality_min_remaining_min": 15, + "allow_unknown_expiry": False, + "NM3U8DL_QUALITY_RANKING_MOTION_CAP_FPS": 50.0, + "source_selection": selection_module, + "time": SimpleNamespace(time=lambda: 1000), + "get_nm3u8dl_quality_upgrade_selection_decision": pick, + "_confirm_nm3u8dl_quality_upgrade_idet": choose_confirmation, + "log": lambda message, level="INFO": logs.append(message), + "log_quality_upgrade_found": ( + lambda candidate, check_time: approvals.append(candidate) + ), + "_nm3u8dl_video_quality_rank": ( + lambda item: selection_module.video_quality_rank( + item, motion_cap_fps=50 + ) + ), + "_nm3u8dl_ranking_motion_fps": ( + lambda item: selection_module.ranking_motion_fps( + item, motion_cap_fps=50 + ) + ), + "get_nm3u8dl_candidate_quality_rank": ( + lambda item: ( + int(item.get("preferred_qualifier_score") or 0), + *selection_module.video_quality_rank(item, motion_cap_fps=50), + ) + ), + "format_nm3u8dl_candidate_quality": ( + lambda item: item.get("entry_title") or "stream" + ), + "fmt_hms": lambda seconds: str(seconds), + } + exec(self.decision_code, env) + env["run_decision"]() + return state, approvals, logs + + def test_failed_idet_rechecks_alternative_without_switching_to_failed(self): + current = source(4_000_000, entry_title="current") + first = source( + 5_200_000, entry_title="failed-idet", + video_scan_type_source="idet", + ) + second = source( + 4_600_000, entry_title="alternative", + video_scan_type_source="manifest", + ) + state, approvals, logs = self.evaluate( + current, [second, first], + confirm=lambda candidate: ( + "failed" if candidate["entry_title"] == "failed-idet" + else "not_required" + ), + ) + self.assertEqual(state.nm3u8dl_pending_source["entry_title"], "alternative") + self.assertEqual([x["entry_title"] for x in approvals], ["alternative"]) + self.assertTrue(any("FAILED" in item for item in logs)) + self.assertTrue(any("alternative candidate" in item for item in logs)) + + def test_small_bitrate_gain_keeps_recording(self): + current = source(4_380_000, entry_title="current") + slight = source( + 4_478_000, entry_title="slight", + video_scan_type_source="manifest", + ) + state, approvals, logs = self.evaluate(current, [slight]) + self.assertIsNone(state.nm3u8dl_pending_source) + self.assertEqual(approvals, []) + self.assertTrue(any("less than 10%" in item for item in logs)) + + def test_recovery_started_during_confirmation_never_commits_upgrade(self): + current = source(4_000_000) + alternative = source( + 4_800_000, video_scan_type_source="idet", + ) + state, approvals, _ = self.evaluate( + current, [alternative], confirm=lambda c: "confirmed", stopped=True, + ) + self.assertIsNone(state.nm3u8dl_pending_source) + self.assertEqual(approvals, []) + + if __name__ == "__main__": unittest.main()