music-assistant-server

1"""Tests for the candidate selector: scoring, rejection filtering, and tie-breaking."""
2
3from __future__ import annotations
4
5import dataclasses
6import logging
7
8import numpy as np
9
10from music_assistant.controllers.streams.smart_fades.models import Deck, TransitionTier
11from music_assistant.controllers.streams.smart_fades.planner.candidates import Candidate
12from music_assistant.controllers.streams.smart_fades.planner.context import TransitionContext
13from music_assistant.controllers.streams.smart_fades.planner.policies import Policy, Verdict
14from music_assistant.controllers.streams.smart_fades.planner.selection import (
15    CandidateSelector,
16    ScoredCandidate,
17)
18from music_assistant.models.audio_analysis import AudioAnalysisData
19from tests.controllers.streams.smart_fades.conftest import build_test_candidate
20
21
22class _FixedPenaltyPolicy(Policy):
23    """A stub policy that returns one fixed, deterministic verdict for every candidate."""
24
25    def __init__(self, penalty: float = 0.0, rejected: bool = False, reason: str = "") -> None:
26        self._verdict = Verdict(penalty=penalty, rejected=rejected, reason=reason)
27
28    def evaluate(self, candidate: Candidate, ctx: TransitionContext) -> Verdict:
29        """Return this instance's fixed verdict, regardless of the candidate."""
30        return self._verdict
31
32
33class _BySourcePenaltyPolicy(Policy):
34    """A stub policy that maps ``spec.source`` to a fixed per-candidate penalty."""
35
36    def __init__(self, penalties: dict[str, float]) -> None:
37        self._penalties = penalties
38
39    def evaluate(self, candidate: Candidate, ctx: TransitionContext) -> Verdict:
40        """Return the configured penalty for this candidate's ``spec.source``."""
41        return Verdict.ok(self._penalties[candidate.spec.source])
42
43
44def _ctx() -> TransitionContext:
45    """Build a minimal TransitionContext; stub policies never read its fields."""
46    deck = Deck(
47        analysis=AudioAnalysisData(),
48        bpm=120.0,
49        beats=np.array([], dtype=np.float32),
50        downbeats=np.array([], dtype=np.float32),
51    )
52    return TransitionContext(
53        outgoing=deck,
54        incoming=deck,
55        outgoing_profile=None,
56        incoming_profile=None,
57        buffer_duration=45.0,
58        buffer_offset=0.0,
59        audio_end=45.0,
60        default_anchor=45.0,
61        mix_out_anchor=None,
62        kick_anchor=None,
63        fade_onset=None,
64        coda_zone=None,
65        tier=TransitionTier.FULL_BLEND,
66        cross_meter=False,
67        bpm_diff_percent=0.0,
68        vocal_out_placement=None,
69        vocal_in_placement=None,
70        vocal_out_scoring=None,
71        vocal_in_scoring=None,
72        natural_entry=0.0,
73        protective_downbeats=(),
74    )
75
76
77def _named(source: str) -> Candidate:
78    """Build a test candidate distinguishable only by its ``spec.source``."""
79    candidate = build_test_candidate()
80    return dataclasses.replace(candidate, spec=dataclasses.replace(candidate.spec, source=source))
81
82
83class TestCandidateSelector:
84    """CandidateSelector.select: full scoring, rejection filtering, tie-breaking."""
85
86    def test_lowest_penalty_wins(self) -> None:
87        """The survivor with the smallest total penalty is selected."""
88        low = _named("low")
89        high = _named("high")
90        selector = CandidateSelector(
91            policies=[_BySourcePenaltyPolicy({"low": 1.0, "high": 5.0})],
92            logger=logging.getLogger(__name__),
93        )
94
95        result = selector.select([high, low], _ctx())
96
97        assert result is not None
98        assert result.candidate is low
99        assert result.total_penalty == 1.0
100
101    def test_tie_breaks_to_first_in_input_order(self) -> None:
102        """Equal-penalty survivors resolve to whichever came first in the input sequence."""
103        first = _named("first")
104        second = _named("second")
105        selector = CandidateSelector(
106            policies=[_FixedPenaltyPolicy(penalty=3.0)],
107            logger=logging.getLogger(__name__),
108        )
109
110        result = selector.select([first, second], _ctx())
111
112        assert result is not None
113        assert result.candidate is first
114
115    def test_all_rejected_returns_none(self) -> None:
116        """When every candidate is rejected by some policy, select returns None."""
117        selector = CandidateSelector(
118            policies=[_FixedPenaltyPolicy(rejected=True, reason="always rejected")],
119            logger=logging.getLogger(__name__),
120        )
121
122        result = selector.select([_named("a"), _named("b")], _ctx())
123
124        assert result is None
125
126    def test_rejected_candidate_never_wins_even_with_lowest_penalty(self) -> None:
127        """A rejected candidate is excluded from ranking even if its penalty sum is lowest."""
128        rejected = _named("rejected")
129        survivor = _named("survivor")
130        penalty_policy = _BySourcePenaltyPolicy({"rejected": 0.0, "survivor": 100.0})
131
132        class _RejectByName(Policy):
133            def evaluate(self, candidate: Candidate, ctx: TransitionContext) -> Verdict:
134                if candidate.spec.source == "rejected":
135                    return Verdict.reject("rejected by name")
136                return Verdict.ok()
137
138        selector = CandidateSelector(
139            policies=[penalty_policy, _RejectByName()],
140            logger=logging.getLogger(__name__),
141        )
142
143        result = selector.select([rejected, survivor], _ctx())
144
145        assert result is not None
146        assert result.candidate is survivor
147
148    def test_empty_input_returns_none(self) -> None:
149        """An empty candidate sequence yields None."""
150        selector = CandidateSelector(
151            policies=[_FixedPenaltyPolicy()], logger=logging.getLogger(__name__)
152        )
153
154        result = selector.select([], _ctx())
155
156        assert result is None
157
158    def test_scored_candidate_carries_all_verdicts(self) -> None:
159        """The returned ScoredCandidate carries one verdict per policy, in evaluation order."""
160        policies = [_FixedPenaltyPolicy(penalty=1.0), _FixedPenaltyPolicy(penalty=2.0)]
161        selector = CandidateSelector(policies=policies, logger=logging.getLogger(__name__))
162
163        result = selector.select([_named("only")], _ctx())
164
165        assert result is not None
166        assert isinstance(result, ScoredCandidate)
167        assert len(result.verdicts) == len(policies)
168        assert result.total_penalty == 3.0
169