mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-19 09:34:38 +00:00
223dff33ed
Add end-to-end catch-up support for XC clients and native apps: provider proxy with failover and per-viewer session pooling, REST session minting for tokenless playback URLs, catch-up admin stats, combined connection stats, and Stats UI with dedicated cards plus websocket updates. Includes Redis namespace consolidation under timeshift:* (dropping legacy timeshift_ id prefixes), dedicated catch-up stop by session_id, and cleaner channel/client metadata split for stats keys. Closes #133
347 lines
14 KiB
Python
347 lines
14 KiB
Python
"""Tests for catch-up stats API and builders."""
|
|
|
|
import json
|
|
import time
|
|
from datetime import datetime, timedelta, timezone as dt_timezone
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from django.test import TestCase
|
|
from rest_framework.test import APIRequestFactory, force_authenticate
|
|
|
|
from apps.accounts.models import User
|
|
from apps.proxy.live_proxy.constants import ChannelMetadataField, ChannelState
|
|
from apps.timeshift.redis_keys import TimeshiftRedisKeys as RedisKeys, parse_stats_channel_id
|
|
from apps.timeshift import stats_views
|
|
from apps.timeshift.helpers import get_programme_info
|
|
from apps.timeshift.stats import (
|
|
build_timeshift_stats_data,
|
|
compute_playback_base_from_byte_range,
|
|
compute_playback_position_secs,
|
|
find_stats_channel_for_session,
|
|
resolve_stats_playback_fields,
|
|
seed_stream_stats_metadata,
|
|
stream_stats_to_metadata_fields,
|
|
)
|
|
from apps.timeshift.tests.test_views import _FakeRedis
|
|
|
|
TEST_SESSION_ID = "testsession1"
|
|
STATS_CHANNEL_ID = f"8_{TEST_SESSION_ID}"
|
|
|
|
|
|
class TimeshiftHelperParsingTests(TestCase):
|
|
def test_parse_stats_channel_id(self):
|
|
parsed = parse_stats_channel_id(STATS_CHANNEL_ID)
|
|
self.assertEqual(parsed["channel_id"], 8)
|
|
self.assertEqual(parsed["session_id"], TEST_SESSION_ID)
|
|
|
|
|
|
class GetProgrammeInfoTests(TestCase):
|
|
def _channel_with_programme(self, title="Evening News", sub_title="Local Edition", minutes=45):
|
|
start = datetime(2026, 6, 8, 17, 0, tzinfo=dt_timezone.utc)
|
|
programme = MagicMock(
|
|
title=title,
|
|
sub_title=sub_title,
|
|
description="Details",
|
|
start_time=start,
|
|
end_time=start + timedelta(minutes=minutes),
|
|
)
|
|
channel = MagicMock()
|
|
channel.epg_data.programs.filter.return_value.first.return_value = programme
|
|
return channel
|
|
|
|
def test_get_programme_info_resolves_title(self):
|
|
info = get_programme_info(self._channel_with_programme(), "2026-06-08:17-00")
|
|
self.assertEqual(info["title"], "Evening News")
|
|
self.assertEqual(info["sub_title"], "Local Edition")
|
|
self.assertEqual(info["duration_secs"], 45 * 60)
|
|
|
|
|
|
class ComputePlaybackPositionTests(TestCase):
|
|
EPG_START = "2026-07-10T14:00:00+00:00"
|
|
|
|
def test_byte_range_base_plus_elapsed(self):
|
|
pos = compute_playback_position_secs(
|
|
"2026-07-10:14-00",
|
|
self.EPG_START,
|
|
position_anchor_at=1000.0,
|
|
current_time=1030.0,
|
|
duration_secs=3600,
|
|
playback_base_secs=1800.0,
|
|
)
|
|
self.assertAlmostEqual(pos, 1830.0)
|
|
|
|
def test_url_offset_plus_elapsed(self):
|
|
# Seeked to 14:19 (19 min into a 14:00 programme), stream opened 30s ago.
|
|
pos = compute_playback_position_secs(
|
|
"2026-07-10:14-19",
|
|
self.EPG_START,
|
|
position_anchor_at=1000.0,
|
|
current_time=1030.0,
|
|
duration_secs=3600,
|
|
)
|
|
self.assertAlmostEqual(pos, 19 * 60 + 30)
|
|
|
|
def test_capped_at_duration(self):
|
|
pos = compute_playback_position_secs(
|
|
"2026-07-10:14-19",
|
|
self.EPG_START,
|
|
position_anchor_at=1000.0,
|
|
current_time=100000.0,
|
|
duration_secs=1800,
|
|
)
|
|
self.assertEqual(pos, 1800)
|
|
|
|
def test_missing_epg_returns_none(self):
|
|
self.assertIsNone(
|
|
compute_playback_position_secs(
|
|
"2026-07-10:14-19", None, 1000.0, 1030.0,
|
|
)
|
|
)
|
|
|
|
def test_no_anchor_uses_url_offset_only(self):
|
|
pos = compute_playback_position_secs(
|
|
"2026-07-10:14-05",
|
|
self.EPG_START,
|
|
position_anchor_at=None,
|
|
current_time=1030.0,
|
|
duration_secs=3600,
|
|
)
|
|
self.assertAlmostEqual(pos, 5 * 60)
|
|
|
|
|
|
class ByteRangePlaybackTests(TestCase):
|
|
def test_compute_playback_base_from_byte_range(self):
|
|
base = compute_playback_base_from_byte_range(
|
|
506786520, 833563944, 3900,
|
|
)
|
|
self.assertAlmostEqual(base, 2371.0, delta=2.0)
|
|
|
|
def test_resolve_byte_seek_reanchors(self):
|
|
base, anchor = resolve_stats_playback_fields(
|
|
timestamp_utc="2026-07-09:20:00:00",
|
|
existing_programme_start="2026-07-09:20:00:00",
|
|
existing_position_anchor="1000.0",
|
|
existing_playback_base=None,
|
|
range_start=506786520,
|
|
representation_length=833563944,
|
|
programme_duration_secs=3900,
|
|
now="2000.0",
|
|
)
|
|
self.assertAlmostEqual(base, 2371.0, delta=2.0)
|
|
self.assertEqual(anchor, "2000.0")
|
|
|
|
def test_resolve_same_programme_preserves_without_range(self):
|
|
base, anchor = resolve_stats_playback_fields(
|
|
timestamp_utc="2026-06-08:17-00",
|
|
existing_programme_start="2026-06-08:17-00",
|
|
existing_position_anchor="1000.0",
|
|
existing_playback_base="900.0",
|
|
range_start=None,
|
|
representation_length=833563944,
|
|
programme_duration_secs=3900,
|
|
now="2000.0",
|
|
)
|
|
self.assertAlmostEqual(base, 900.0)
|
|
self.assertEqual(anchor, "1000.0")
|
|
|
|
|
|
class TimeshiftStreamStatsTests(TestCase):
|
|
def test_stream_stats_to_metadata_fields(self):
|
|
fields = stream_stats_to_metadata_fields({
|
|
"resolution": "1920x1080",
|
|
"source_fps": 30,
|
|
"video_codec": "h264",
|
|
})
|
|
self.assertEqual(fields[ChannelMetadataField.RESOLUTION], "1920x1080")
|
|
self.assertEqual(fields[ChannelMetadataField.SOURCE_FPS], "30")
|
|
self.assertEqual(fields[ChannelMetadataField.VIDEO_CODEC], "h264")
|
|
self.assertIn(ChannelMetadataField.STREAM_INFO_UPDATED, fields)
|
|
|
|
def test_seed_stream_stats_metadata_skips_when_stream_unchanged(self):
|
|
redis = _FakeRedis()
|
|
metadata_key = RedisKeys.channel_metadata(STATS_CHANNEL_ID)
|
|
redis.hset(metadata_key, ChannelMetadataField.STREAM_ID, "7")
|
|
payload = {}
|
|
seed_stream_stats_metadata(
|
|
redis, metadata_key, payload,
|
|
stats_stream_id=7,
|
|
stream_stats={"resolution": "1920x1080"},
|
|
)
|
|
self.assertEqual(payload, {})
|
|
|
|
def test_seed_stream_stats_metadata_applies_new_stream(self):
|
|
redis = _FakeRedis()
|
|
metadata_key = RedisKeys.channel_metadata(STATS_CHANNEL_ID)
|
|
payload = {}
|
|
seed_stream_stats_metadata(
|
|
redis, metadata_key, payload,
|
|
stats_stream_id=7,
|
|
stream_stats={"resolution": "1280x720", "video_codec": "h264"},
|
|
)
|
|
self.assertEqual(payload[ChannelMetadataField.STREAM_ID], "7")
|
|
self.assertEqual(payload[ChannelMetadataField.RESOLUTION], "1280x720")
|
|
|
|
|
|
class BuildTimeshiftStatsDataTests(TestCase):
|
|
def setUp(self):
|
|
self.channel_id = 42
|
|
self.session_id = TEST_SESSION_ID
|
|
self.stats_channel_id = f"{self.channel_id}_{self.session_id}"
|
|
self.redis = _FakeRedis()
|
|
now = time.time()
|
|
metadata_key = RedisKeys.channel_metadata(self.stats_channel_id)
|
|
client_set_key = RedisKeys.clients(self.stats_channel_id)
|
|
client_key = RedisKeys.client_metadata(self.stats_channel_id, self.session_id)
|
|
|
|
self.redis.hset(
|
|
metadata_key,
|
|
mapping={
|
|
ChannelMetadataField.STATE: ChannelState.ACTIVE,
|
|
ChannelMetadataField.CHANNEL_ID: str(self.channel_id),
|
|
ChannelMetadataField.CHANNEL_UUID: "00000000-0000-0000-0000-000000000042",
|
|
ChannelMetadataField.CHANNEL_NAME: "Catch-up Stats Channel",
|
|
ChannelMetadataField.INIT_TIME: str(now - 120),
|
|
ChannelMetadataField.TOTAL_BYTES: "1048576",
|
|
ChannelMetadataField.LOGO_ID: "0",
|
|
ChannelMetadataField.RESOLUTION: "1920x1080",
|
|
ChannelMetadataField.SOURCE_FPS: "29.97",
|
|
ChannelMetadataField.VIDEO_CODEC: "h264",
|
|
},
|
|
)
|
|
self.redis.sadd(client_set_key, self.session_id)
|
|
self.anchor_at = now - 30
|
|
self.redis.hset(
|
|
client_key,
|
|
mapping={
|
|
"user_agent": "VLC/3.0.0",
|
|
"ip_address": "10.0.0.5",
|
|
"connected_at": str(now - 60),
|
|
"user_id": "1",
|
|
"username": "viewer",
|
|
"programme_vid": f"8_{self.channel_id}_2026-06-08-17-00_111",
|
|
"programme_start": "2026-06-08:17-15",
|
|
"position_anchor_at": str(self.anchor_at),
|
|
},
|
|
)
|
|
|
|
@patch("apps.timeshift.stats.Channel")
|
|
def test_build_timeshift_stats_data_redis_only_programme_fields(
|
|
self, mock_channel_model,
|
|
):
|
|
channel = MagicMock()
|
|
channel.id = self.channel_id
|
|
channel.name = "Catch-up Stats Channel"
|
|
channel.uuid = "00000000-0000-0000-0000-000000000042"
|
|
channel.logo_id = None
|
|
channel.logo = None
|
|
mock_channel_model.objects.select_related.return_value.filter.return_value = [
|
|
channel,
|
|
]
|
|
payload = build_timeshift_stats_data(self.redis)
|
|
self.assertEqual(payload["total_connections"], 1)
|
|
self.assertEqual(len(payload["timeshift_sessions"]), 1)
|
|
session = payload["timeshift_sessions"][0]
|
|
self.assertEqual(session["session_id"], self.session_id)
|
|
self.assertEqual(session["programme_start"], "2026-06-08:17-15")
|
|
self.assertAlmostEqual(session["position_anchor_at"], self.anchor_at, places=3)
|
|
self.assertNotIn("programme_title", session)
|
|
self.assertNotIn("playback_position_secs", session)
|
|
self.assertEqual(session["channel_name"], "Catch-up Stats Channel")
|
|
self.assertEqual(session["resolution"], "1920x1080")
|
|
self.assertEqual(session["connections"][0]["ip_address"], "10.0.0.5")
|
|
|
|
def test_find_stats_channel_for_session(self):
|
|
found = find_stats_channel_for_session(self.redis, self.session_id)
|
|
self.assertEqual(found, self.stats_channel_id)
|
|
|
|
@patch("apps.timeshift.stats.Channel")
|
|
def test_skips_clients_missing_required_metadata(self, mock_channel_model):
|
|
mock_channel_model.objects.select_related.return_value.filter.return_value = []
|
|
orphan_key = RedisKeys.client_metadata(self.stats_channel_id, "orphan")
|
|
self.redis.hset(orphan_key, mapping={"user_agent": "VLC"})
|
|
self.redis.sadd(RedisKeys.clients(self.stats_channel_id), "orphan")
|
|
payload = build_timeshift_stats_data(self.redis)
|
|
self.assertEqual(payload["total_connections"], 1)
|
|
|
|
|
|
class TimeshiftStatsApiTests(TestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
cls.admin = User.objects.create(
|
|
username="catchup-stats-admin",
|
|
user_level=User.UserLevel.ADMIN,
|
|
)
|
|
|
|
def setUp(self):
|
|
self.factory = APIRequestFactory()
|
|
self.redis = _FakeRedis()
|
|
self.session_id = TEST_SESSION_ID
|
|
self.stats_channel_id = f"8_{self.session_id}"
|
|
|
|
@patch("apps.timeshift.stats_views.RedisClient.get_client")
|
|
@patch("apps.timeshift.stats_views.build_timeshift_stats_data")
|
|
def test_timeshift_stats_requires_admin(self, mock_build, redis_mock):
|
|
mock_build.return_value = {
|
|
"timeshift_sessions": [],
|
|
"total_connections": 0,
|
|
"timestamp": 1,
|
|
}
|
|
redis_mock.return_value = self.redis
|
|
|
|
request = self.factory.get("/proxy/catchup/stats/")
|
|
force_authenticate(request, user=self.admin)
|
|
response = stats_views.timeshift_stats(request)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn("timeshift_sessions", json.loads(response.content))
|
|
|
|
@patch("apps.timeshift.stats_views.stop_timeshift_client")
|
|
@patch("apps.timeshift.stats_views.RedisClient.get_client")
|
|
def test_stop_timeshift_session(self, redis_mock, stop_mock):
|
|
redis_mock.return_value = self.redis
|
|
stop_mock.return_value = {"status": "success"}
|
|
|
|
metadata_key = RedisKeys.channel_metadata(self.stats_channel_id)
|
|
client_key = RedisKeys.client_metadata(self.stats_channel_id, self.session_id)
|
|
self.redis.hset(metadata_key, mapping={ChannelMetadataField.STATE: ChannelState.ACTIVE})
|
|
self.redis.sadd(RedisKeys.clients(self.stats_channel_id), self.session_id)
|
|
self.redis.hset(
|
|
client_key,
|
|
mapping={
|
|
"programme_start": "2026-06-08:17-00",
|
|
},
|
|
)
|
|
|
|
request = self.factory.post(
|
|
"/proxy/catchup/stop_client/",
|
|
{"session_id": self.session_id},
|
|
format="json",
|
|
)
|
|
force_authenticate(request, user=self.admin)
|
|
response = stats_views.stop_timeshift_session(request)
|
|
self.assertEqual(response.status_code, 200)
|
|
stop_mock.assert_called_once_with(
|
|
self.redis, self.stats_channel_id, self.session_id,
|
|
)
|
|
|
|
@patch("apps.timeshift.stats_views.get_catchup_programmes_for_sessions")
|
|
def test_catchup_programmes_batch(self, programmes_mock):
|
|
programmes_mock.return_value = [{
|
|
"session_id": self.session_id,
|
|
"title": "Evening News",
|
|
"start_time": "2026-06-08T17:00:00+00:00",
|
|
}]
|
|
request = self.factory.post(
|
|
"/proxy/catchup/programs/",
|
|
{"sessions": [{
|
|
"session_id": self.session_id,
|
|
"channel_uuid": "00000000-0000-0000-0000-000000000099",
|
|
"programme_start": "2026-06-08:17-00",
|
|
}]},
|
|
format="json",
|
|
)
|
|
force_authenticate(request, user=self.admin)
|
|
response = stats_views.catchup_programmes(request)
|
|
self.assertEqual(response.status_code, 200)
|
|
body = json.loads(response.content)
|
|
self.assertEqual(body["sessions"][0]["title"], "Evening News")
|
|
programmes_mock.assert_called_once()
|