Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
82 changes: 80 additions & 2 deletions Lineuparr/plugin.py
Original file line number Diff line number Diff line change
Expand Up @@ -156,6 +156,13 @@ class PluginConfig:
DEFAULT_RATE_LIMITING = "none"
DEFAULT_CHANNEL_NUMBERING = "lineup"

# Sentinel category key _apply_category_detail uses for "None - single
# group" (Category Detail = "none"): every channel is merged under this one
# key, and _make_group_name special-cases it to mean "prefix only, no
# category suffix". Shared here so the two places that must agree on the
# exact string cannot drift apart.
SINGLE_GROUP_CATEGORY = "All"

RATE_LIMIT_NONE = 0.0
RATE_LIMIT_LOW = 0.1
RATE_LIMIT_MEDIUM = 0.5
Expand Down Expand Up @@ -845,7 +852,7 @@ def _apply_category_detail(self, data, detail):
all_channels = []
for channels in original_cats.values():
all_channels.extend(channels)
result["categories"] = {"All": all_channels}
result["categories"] = {PluginConfig.SINGLE_GROUP_CATEGORY: all_channels}
return result

if detail in ("simple", "refined"):
Expand Down Expand Up @@ -1037,7 +1044,22 @@ def _make_group_name(self, prefix, category):
"""Build full group name from prefix and category.
If prefix ends with a separator character, append category directly.
Otherwise add ': ' between prefix and category.
Examples: 'DTV-' + 'News' = 'DTV-News', 'DIRECTV' + 'News' = 'DIRECTV: News'"""
Examples: 'DTV-' + 'News' = 'DTV-News', 'DIRECTV' + 'News' = 'DIRECTV: News'

Category Detail = "none" ("single group (prefix only, no categories)")
is the one exception: it merges every channel under the sentinel
category PluginConfig.SINGLE_GROUP_CATEGORY, and the resulting group
name must be the prefix ALONE, not "<prefix>: All" - the setting's own
description promises "no categories". Every caller reaches this
sentinel the same way (it flows out of _apply_category_detail via
lineup["categories"]), so handling it here is the single place that
keeps every one of this method's callers in agreement. A real lineup
category that happens to be named exactly "All" under Normal/Simple/
Refined detail collapses the same way; that is an accepted, harmless
edge case rather than something worth a second sentinel value."""
if category == PluginConfig.SINGLE_GROUP_CATEGORY:
stripped_prefix = prefix.rstrip(" :_-/") if prefix else ""
return stripped_prefix or PluginConfig.SINGLE_GROUP_CATEGORY
if prefix:
if prefix[-1] in ":_-/ ":
return f"{prefix}{category}"
Expand Down Expand Up @@ -2727,13 +2749,59 @@ def _do_preview_stream_match(self, settings, logger):
# DESTRUCTIVE ACTIONS
# ========================================================================

def _migrate_single_group_name(self, prefix, dry_run, logger):
"""Rename a pre-fix "<prefix>: All" group to the new prefix-only name.

Before this fix, Category Detail = "none" built its group name the
same way every other category did, giving "<prefix>: All" instead of
the "prefix only" the setting promises. Someone already using this
mode has that old group in Dispatcharr; without this step, the next
sync would create a second, new "<prefix>" group under the fixed
name and leave the old one behind as an orphaned duplicate holding
their existing channels.

Renaming (not recreating) preserves the group's id, so every channel
already assigned to it stays assigned - only its name changes. If a
"<prefix>" group already exists too, merging the two would mean
moving channels between groups, which is a bigger and riskier
decision than this fix should make silently, so both are left alone
and a warning is logged instead.
"""
old_name = f"{prefix}: All" if prefix and prefix[-1] not in ":_-/ " else f"{prefix}All" if prefix else "All"
new_name = self._make_group_name(prefix, PluginConfig.SINGLE_GROUP_CATEGORY)
if old_name == new_name:
return # no prefix - "All" both before and after the fix

old_group = ChannelGroup.objects.filter(name=old_name).first()
if not old_group:
return # nothing to migrate

if ChannelGroup.objects.filter(name=new_name).exists():
logger.warning(
f"{LOG_PREFIX} Both '{old_name}' and '{new_name}' channel groups exist; "
f"leaving both as-is. Merge them manually if '{old_name}' is a leftover "
f"from before this fix."
)
return

if dry_run:
logger.info(f"{LOG_PREFIX} [DRY RUN] Would rename group '{old_name}' to '{new_name}'")
return

old_group.name = new_name
old_group.save(update_fields=["name"])
logger.info(f"{LOG_PREFIX} Renamed group '{old_name}' to '{new_name}' (ID: {old_group.id})")

def _sync_groups(self, settings, logger):
"""Create/update ChannelGroups from lineup categories."""
dry_run = settings.get("dry_run_mode", False)
lineup = self._load_lineup(settings, logger)
prefix = self._get_group_prefix(settings, lineup)
rate_limiter = SmartRateLimiter(settings.get("rate_limiting", PluginConfig.DEFAULT_RATE_LIMITING))

if PluginConfig.SINGLE_GROUP_CATEGORY in lineup["categories"]:
self._migrate_single_group_name(prefix, dry_run, logger)

created = 0
existed = 0

Expand Down Expand Up @@ -2817,6 +2885,16 @@ def _do_sync_channels(self, settings, logger):
rate_limiter = SmartRateLimiter(settings.get("rate_limiting", PluginConfig.DEFAULT_RATE_LIMITING))
assigner = self._init_assigner_state(settings)

# Sync Channels builds its own group lookup below rather than going
# through _sync_groups, so it needs this same migration call: without
# it, a lineup using Category Detail = "none" whose only prior sync
# was ever via this button (there is no Sync Groups button in the
# UI) would never see its pre-fix "<prefix>: All" group renamed, and
# get_or_create just below would create a second "<prefix>" group
# instead, orphaning the channels already on the old one.
if PluginConfig.SINGLE_GROUP_CATEGORY in lineup["categories"]:
self._migrate_single_group_name(prefix, dry_run, logger)

# Ensure groups exist
existing_groups = {g['name']: g['id'] for g in ChannelGroup.objects.all().values('id', 'name')}
for category in lineup["categories"]:
Expand Down
235 changes: 235 additions & 0 deletions tests/test_group_prefix_only.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,235 @@
"""Regression for GitHub issue #30: Category Detail "None - single group
(prefix only, no categories)" created "<prefix>: All" instead of a
prefix-only group name.

Root cause: _apply_category_detail("none") merges every channel under the
sentinel category "All", and _make_group_name treated "All" like any other
category, appending it after the prefix the same way it appends "News" or
"Movies". The setting's own description promises "prefix only, no
categories", so the "All" suffix was a bug.

The fix (in _make_group_name, the single place every caller goes through)
special-cases the sentinel category so the result is the prefix alone. It
also has to handle two things without a setting or a code change from the
maintainer's review of issue #30:

1. Someone already running this mode has a "<prefix>: All" group in
Dispatcharr. _migrate_single_group_name renames that existing group
(preserving its id and channel assignments) instead of leaving it as an
orphaned duplicate next to a freshly created "<prefix>" group - unless a
"<prefix>" group already exists too, in which case both are left alone
and a warning is logged (merging channels between groups is out of
scope for a silent migration).
2. A prefix that already ends in a separator (e.g. "DTV-") must not leave
that separator dangling ("DTV-" not "DTV-All" -> "DTV", not "DTV-").
"""
import logging

import pytest

from Lineuparr import plugin as plugin_module
from Lineuparr.plugin import Plugin, PluginConfig


@pytest.fixture
def plugin():
return Plugin.__new__(Plugin)


@pytest.fixture
def logger():
return logging.getLogger("test")


class TestApplyCategoryDetailUsesSharedSentinel:
def test_none_detail_merges_under_the_shared_sentinel(self, plugin):
data = {"categories": {"News": [{"name": "CNN"}], "Sports": [{"name": "ESPN"}]}}
result = plugin._apply_category_detail(data, "none")
assert list(result["categories"].keys()) == [PluginConfig.SINGLE_GROUP_CATEGORY]
assert len(result["categories"][PluginConfig.SINGLE_GROUP_CATEGORY]) == 2


class TestMakeGroupNameSingleGroupMode:
@pytest.mark.parametrize("prefix,expected", [
("Spectrum", "Spectrum"),
("DIRECTV", "DIRECTV"),
("", "All"),
("DTV-", "DTV"),
("DTV: ", "DTV"),
("DTV ", "DTV"),
("DTV_", "DTV"),
("DTV/", "DTV"),
("---", "All"), # prefix is nothing but separators - falls back
])
def test_single_group_category_is_prefix_only(self, plugin, prefix, expected):
assert plugin._make_group_name(prefix, PluginConfig.SINGLE_GROUP_CATEGORY) == expected


class TestMakeGroupNameOtherDetailModesUnchanged:
"""The fix must not touch Normal/Simple/Refined category naming."""

@pytest.mark.parametrize("prefix,category,expected", [
("Spectrum", "Movies", "Spectrum: Movies"),
("Spectrum", "News", "Spectrum: News"),
("DTV-", "News", "DTV-News"),
("", "News", "News"),
])
def test_normal_categories_unchanged(self, plugin, prefix, category, expected):
assert plugin._make_group_name(prefix, category) == expected

def test_a_real_category_literally_named_all_also_collapses(self, plugin):
# Accepted edge case, documented on _make_group_name: a genuine
# lineup category named exactly "All" collapses the same way the
# sentinel does. Harmless (it already reads as one merged group).
assert plugin._make_group_name("Spectrum", "All") == "Spectrum"


# --- Migration: renaming a pre-fix "<prefix>: All" group -------------------

class _FakeGroup:
def __init__(self, name, id):
self.name = name
self.id = id
self.saved = False

def save(self, update_fields=None):
self.saved = True


class _FakeQuerySet:
def __init__(self, items):
self._items = items

def first(self):
return self._items[0] if self._items else None

def exists(self):
return bool(self._items)

def values(self, *fields):
return [{f: getattr(g, f) for f in fields} for g in self._items]


class _FakeManager:
def __init__(self, groups):
self._groups = groups

def filter(self, name=None):
return _FakeQuerySet([g for g in self._groups if g.name == name])

def all(self):
return _FakeQuerySet(list(self._groups))

def get_or_create(self, name, defaults=None):
for g in self._groups:
if g.name == name:
return g, False
new_group = _FakeGroup(name, id=max((g.id for g in self._groups), default=0) + 1)
self._groups.append(new_group)
return new_group, True


class _FakeChannelGroup:
def __init__(self, groups):
self.objects = _FakeManager(groups)


class TestMigrateSingleGroupName:
def test_renames_old_group_when_new_name_is_free(self, plugin, logger, monkeypatch):
old = _FakeGroup("Spectrum: All", id=7)
fake = _FakeChannelGroup([old])
monkeypatch.setattr(plugin_module, "ChannelGroup", fake)

plugin._migrate_single_group_name("Spectrum", dry_run=False, logger=logger)

assert old.name == "Spectrum"
assert old.saved is True

def test_leaves_both_alone_and_warns_when_new_name_already_exists(self, plugin, caplog, monkeypatch):
old = _FakeGroup("Spectrum: All", id=7)
new = _FakeGroup("Spectrum", id=9)
fake = _FakeChannelGroup([old, new])
monkeypatch.setattr(plugin_module, "ChannelGroup", fake)

with caplog.at_level(logging.WARNING):
plugin._migrate_single_group_name("Spectrum", dry_run=False, logger=logging.getLogger("test"))

assert old.name == "Spectrum: All" # unchanged
assert old.saved is False
assert any("Spectrum: All" in r.message and "Spectrum" in r.message for r in caplog.records)

def test_dry_run_does_not_rename(self, plugin, logger, monkeypatch):
old = _FakeGroup("Spectrum: All", id=7)
fake = _FakeChannelGroup([old])
monkeypatch.setattr(plugin_module, "ChannelGroup", fake)

plugin._migrate_single_group_name("Spectrum", dry_run=True, logger=logger)

assert old.name == "Spectrum: All"
assert old.saved is False

def test_no_old_group_is_a_silent_no_op(self, plugin, logger, monkeypatch):
fake = _FakeChannelGroup([])
monkeypatch.setattr(plugin_module, "ChannelGroup", fake)

# Must not raise.
plugin._migrate_single_group_name("Spectrum", dry_run=False, logger=logger)

def test_no_prefix_is_a_no_op_without_touching_the_database(self, plugin, logger, monkeypatch):
# Both old and new names are "All" with no prefix - nothing to migrate,
# and the method should return before ever calling ChannelGroup.objects.
class _ExplodingManager:
def filter(self, **kwargs):
raise AssertionError("should not query the database")

class _ExplodingChannelGroup:
objects = _ExplodingManager()

monkeypatch.setattr(plugin_module, "ChannelGroup", _ExplodingChannelGroup)

plugin._migrate_single_group_name("", dry_run=False, logger=logger)

def test_renames_with_a_separator_terminated_prefix(self, plugin, logger, monkeypatch):
old = _FakeGroup("DTV-All", id=3)
fake = _FakeChannelGroup([old])
monkeypatch.setattr(plugin_module, "ChannelGroup", fake)

plugin._migrate_single_group_name("DTV-", dry_run=False, logger=logger)

assert old.name == "DTV"
assert old.saved is True


class TestSyncChannelsAlsoMigrates:
"""Regression from code review on this PR: the migration must run from
the Sync Channels button too, not only Full Sync. There is no Sync
Groups button in the UI, so Sync Channels (_do_sync_channels) is, in
practice, the only path some installs ever take. Before this fix it
built its own group lookup independently of _sync_groups and never
called the migration, so pressing Sync Channels against an existing
"<prefix>: All" group created a second "<prefix>" group instead of
renaming the first - leaving the old group's channels orphaned.
"""

def test_sync_channels_renames_the_old_group_instead_of_duplicating_it(
self, plugin, logger, monkeypatch
):
old = _FakeGroup("Spectrum: All", id=7)
fake = _FakeChannelGroup([old])
monkeypatch.setattr(plugin_module, "ChannelGroup", fake)

# No channels in the lineup: this test is about which group exists
# afterward, not channel assignment, so Channel.objects is never
# touched and does not need its own fake.
monkeypatch.setattr(
plugin, "_load_lineup",
lambda settings, logger: {"categories": {PluginConfig.SINGLE_GROUP_CATEGORY: []}},
)
monkeypatch.setattr(plugin, "_get_group_prefix", lambda settings, lineup: "Spectrum")
plugin._suppress_refresh = False # Plugin.__new__ skips __init__

plugin._do_sync_channels({}, logger)

names = [g.name for g in fake.objects._groups]
assert names == ["Spectrum"], f"expected the old group renamed in place, got {names}"
assert old.saved is True
Loading