From 5c43f14c19278838886e0815e243b29ea95ab44a Mon Sep 17 00:00:00 2001 From: francescodg89-crypto Date: Tue, 30 Jun 2026 13:32:40 +0200 Subject: [PATCH] Add files via upload --- .../vod_proxy/tests/test vod failover.py | 166 ++++++++++++++++++ 1 file changed, 166 insertions(+) create mode 100644 apps/proxy/vod_proxy/tests/test vod failover.py diff --git a/apps/proxy/vod_proxy/tests/test vod failover.py b/apps/proxy/vod_proxy/tests/test vod failover.py new file mode 100644 index 00000000..2f4543c7 --- /dev/null +++ b/apps/proxy/vod_proxy/tests/test vod failover.py @@ -0,0 +1,166 @@ +""" +Tests for VOD provider failover (PR: "Add VOD failover logic for M3U relations"). + +The VOD proxy previously selected a single highest-priority relation and +returned 503 if that account was at capacity, without trying other accounts +that carry the same title. `_get_all_relations_ordered()` is the helper that +enables failover by returning ALL active relations ordered by account +priority, with the already-selected preferred relation first. + +These tests cover the helper's ordering, preferred-first placement, +de-duplication, content-type handling (movie / episode / series) and its +defensive fallbacks. +""" + +from unittest.mock import MagicMock, patch +from django.test import TestCase + + +def _rel(rel_id, priority): + """Build a fake relation with an account of the given priority.""" + rel = MagicMock() + rel.id = rel_id + rel.m3u_account = MagicMock() + rel.m3u_account.priority = priority + return rel + + +class FakeRelationQuerySet: + """Minimal stand-in for the m3u_relations manager/queryset chain. + + Records the filter/order_by calls so tests can assert the helper only + asks for active accounts, and returns a fixed, already-ordered list. + """ + + def __init__(self, relations): + self._relations = relations + self.filter_kwargs = None + self.order_by_args = None + + def filter(self, **kwargs): + self.filter_kwargs = kwargs + return self + + def select_related(self, *args): + return self + + def order_by(self, *args): + self.order_by_args = args + return self + + def __iter__(self): + return iter(self._relations) + + def __list__(self): + return list(self._relations) + + +def _import_helper(): + """Import the helper with heavy Django model deps stubbed out.""" + import sys + for mod in ['apps.vod.models', 'apps.m3u.models', 'core.utils']: + if mod not in sys.modules: + sys.modules[mod] = MagicMock() + from apps.proxy.vod_proxy.views import _get_all_relations_ordered + return _get_all_relations_ordered + + +class TestGetAllRelationsOrdered(TestCase): + def setUp(self): + self.helper = _import_helper() + + def test_preferred_relation_is_placed_first(self): + """The already-selected relation must come first even if not top priority.""" + preferred = _rel(rel_id=2, priority=0) + all_rels = [_rel(rel_id=1, priority=5), preferred, _rel(rel_id=3, priority=2)] + + content = MagicMock() + content.m3u_relations = FakeRelationQuerySet(all_rels) + + result = self.helper(content, 'movie', preferred_relation=preferred) + + self.assertEqual(result[0].id, 2, "Preferred relation must be first") + # All relations are still present + self.assertEqual({r.id for r in result}, {1, 2, 3}) + + def test_preferred_relation_not_duplicated(self): + """If the preferred relation is also in the queryset, it must appear once.""" + preferred = _rel(rel_id=2, priority=0) + all_rels = [preferred, _rel(rel_id=1, priority=5)] + + content = MagicMock() + content.m3u_relations = FakeRelationQuerySet(all_rels) + + result = self.helper(content, 'movie', preferred_relation=preferred) + + ids = [r.id for r in result] + self.assertEqual(ids.count(2), 1, "Preferred relation must not be duplicated") + self.assertEqual(len(result), 2) + + def test_only_active_accounts_requested(self): + """Helper must filter on active accounts only.""" + all_rels = [_rel(rel_id=1, priority=0)] + qs = FakeRelationQuerySet(all_rels) + content = MagicMock() + content.m3u_relations = qs + + self.helper(content, 'movie', preferred_relation=None) + + self.assertEqual(qs.filter_kwargs, {'m3u_account__is_active': True}) + + def test_no_preferred_returns_all(self): + """Without a preferred relation, all relations are returned.""" + all_rels = [_rel(rel_id=1, priority=0), _rel(rel_id=2, priority=5)] + content = MagicMock() + content.m3u_relations = FakeRelationQuerySet(all_rels) + + result = self.helper(content, 'movie', preferred_relation=None) + + self.assertEqual({r.id for r in result}, {1, 2}) + + def test_series_uses_first_episode_relations(self): + """For series, relations come from the first episode.""" + episode = MagicMock() + episode.m3u_relations = FakeRelationQuerySet([_rel(rel_id=10, priority=0)]) + + series = MagicMock() + series.episodes.first.return_value = episode + + result = self.helper(series, 'series', preferred_relation=None) + + series.episodes.first.assert_called_once() + self.assertEqual([r.id for r in result], [10]) + + def test_missing_relations_attr_falls_back_to_preferred(self): + """If the content object has no m3u_relations, return just the preferred.""" + preferred = _rel(rel_id=4, priority=0) + + class NoRelations: + pass + + result = self.helper(NoRelations(), 'movie', preferred_relation=preferred) + + self.assertEqual(result, [preferred]) + + def test_missing_relations_attr_and_no_preferred_returns_empty(self): + """No relations and no preferred relation must yield an empty list, not raise.""" + class NoRelations: + pass + + result = self.helper(NoRelations(), 'movie', preferred_relation=None) + + self.assertEqual(result, []) + + def test_db_error_falls_back_to_preferred(self): + """Any error while collecting relations must not raise; return preferred.""" + preferred = _rel(rel_id=7, priority=0) + + content = MagicMock() + # Accessing .filter raises to simulate a DB/ORM failure + broken_qs = MagicMock() + broken_qs.filter.side_effect = RuntimeError("DB down") + content.m3u_relations = broken_qs + + result = self.helper(content, 'movie', preferred_relation=preferred) + + self.assertEqual(result, [preferred])