mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-18 09:06:06 +00:00
72c0a4e69a
Refactor tests for VOD provider failover logic to improve clarity and remove unnecessary components. Introduce new tests for order candidates and ensure no database access occurs during processing.
90 lines
3.3 KiB
Python
90 lines
3.3 KiB
Python
"""
|
|
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_content_and_relation()` now materialises the active, priority-ordered
|
|
relations once (single DB query) and returns that list. `_order_candidates()`
|
|
is a pure in-memory helper that moves the preferred relation to the front and
|
|
removes duplicates, so the initial connection path hits the database exactly
|
|
once. stream_vod()/head_vod() then walk the ordered list and use the first
|
|
account with spare capacity.
|
|
|
|
These tests cover the in-memory ordering helper: preferred-first placement,
|
|
de-duplication, empty-input fallbacks, and the guarantee that it performs no
|
|
database access.
|
|
"""
|
|
|
|
from unittest.mock import MagicMock
|
|
from django.test import TestCase
|
|
|
|
from apps.proxy.vod_proxy.views import _order_candidates
|
|
|
|
|
|
def _rel(rel_id, priority):
|
|
rel = MagicMock()
|
|
rel.id = rel_id
|
|
rel.m3u_account = MagicMock()
|
|
rel.m3u_account.priority = priority
|
|
return rel
|
|
|
|
|
|
class TestOrderCandidates(TestCase):
|
|
def test_preferred_relation_is_placed_first(self):
|
|
preferred = _rel(rel_id=2, priority=0)
|
|
candidates = [_rel(rel_id=1, priority=5), _rel(rel_id=3, priority=2), preferred]
|
|
|
|
result = _order_candidates(candidates, preferred_relation=preferred)
|
|
|
|
self.assertEqual(result[0].id, 2, "Preferred relation must be first")
|
|
self.assertEqual({r.id for r in result}, {1, 2, 3})
|
|
|
|
def test_preferred_relation_not_duplicated(self):
|
|
preferred = _rel(rel_id=2, priority=0)
|
|
candidates = [preferred, _rel(rel_id=1, priority=5)]
|
|
|
|
result = _order_candidates(candidates, 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_no_preferred_keeps_order(self):
|
|
candidates = [_rel(rel_id=1, priority=0), _rel(rel_id=2, priority=5)]
|
|
|
|
result = _order_candidates(candidates, preferred_relation=None)
|
|
|
|
self.assertEqual([r.id for r in result], [1, 2])
|
|
|
|
def test_empty_with_preferred_returns_preferred(self):
|
|
preferred = _rel(rel_id=7, priority=0)
|
|
|
|
result = _order_candidates([], preferred_relation=preferred)
|
|
|
|
self.assertEqual(result, [preferred])
|
|
|
|
def test_empty_without_preferred_returns_empty(self):
|
|
result = _order_candidates([], preferred_relation=None)
|
|
|
|
self.assertEqual(result, [])
|
|
|
|
def test_no_database_access(self):
|
|
"""The helper must be pure in-memory: it must never touch the ORM."""
|
|
class Boom:
|
|
def __init__(self, rel_id, priority):
|
|
self.id = rel_id
|
|
self.m3u_account = MagicMock()
|
|
self.m3u_account.priority = priority
|
|
|
|
def __getattr__(self, name):
|
|
if name in ('filter', 'objects', 'all', 'select_related', 'order_by'):
|
|
raise AssertionError(f"ORM access attempted via .{name}()")
|
|
raise AttributeError(name)
|
|
|
|
candidates = [Boom(1, 0), Boom(2, 5)]
|
|
|
|
result = _order_candidates(candidates, preferred_relation=None)
|
|
|
|
self.assertEqual([r.id for r in result], [1, 2])
|