mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-18 17:16:26 +00:00
0dc8898e8b
- Moved the `programme_index` from the `EPGSource` model to a new `EPGSourceIndex` table, ensuring that the large JSON blob is only loaded when explicitly accessed, thus improving query performance and memory efficiency. - Updated related queries and API views to utilize the new structure, including adjustments to EPG generation and import logic to prevent unnecessary data loading. - Enhanced memory management in the EPG grid endpoint to reduce worker RSS during response handling.
310 lines
12 KiB
Python
310 lines
12 KiB
Python
from unittest.mock import patch, MagicMock
|
|
|
|
from django.test import TestCase, SimpleTestCase
|
|
|
|
from apps.epg.models import EPGSource, EPGSourceIndex
|
|
from core.models import CoreSettings, DVR_SETTINGS_KEY, EPG_SETTINGS_KEY
|
|
|
|
|
|
class DispatcharrUserAgentTests(TestCase):
|
|
@patch('version.__version__', '1.2.3')
|
|
def test_dispatcharr_user_agent(self):
|
|
from core.utils import dispatcharr_user_agent
|
|
self.assertEqual(dispatcharr_user_agent(), 'Dispatcharr/1.2.3')
|
|
|
|
def test_dispatcharr_dvr_user_agent(self):
|
|
from core.utils import dispatcharr_dvr_user_agent
|
|
self.assertEqual(dispatcharr_dvr_user_agent(42), 'Dispatcharr-DVR/recording-42')
|
|
|
|
@patch('version.__version__', '1.2.3')
|
|
def test_dispatcharr_http_headers_with_token(self):
|
|
from core.utils import dispatcharr_http_headers
|
|
headers = dispatcharr_http_headers(token='tok123')
|
|
self.assertEqual(headers, {
|
|
'User-Agent': 'Dispatcharr/1.2.3',
|
|
'Content-Type': 'application/json',
|
|
'token': 'tok123',
|
|
})
|
|
|
|
@patch('version.__version__', '1.2.3')
|
|
def test_dispatcharr_http_headers_without_content_type(self):
|
|
from core.utils import dispatcharr_http_headers
|
|
self.assertEqual(
|
|
dispatcharr_http_headers(content_type=None),
|
|
{'User-Agent': 'Dispatcharr/1.2.3'},
|
|
)
|
|
|
|
|
|
class ProgrammeIndexRebuildTests(TestCase):
|
|
def test_startup_rebuild_does_not_lock_out_queued_build_task(self):
|
|
source = EPGSource.objects.create(
|
|
name="Missing Index",
|
|
source_type="xmltv",
|
|
is_active=True,
|
|
)
|
|
|
|
class FakeRedis:
|
|
def __init__(self):
|
|
self.keys = set()
|
|
|
|
def set(self, key, value, nx=False, ex=None):
|
|
if nx and key in self.keys:
|
|
return False
|
|
self.keys.add(key)
|
|
return True
|
|
|
|
def delete(self, key):
|
|
self.keys.discard(key)
|
|
|
|
fake_redis = FakeRedis()
|
|
|
|
from apps.epg.tasks import build_programme_index_task
|
|
from core.tasks import _rebuild_programme_indices
|
|
|
|
def run_task_immediately(source_id):
|
|
build_programme_index_task(source_id)
|
|
|
|
with patch(
|
|
"core.tasks.RedisClient.get_client", return_value=fake_redis
|
|
), patch(
|
|
"core.utils.RedisClient.get_client", return_value=fake_redis
|
|
), patch(
|
|
"apps.epg.tasks.build_programme_index"
|
|
) as mock_build, patch(
|
|
"apps.epg.tasks.build_programme_index_task.delay",
|
|
side_effect=run_task_immediately,
|
|
):
|
|
_rebuild_programme_indices()
|
|
|
|
mock_build.assert_called_once_with(source.id)
|
|
|
|
|
|
class GetDvrSeriesRulesTest(TestCase):
|
|
"""Verify get_dvr_series_rules handles corrupted stored data."""
|
|
|
|
def _set_series_rules_raw(self, raw_value):
|
|
"""Write a raw series_rules value into the DB, bypassing set_dvr_series_rules."""
|
|
obj, _ = CoreSettings.objects.get_or_create(
|
|
key=DVR_SETTINGS_KEY,
|
|
defaults={"name": "DVR Settings", "value": {}},
|
|
)
|
|
current = obj.value if isinstance(obj.value, dict) else {}
|
|
current["series_rules"] = raw_value
|
|
obj.value = current
|
|
obj.save()
|
|
|
|
def test_valid_rules_returned_as_is(self):
|
|
rules = [{"tvg_id": "abc", "mode": "all", "title": "Show"}]
|
|
self._set_series_rules_raw(rules)
|
|
result = CoreSettings.get_dvr_series_rules()
|
|
self.assertEqual(result, rules)
|
|
|
|
def test_non_dict_elements_filtered(self):
|
|
"""Strings in the list cause 'str' has no attribute 'get'."""
|
|
self._set_series_rules_raw(["bad_string", {"tvg_id": "abc", "mode": "all", "title": ""}])
|
|
result = CoreSettings.get_dvr_series_rules()
|
|
self.assertEqual(len(result), 1)
|
|
self.assertEqual(result[0]["tvg_id"], "abc")
|
|
|
|
def test_non_list_value_returns_empty(self):
|
|
"""If series_rules is a JSON string instead of a list, return empty."""
|
|
self._set_series_rules_raw("[]")
|
|
result = CoreSettings.get_dvr_series_rules()
|
|
self.assertEqual(result, [])
|
|
|
|
def test_none_value_returns_empty(self):
|
|
self._set_series_rules_raw(None)
|
|
result = CoreSettings.get_dvr_series_rules()
|
|
self.assertEqual(result, [])
|
|
|
|
def test_mixed_corrupt_elements(self):
|
|
self._set_series_rules_raw([42, None, True, {"tvg_id": "x", "mode": "new", "title": "T"}])
|
|
result = CoreSettings.get_dvr_series_rules()
|
|
self.assertEqual(len(result), 1)
|
|
self.assertEqual(result[0]["tvg_id"], "x")
|
|
|
|
|
|
class SetDvrSeriesRulesTest(TestCase):
|
|
"""Verify set_dvr_series_rules sanitizes input before persisting."""
|
|
|
|
def test_valid_rules_persisted(self):
|
|
rules = [{"tvg_id": "abc", "mode": "all", "title": "Show"}]
|
|
result = CoreSettings.set_dvr_series_rules(rules)
|
|
self.assertEqual(result, rules)
|
|
self.assertEqual(CoreSettings.get_dvr_series_rules(), rules)
|
|
|
|
def test_non_dict_elements_stripped_on_write(self):
|
|
dirty = ["bad", 42, {"tvg_id": "abc", "mode": "all", "title": ""}]
|
|
result = CoreSettings.set_dvr_series_rules(dirty)
|
|
self.assertEqual(len(result), 1)
|
|
self.assertEqual(result[0]["tvg_id"], "abc")
|
|
self.assertEqual(CoreSettings.get_dvr_series_rules(), result)
|
|
|
|
def test_non_list_input_stores_empty(self):
|
|
result = CoreSettings.set_dvr_series_rules("not a list")
|
|
self.assertEqual(result, [])
|
|
self.assertEqual(CoreSettings.get_dvr_series_rules(), [])
|
|
|
|
|
|
class CoreSettingsSerializerDvrTest(TestCase):
|
|
"""Verify the generic settings API sanitizes series_rules on save."""
|
|
|
|
def test_serializer_strips_corrupt_series_rules(self):
|
|
"""Settings page round-trip must not persist corrupt series_rules."""
|
|
from core.serializers import CoreSettingsSerializer
|
|
|
|
obj, _ = CoreSettings.objects.get_or_create(
|
|
key=DVR_SETTINGS_KEY,
|
|
defaults={"name": "DVR Settings", "value": {"series_rules": []}},
|
|
)
|
|
dirty_value = {
|
|
**obj.value,
|
|
"series_rules": ["bad", {"tvg_id": "ok", "mode": "all", "title": ""}],
|
|
}
|
|
serializer = CoreSettingsSerializer(obj, data={"value": dirty_value}, partial=True)
|
|
serializer.is_valid(raise_exception=True)
|
|
serializer.save()
|
|
obj.refresh_from_db()
|
|
rules = obj.value.get("series_rules", [])
|
|
self.assertEqual(len(rules), 1)
|
|
self.assertEqual(rules[0]["tvg_id"], "ok")
|
|
|
|
def test_serializer_handles_non_list_series_rules(self):
|
|
from core.serializers import CoreSettingsSerializer
|
|
|
|
obj, _ = CoreSettings.objects.get_or_create(
|
|
key=DVR_SETTINGS_KEY,
|
|
defaults={"name": "DVR Settings", "value": {"series_rules": []}},
|
|
)
|
|
dirty_value = {**obj.value, "series_rules": "not a list"}
|
|
serializer = CoreSettingsSerializer(obj, data={"value": dirty_value}, partial=True)
|
|
serializer.is_valid(raise_exception=True)
|
|
serializer.save()
|
|
obj.refresh_from_db()
|
|
self.assertEqual(obj.value.get("series_rules"), [])
|
|
|
|
|
|
class EpgIgnoreListsTest(TestCase):
|
|
"""Verify EPG ignore list getters handle corrupted stored data."""
|
|
|
|
def _set_epg_field_raw(self, field, raw_value):
|
|
obj, _ = CoreSettings.objects.get_or_create(
|
|
key=EPG_SETTINGS_KEY,
|
|
defaults={"name": "EPG Settings", "value": {}},
|
|
)
|
|
current = obj.value if isinstance(obj.value, dict) else {}
|
|
current[field] = raw_value
|
|
obj.value = current
|
|
obj.save()
|
|
|
|
def test_valid_string_lists_returned(self):
|
|
for field, getter in [
|
|
("epg_match_ignore_prefixes", CoreSettings.get_epg_match_ignore_prefixes),
|
|
("epg_match_ignore_suffixes", CoreSettings.get_epg_match_ignore_suffixes),
|
|
("epg_match_ignore_custom", CoreSettings.get_epg_match_ignore_custom),
|
|
]:
|
|
self._set_epg_field_raw(field, ["HD", "SD"])
|
|
self.assertEqual(getter(), ["HD", "SD"])
|
|
|
|
def test_non_string_elements_filtered(self):
|
|
for field, getter in [
|
|
("epg_match_ignore_prefixes", CoreSettings.get_epg_match_ignore_prefixes),
|
|
("epg_match_ignore_suffixes", CoreSettings.get_epg_match_ignore_suffixes),
|
|
("epg_match_ignore_custom", CoreSettings.get_epg_match_ignore_custom),
|
|
]:
|
|
self._set_epg_field_raw(field, [42, None, "HD", True, "SD"])
|
|
result = getter()
|
|
self.assertEqual(result, ["HD", "SD"])
|
|
|
|
def test_non_list_value_returns_empty(self):
|
|
for field, getter in [
|
|
("epg_match_ignore_prefixes", CoreSettings.get_epg_match_ignore_prefixes),
|
|
("epg_match_ignore_suffixes", CoreSettings.get_epg_match_ignore_suffixes),
|
|
("epg_match_ignore_custom", CoreSettings.get_epg_match_ignore_custom),
|
|
]:
|
|
self._set_epg_field_raw(field, "not a list")
|
|
self.assertEqual(getter(), [])
|
|
|
|
|
|
class DropDBCommandTlsTest(TestCase):
|
|
"""Verify dropdb management command passes TLS parameters to psycopg."""
|
|
databases = []
|
|
|
|
_DB_WITH_TLS = {
|
|
'default': {
|
|
'ENGINE': 'django.db.backends.postgresql',
|
|
'NAME': 'testdb',
|
|
'USER': 'testuser',
|
|
'PASSWORD': 'testpass',
|
|
'HOST': 'localhost',
|
|
'PORT': 5432,
|
|
'OPTIONS': {
|
|
'sslmode': 'verify-full',
|
|
'sslrootcert': '/certs/ca.crt',
|
|
'sslcert': '/certs/client.crt',
|
|
'sslkey': '/certs/client.key',
|
|
},
|
|
}
|
|
}
|
|
|
|
_DB_NO_TLS = {
|
|
'default': {
|
|
'ENGINE': 'django.db.backends.postgresql',
|
|
'NAME': 'testdb',
|
|
'USER': 'testuser',
|
|
'PASSWORD': 'testpass',
|
|
'HOST': 'localhost',
|
|
'PORT': 5432,
|
|
}
|
|
}
|
|
|
|
@patch('core.management.commands.dropdb.psycopg.connect')
|
|
@patch('core.management.commands.dropdb.connection')
|
|
@patch('builtins.input', return_value='yes')
|
|
def test_dropdb_passes_ssl_kwargs_when_tls_enabled(self, _inp, _conn, mock_connect):
|
|
mock_pg = MagicMock()
|
|
mock_connect.return_value = mock_pg
|
|
mock_pg.cursor.return_value = MagicMock()
|
|
|
|
with self.settings(DATABASES=self._DB_WITH_TLS):
|
|
from django.core.management import call_command
|
|
call_command('dropdb')
|
|
|
|
mock_connect.assert_called_once_with(
|
|
dbname='postgres', user='testuser', password='testpass',
|
|
host='localhost', port=5432,
|
|
autocommit=True,
|
|
sslmode='verify-full',
|
|
sslrootcert='/certs/ca.crt',
|
|
sslcert='/certs/client.crt',
|
|
sslkey='/certs/client.key',
|
|
)
|
|
|
|
@patch('core.management.commands.dropdb.psycopg.connect')
|
|
@patch('core.management.commands.dropdb.connection')
|
|
@patch('builtins.input', return_value='yes')
|
|
def test_dropdb_no_ssl_kwargs_when_tls_disabled(self, _inp, _conn, mock_connect):
|
|
mock_pg = MagicMock()
|
|
mock_connect.return_value = mock_pg
|
|
mock_pg.cursor.return_value = MagicMock()
|
|
|
|
with self.settings(DATABASES=self._DB_NO_TLS):
|
|
from django.core.management import call_command
|
|
call_command('dropdb')
|
|
|
|
mock_connect.assert_called_once_with(
|
|
dbname='postgres', user='testuser', password='testpass',
|
|
host='localhost', port=5432,
|
|
autocommit=True,
|
|
)
|
|
|
|
|
|
class MallocTrimTests(SimpleTestCase):
|
|
def test_trim_is_noop_when_libc_has_no_malloc_trim(self):
|
|
from core.utils import trim_c_allocator_heap
|
|
|
|
fake_libc = MagicMock(spec=[])
|
|
with patch('ctypes.util.find_library', return_value='libc.so.6'), patch(
|
|
'ctypes.CDLL', return_value=fake_libc
|
|
):
|
|
self.assertFalse(trim_c_allocator_heap())
|