/
/
/
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