diff --git a/README.md b/README.md index ef7137b..bed8435 100644 --- a/README.md +++ b/README.md @@ -58,6 +58,9 @@ cusp stream -d "USB Audio" -t "Living Room" # From the system audio output on Linux (see "System audio" below) cusp stream -d system -t "Living Room" +# Stream to an AirPlay 2 group (see "AirPlay 2 groups" below) +cusp stream -d "USB Audio" -t "Living Room,Kitchen,Office" + # Or use a config file cusp stream --config cusp.toml ``` @@ -66,7 +69,7 @@ The `-d` flag accepts: - a device name (substring match) or index number from `cusp devices` - the literal `system` to capture system audio output (Linux only) -The `-t` flag accepts an AirPlay receiver name. +The `-t` flag accepts an AirPlay receiver name, or a comma-separated list for a group. ### Pair with a device @@ -144,6 +147,27 @@ log_level = "INFO" Config file search order: `--config` flag, then `./cusp.toml`, then `~/.config/cusp/cusp.toml`. Command line arguments override config file values. +## AirPlay 2 groups + +cusp can fan a single capture out to multiple AirPlay receivers at once. Pass a comma-separated list on the CLI, or a TOML array in the config: + +```bash +cusp stream -d "USB Audio" -t "Living Room,Kitchen,Office" +``` + +```toml +[airplay] +target = ["Living Room", "Kitchen", "Office"] +``` + +Single-target usage is unchanged — `-t "Living Room"` or `target = "Living Room"` still works and connects to one device. + +**Leader semantics.** The first name in the list is the group leader; the rest are followers. The leader is what cusp retries against if no device can be reached at startup, and its name is used for logging and for looking up stored pairing credentials. + +**Graceful degradation.** If a follower can't be resolved on the network or its initial connection fails, cusp logs a warning and keeps streaming to whichever devices did connect. If a follower drops mid-stream, the same applies — the rest of the group keeps playing. The session only fails (and triggers `auto_reconnect`) when every device has dropped. You can leave an occasionally-offline speaker in the list without breaking playback for the rest of the house. + +**Sync drift.** Each receiver gets its own RAOP session — there is no shared timing anchor between them. Receivers in the same room will drift enough to be audible over time (on the order of tens of milliseconds within a few minutes). This is a known tradeoff of running an application-level fan-out instead of a true Apple-managed AirPlay 2 group. If you need tightly synced playback in a single room, group the speakers in the Home app and target the group leader's name from cusp instead. + ## Running on a Raspberry Pi ### Systemd service diff --git a/cusp.toml.example b/cusp.toml.example index 939e6d5..670e35f 100644 --- a/cusp.toml.example +++ b/cusp.toml.example @@ -9,9 +9,17 @@ device = "USB Audio" blocksize = 1024 [airplay] -# AirPlay receiver name (substring match) -# Run `cusp devices` to see available receivers +# AirPlay receiver name (substring match). +# Run `cusp devices` to see available receivers. target = "Living Room" +# To stream to a group, pass a list of names; the first is the leader +# and the rest are followers. If a follower can't be reached, it is +# skipped with a warning and the stream continues to the survivors. +# Note: each receiver gets its own RAOP session, so playback will drift +# between them over time (audible within a few minutes in the same room). +# For tightly synced playback, group the speakers in the Home app and +# target the group leader's name here. +# target = ["Living Room", "Kitchen", "Office"] # password = "secret" [behavior] diff --git a/src/cusp/airplay.py b/src/cusp/airplay.py index bbb2d16..7a0f941 100644 --- a/src/cusp/airplay.py +++ b/src/cusp/airplay.py @@ -82,36 +82,68 @@ async def pair_device(name: str) -> None: raise RuntimeError(f"No RAOP service found on {target.name}") -async def resolve_target(config: CuspConfig) -> BaseConfig: - """Scan for the configured AirPlay target and apply credentials/password. +def _apply_auth(target: BaseConfig, config: CuspConfig) -> None: + """Attach stored RAOP credentials and the configured password to a target.""" + creds = _load_credentials() + stored = creds.get(str(target.identifier)) + for service in target.services: + if service.protocol != Protocol.RAOP: + continue + if stored and "raop" in stored: + service.credentials = stored["raop"] + if config.airplay_password: + service.password = config.airplay_password + - Returns a pyatv device config ready to be passed to `connect_target`. - Raises ConnectionError if the device is not found on the network. +async def resolve_targets(config: CuspConfig) -> list[BaseConfig]: + """Resolve all configured AirPlay targets in a single network scan. + + Returns the surviving devices in configured order — the first entry is the + group leader. Followers that are missing or ambiguous are logged and + skipped so one offline speaker cannot kill the session. An ambiguous + *leader* still fails hard (surfaces as ValueError from `_find_device`). + + Raises ConnectionError with the list of missed names if zero devices + resolve. """ + if not config.airplay_target: + raise ConnectionError("No AirPlay target configured") + loop = asyncio.get_event_loop() devices = await pyatv.scan(loop, timeout=5) - target = _find_device(devices, config.airplay_target) - if target is None: + + resolved: list[BaseConfig] = [] + missed: list[str] = [] + for index, name in enumerate(config.airplay_target): + is_leader = index == 0 + try: + target = _find_device(devices, name) + except ValueError: + if is_leader: + raise + logger.warning("AirPlay follower '%s' ambiguous, skipping", name) + missed.append(name) + continue + if target is None: + role = "leader" if is_leader else "follower" + logger.warning("AirPlay %s '%s' not found, skipping", role, name) + missed.append(name) + continue + _apply_auth(target, config) + resolved.append(target) + + if not resolved: raise ConnectionError( - f"AirPlay device '{config.airplay_target}' not found on network. " + f"No AirPlay devices found on network (tried: {', '.join(missed)}). " "Run `cusp devices` to see available receivers." ) - # Apply stored credentials - creds = _load_credentials() - if str(target.identifier) in creds: - stored = creds[str(target.identifier)] - for service in target.services: - if service.protocol == Protocol.RAOP and "raop" in stored: - service.credentials = stored["raop"] - - # Apply password if configured - if config.airplay_password: - for service in target.services: - if service.protocol == Protocol.RAOP: - service.password = config.airplay_password - - return target + return resolved + + +async def resolve_target(config: CuspConfig) -> BaseConfig: + """Resolve the configured leader. Thin wrapper over `resolve_targets`.""" + return (await resolve_targets(config))[0] async def connect_target(target: BaseConfig) -> AppleTV: diff --git a/src/cusp/cli.py b/src/cusp/cli.py index 0370716..4ed09c5 100644 --- a/src/cusp/cli.py +++ b/src/cusp/cli.py @@ -70,7 +70,15 @@ async def _pair_device(name: str) -> None: default=None, help='Audio input device name, index, or "system" to capture system audio.', ) -@click.option("-t", "--target", default=None, help="AirPlay receiver name.") +@click.option( + "-t", + "--target", + default=None, + help=( + "AirPlay receiver name. Pass a comma-separated list to stream to a " + "group (first name is the leader)." + ), +) @click.option( "-c", "--config", @@ -116,10 +124,15 @@ def stream( else: audio_device = device + # Split -t on commas into a group list; first entry is the leader. + airplay_target: list[str] | None = None + if target is not None: + airplay_target = [t.strip() for t in target.split(",") if t.strip()] + config = load_config( config_path=config_path, audio_device=audio_device, - airplay_target=target, + airplay_target=airplay_target, sample_rate=sample_rate, channels=channels, blocksize=blocksize, diff --git a/src/cusp/config.py b/src/cusp/config.py index a91bfdc..1264171 100644 --- a/src/cusp/config.py +++ b/src/cusp/config.py @@ -18,8 +18,8 @@ class CuspConfig: channels: int | None = None blocksize: int = 1024 - # AirPlay target - airplay_target: str | None = None + # AirPlay target(s). First entry is the group leader; remaining are followers. + airplay_target: list[str] | None = None airplay_password: str | None = None # Behavior @@ -32,6 +32,24 @@ class CuspConfig: log_file: str | None = None +def _normalize_targets(value: object) -> list[str]: + """Normalize a TOML `airplay.target` value to a list. + + Accepts a TOML array (`["A", "B"]`) or a string; strings are split on + commas to match the CLI `-t` form, so `"A, B"` yields `["A", "B"]`. + """ + if isinstance(value, str): + items = value.split(",") + elif isinstance(value, list): + items = [str(v) for v in value] + else: + raise TypeError( + "airplay.target must be a string or array of strings, " + f"got {type(value).__name__}" + ) + return [s.strip() for s in items if s.strip()] + + def _find_config_file() -> Path | None: candidates = [ Path("cusp.toml"), @@ -68,7 +86,7 @@ def load_config(config_path: str | None = None, **cli_overrides: object) -> Cusp if "blocksize" in audio: data["blocksize"] = audio["blocksize"] if "target" in airplay: - data["airplay_target"] = airplay["target"] + data["airplay_target"] = _normalize_targets(airplay["target"]) if "password" in airplay: data["airplay_password"] = airplay["password"] for key in ( diff --git a/src/cusp/pipeline.py b/src/cusp/pipeline.py index 3e465d3..f28e57c 100644 --- a/src/cusp/pipeline.py +++ b/src/cusp/pipeline.py @@ -9,7 +9,7 @@ import numpy as np -from cusp.airplay import connect_target, resolve_target +from cusp.airplay import connect_target, resolve_targets from cusp.audio import make_capture if TYPE_CHECKING: @@ -52,9 +52,10 @@ def _wav_header(sample_rate: int, channels: int, bits_per_sample: int = 16) -> b class StreamingSession: """Owns one AirPlay connection + reader + consumer task.""" - def __init__(self, atv: AppleTV, config: CuspConfig) -> None: + def __init__(self, atv: AppleTV, config: CuspConfig, name: str = "") -> None: self._atv = atv self._config = config + self.name = name self._reader = asyncio.StreamReader(limit=2**20) # 1MB buffer # Write a WAV header so pyatv can identify the audio format immediately. # Raw PCM data follows directly — no MP3 encode/decode round-trip needed. @@ -64,7 +65,7 @@ def __init__(self, atv: AppleTV, config: CuspConfig) -> None: @classmethod async def start(cls, target: BaseConfig, config: CuspConfig) -> StreamingSession: atv = await connect_target(target) - return cls(atv, config) + return cls(atv, config, name=target.name) async def _consume(self) -> None: """Stream audio data to the AirPlay receiver via pyatv.""" @@ -84,17 +85,18 @@ def exception(self) -> BaseException | None: return None async def stop(self) -> None: - logger.info("Disconnecting from AirPlay receiver") + label = self.name or "AirPlay receiver" + logger.info("Disconnecting from %s", label) self._reader.feed_eof() try: await asyncio.wait_for(self._consumer_task, timeout=5.0) except asyncio.TimeoutError: - logger.warning("Consumer task did not finish within 5s, cancelling") + logger.warning("Consumer task for %s did not finish within 5s", label) self._consumer_task.cancel() with contextlib.suppress(BaseException): await self._consumer_task except Exception as e: - logger.warning("Consumer task finished with error: %s", e) + logger.warning("Consumer task for %s finished with error: %s", label, e) finally: # pyatv's close() returns a set of cleanup tasks (RAOP teardown, # zeroconf unregister, etc). They MUST be awaited or the receiver @@ -107,8 +109,88 @@ async def stop(self) -> None: timeout=5.0, ) except asyncio.TimeoutError: - logger.warning("pyatv close did not finish within 5s") - logger.info("AirPlay receiver disconnected") + logger.warning("pyatv close for %s did not finish within 5s", label) + logger.info("%s disconnected", label) + + +class GroupStreamingSession: + """Fan out one capture stream to N AirPlay receivers. + + Owns a list of independent `StreamingSession`s (one pyatv connection + + StreamReader + consumer task per receiver). A follower dropping mid-stream + logs a warning but does not affect siblings; the group as a whole only + surfaces failure once every sub-session has died, which lets the caller's + reconnect logic run on total loss. + """ + + def __init__(self, sessions: list[StreamingSession]) -> None: + self._sessions = sessions + self._logged_drops: set[int] = set() + + @classmethod + async def start( + cls, targets: list[BaseConfig], config: CuspConfig + ) -> GroupStreamingSession: + """Connect all targets in parallel; keep whichever succeed.""" + results = await asyncio.gather( + *(connect_target(t) for t in targets), + return_exceptions=True, + ) + sessions: list[StreamingSession] = [] + for target, result in zip(targets, results): + if isinstance(result, BaseException): + logger.warning( + "Failed to connect to AirPlay target '%s': %s", + target.name, + result, + ) + continue + sessions.append(StreamingSession(result, config, name=target.name)) + if not sessions: + raise ConnectionError("Failed to connect to any configured AirPlay target") + logger.info( + "AirPlay group streaming to %d/%d device(s)", len(sessions), len(targets) + ) + return cls(sessions) + + def feed(self, pcm_chunk: bytes) -> None: + """Write `pcm_chunk` to every live sub-reader; skip any that have died.""" + for session in self._sessions: + if session.failed(): + key = id(session) + if key not in self._logged_drops: + self._logged_drops.add(key) + remaining = sum(1 for s in self._sessions if not s.failed()) + logger.warning( + "AirPlay receiver '%s' dropped: %s; " + "%d device(s) still streaming", + session.name, + session.exception(), + remaining, + ) + continue + session.feed(pcm_chunk) + + def failed(self) -> bool: + """True only when every sub-session has failed.""" + return all(s.failed() for s in self._sessions) + + def exception(self) -> BaseException | None: + """First sub-session exception — only returned when the whole group is dead.""" + if not self.failed(): + return None + for s in self._sessions: + exc = s.exception() + if exc is not None: + return exc + return None + + async def stop(self) -> None: + """Tear down every sub-session in parallel.""" + await asyncio.gather( + *(s.stop() for s in self._sessions), + return_exceptions=True, + ) async def run_pipeline(config: CuspConfig) -> None: @@ -121,12 +203,15 @@ async def run_pipeline(config: CuspConfig) -> None: loop = asyncio.get_event_loop() # One-time scan at startup; refreshed periodically while idle. - target: BaseConfig = await resolve_target(config) + targets: list[BaseConfig] = await resolve_targets(config) + # Guards swaps of `targets` so a concurrent reader (session start) never + # sees a torn list during a background refresh. + targets_lock = asyncio.Lock() capture = make_capture(config, loop) await capture.start() - session: StreamingSession | None = None + session: GroupStreamingSession | None = None last_activity: float | None = None threshold_sq = config.silence_threshold**2 @@ -148,17 +233,20 @@ def _handle_signal(sig: int) -> None: loop.add_signal_handler(sig, _handle_signal, sig) async def refresh_target_loop() -> None: - """Periodically re-scan for the AirPlay target while idle.""" - nonlocal target + """Periodically re-scan for AirPlay targets while idle.""" + nonlocal targets while True: await asyncio.sleep(config.target_refresh_interval) if session is not None: continue # don't re-scan while a session is live try: - target = await resolve_target(config) - logger.debug("Refreshed AirPlay target") + new_targets = await resolve_targets(config) except ConnectionError as e: logger.warning("Background target refresh failed: %s", e) + continue + async with targets_lock: + targets = new_targets + logger.debug("Refreshed AirPlay targets (%d)", len(new_targets)) refresh_task = asyncio.create_task(refresh_target_loop()) @@ -189,13 +277,19 @@ async def refresh_target_loop() -> None: # seconds; pause the capture so the queue doesn't saturate # and so we resume from real time, not from a backlog. with capture.paused(): + async with targets_lock: + current_targets = list(targets) try: - session = await StreamingSession.start(target, config) + session = await GroupStreamingSession.start( + current_targets, config + ) except ConnectionError: - # Cached target stale — re-resolve once and retry. - logger.info("Cached target stale, re-scanning") - target = await resolve_target(config) - session = await StreamingSession.start(target, config) + # Cached targets stale — re-resolve once and retry. + logger.info("Cached targets stale, re-scanning") + new_targets = await resolve_targets(config) + async with targets_lock: + targets = new_targets + session = await GroupStreamingSession.start(new_targets, config) continue # discard the in-hand (now stale) chunk if session is not None: diff --git a/tests/test_airplay.py b/tests/test_airplay.py index 1a64364..7d60fae 100644 --- a/tests/test_airplay.py +++ b/tests/test_airplay.py @@ -1,9 +1,13 @@ import sys from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock import pytest +from pyatv.const import Protocol from cusp import airplay +from cusp.config import CuspConfig @pytest.mark.skipif(sys.platform == "win32", reason="POSIX file modes") @@ -36,3 +40,153 @@ def test_save_credentials_roundtrip(monkeypatch, tmp_path): airplay._save_credentials(data) assert airplay._load_credentials() == data + + +def _fake_device(name: str, identifier: str | None = None, has_raop: bool = True): + """Build a scan-result double with the attributes airplay helpers touch.""" + services = [] + if has_raop: + services.append( + SimpleNamespace(protocol=Protocol.RAOP, credentials=None, password=None) + ) + return SimpleNamespace( + name=name, + identifier=identifier or name.lower().replace(" ", "-"), + services=services, + ) + + +@pytest.fixture +def fake_scan(monkeypatch): + """Patch pyatv.scan to return a caller-supplied device list.""" + + def _install(devices): + monkeypatch.setattr(airplay.pyatv, "scan", AsyncMock(return_value=devices)) + + return _install + + +class TestResolveTargets: + async def test_leader_only(self, fake_scan, monkeypatch): + monkeypatch.setattr(airplay, "_load_credentials", lambda: {}) + fake_scan([_fake_device("Kitchen"), _fake_device("Office")]) + config = CuspConfig(airplay_target=["Kitchen"]) + + resolved = await airplay.resolve_targets(config) + + assert [d.name for d in resolved] == ["Kitchen"] + + async def test_all_resolved_preserves_order(self, fake_scan, monkeypatch): + monkeypatch.setattr(airplay, "_load_credentials", lambda: {}) + # Scan order deliberately differs from configured order to prove + # we return the configured order (leader first). + fake_scan( + [ + _fake_device("Office"), + _fake_device("Kitchen"), + _fake_device("Bedroom"), + ] + ) + config = CuspConfig(airplay_target=["Kitchen", "Office", "Bedroom"]) + + resolved = await airplay.resolve_targets(config) + + assert [d.name for d in resolved] == ["Kitchen", "Office", "Bedroom"] + + async def test_partial_failure_skips_missing_follower( + self, fake_scan, monkeypatch, caplog + ): + monkeypatch.setattr(airplay, "_load_credentials", lambda: {}) + fake_scan([_fake_device("Kitchen"), _fake_device("Bedroom")]) + config = CuspConfig(airplay_target=["Kitchen", "Office", "Bedroom"]) + + with caplog.at_level("WARNING", logger=airplay.logger.name): + resolved = await airplay.resolve_targets(config) + + assert [d.name for d in resolved] == ["Kitchen", "Bedroom"] + assert any( + "Office" in r.message and "not found" in r.message for r in caplog.records + ) + + async def test_all_missing_raises(self, fake_scan, monkeypatch): + monkeypatch.setattr(airplay, "_load_credentials", lambda: {}) + fake_scan([_fake_device("Bedroom")]) + config = CuspConfig(airplay_target=["Kitchen", "Office"]) + + with pytest.raises(ConnectionError) as excinfo: + await airplay.resolve_targets(config) + + # All missed names are surfaced in the error for easier debugging. + assert "Kitchen" in str(excinfo.value) + assert "Office" in str(excinfo.value) + + async def test_empty_config_raises(self, fake_scan, monkeypatch): + monkeypatch.setattr(airplay, "_load_credentials", lambda: {}) + fake_scan([_fake_device("Kitchen")]) + config = CuspConfig(airplay_target=None) + + with pytest.raises(ConnectionError): + await airplay.resolve_targets(config) + + async def test_ambiguous_leader_raises(self, fake_scan, monkeypatch): + monkeypatch.setattr(airplay, "_load_credentials", lambda: {}) + fake_scan([_fake_device("Kitchen", "a"), _fake_device("Kitchen", "b")]) + config = CuspConfig(airplay_target=["Kitchen"]) + + with pytest.raises(ValueError, match="Ambiguous"): + await airplay.resolve_targets(config) + + async def test_ambiguous_follower_is_skipped(self, fake_scan, monkeypatch, caplog): + monkeypatch.setattr(airplay, "_load_credentials", lambda: {}) + fake_scan( + [ + _fake_device("Kitchen"), + _fake_device("Office", "a"), + _fake_device("Office", "b"), + ] + ) + config = CuspConfig(airplay_target=["Kitchen", "Office"]) + + with caplog.at_level("WARNING", logger=airplay.logger.name): + resolved = await airplay.resolve_targets(config) + + assert [d.name for d in resolved] == ["Kitchen"] + assert any( + "Office" in r.message and "ambiguous" in r.message for r in caplog.records + ) + + async def test_applies_stored_credentials_and_password( + self, fake_scan, monkeypatch + ): + monkeypatch.setattr( + airplay, + "_load_credentials", + lambda: {"kitchen": {"name": "Kitchen", "raop": "token-k"}}, + ) + fake_scan([_fake_device("Kitchen"), _fake_device("Office")]) + config = CuspConfig(airplay_target=["Kitchen", "Office"], airplay_password="pw") + + resolved = await airplay.resolve_targets(config) + + kitchen_svc = next( + s for s in resolved[0].services if s.protocol == Protocol.RAOP + ) + office_svc = next( + s for s in resolved[1].services if s.protocol == Protocol.RAOP + ) + assert kitchen_svc.credentials == "token-k" + assert kitchen_svc.password == "pw" + # No creds stored for Office — only password applied. + assert office_svc.credentials is None + assert office_svc.password == "pw" + + +class TestResolveTargetWrapper: + async def test_returns_leader(self, fake_scan, monkeypatch): + monkeypatch.setattr(airplay, "_load_credentials", lambda: {}) + fake_scan([_fake_device("Kitchen"), _fake_device("Office")]) + config = CuspConfig(airplay_target=["Kitchen", "Office"]) + + target = await airplay.resolve_target(config) + + assert target.name == "Kitchen" diff --git a/tests/test_config.py b/tests/test_config.py index c84f063..cd6056e 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,5 +1,9 @@ from pathlib import Path +from unittest.mock import patch +from click.testing import CliRunner + +from cusp.cli import main from cusp.config import CuspConfig, _find_config_file, load_config @@ -26,7 +30,7 @@ def test_custom_values(self): sample_rate=44100, channels=1, blocksize=512, - airplay_target="Kitchen", + airplay_target=["Kitchen"], airplay_password="secret", auto_reconnect=False, reconnect_delay=10.0, @@ -40,7 +44,7 @@ def test_custom_values(self): assert cfg.sample_rate == 44100 assert cfg.channels == 1 assert cfg.blocksize == 512 - assert cfg.airplay_target == "Kitchen" + assert cfg.airplay_target == ["Kitchen"] assert cfg.airplay_password == "secret" assert cfg.auto_reconnect is False assert cfg.reconnect_delay == 10.0 @@ -68,7 +72,7 @@ def test_full_toml(self, tmp_path): assert cfg.sample_rate == 44100 assert cfg.channels == 1 assert cfg.blocksize == 512 - assert cfg.airplay_target == "Living Room" + assert cfg.airplay_target == ["Living Room"] assert cfg.airplay_password == "pw" assert cfg.auto_reconnect is False assert cfg.reconnect_delay == 10.0 @@ -125,6 +129,54 @@ def test_cli_overrides_none_ignored(self, tmp_path): cfg = load_config(config_path=str(toml), sample_rate=None) assert cfg.sample_rate == 44100 + def test_airplay_target_string_normalized_to_list(self, tmp_path): + toml = tmp_path / "cusp.toml" + toml.write_text('[airplay]\ntarget = "Living Room"\n') + cfg = load_config(config_path=str(toml)) + assert cfg.airplay_target == ["Living Room"] + + def test_airplay_target_string_comma_separated(self, tmp_path): + toml = tmp_path / "cusp.toml" + toml.write_text('[airplay]\ntarget = "Living Room, Kitchen, Office"\n') + cfg = load_config(config_path=str(toml)) + assert cfg.airplay_target == ["Living Room", "Kitchen", "Office"] + + def test_airplay_target_string_comma_separated_drops_empty_entries(self, tmp_path): + toml = tmp_path / "cusp.toml" + toml.write_text('[airplay]\ntarget = "Living Room,, , Kitchen"\n') + cfg = load_config(config_path=str(toml)) + assert cfg.airplay_target == ["Living Room", "Kitchen"] + + def test_airplay_target_array(self, tmp_path): + toml = tmp_path / "cusp.toml" + toml.write_text('[airplay]\ntarget = ["Living Room", "Kitchen", "Office"]\n') + cfg = load_config(config_path=str(toml)) + assert cfg.airplay_target == ["Living Room", "Kitchen", "Office"] + + def test_airplay_target_array_strips_whitespace(self, tmp_path): + toml = tmp_path / "cusp.toml" + toml.write_text('[airplay]\ntarget = [" Living Room ", "Kitchen"]\n') + cfg = load_config(config_path=str(toml)) + assert cfg.airplay_target == ["Living Room", "Kitchen"] + + def test_airplay_target_array_drops_empty_entries(self, tmp_path): + toml = tmp_path / "cusp.toml" + toml.write_text('[airplay]\ntarget = ["Living Room", "", " "]\n') + cfg = load_config(config_path=str(toml)) + assert cfg.airplay_target == ["Living Room"] + + def test_airplay_target_empty_string_becomes_empty_list(self, tmp_path): + toml = tmp_path / "cusp.toml" + toml.write_text('[airplay]\ntarget = ""\n') + cfg = load_config(config_path=str(toml)) + assert cfg.airplay_target == [] + + def test_airplay_target_cli_list_overrides_toml(self, tmp_path): + toml = tmp_path / "cusp.toml" + toml.write_text('[airplay]\ntarget = "Living Room"\n') + cfg = load_config(config_path=str(toml), airplay_target=["Kitchen", "Office"]) + assert cfg.airplay_target == ["Kitchen", "Office"] + def test_no_file_returns_defaults(self): cfg = load_config(config_path="/nonexistent/path.toml") assert cfg == CuspConfig() @@ -164,3 +216,72 @@ def test_none_when_missing(self, tmp_path, monkeypatch): monkeypatch.chdir(tmp_path) monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path)) assert _find_config_file() is None + + +class TestStreamCLITargets: + """CLI parsing for `cusp stream -t …`. The stream pipeline is mocked.""" + + def _run(self, args, monkeypatch, tmp_path): + # Keep the CLI from finding a user/cwd config file. + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path)) + captured: dict = {} + + async def fake_run_stream(config): + captured["config"] = config + + with patch("cusp.cli._run_stream", fake_run_stream): + runner = CliRunner() + result = runner.invoke(main, ["stream", *args]) + return result, captured + + def test_cli_single_target(self, monkeypatch, tmp_path): + result, captured = self._run(["-t", "Living Room"], monkeypatch, tmp_path) + assert result.exit_code == 0 + assert captured["config"].airplay_target == ["Living Room"] + + def test_cli_comma_separated_targets(self, monkeypatch, tmp_path): + result, captured = self._run( + ["-t", "Living Room, Kitchen, Office"], monkeypatch, tmp_path + ) + assert result.exit_code == 0 + assert captured["config"].airplay_target == [ + "Living Room", + "Kitchen", + "Office", + ] + + def test_cli_comma_separated_drops_empty_entries(self, monkeypatch, tmp_path): + result, captured = self._run( + ["-t", "Living Room,, , Kitchen"], monkeypatch, tmp_path + ) + assert result.exit_code == 0 + assert captured["config"].airplay_target == ["Living Room", "Kitchen"] + + def test_cli_empty_target_rejected(self, monkeypatch, tmp_path): + result, captured = self._run(["-t", ""], monkeypatch, tmp_path) + assert result.exit_code == 1 + assert "No AirPlay target specified" in result.output + assert "config" not in captured + + def test_cli_whitespace_only_target_rejected(self, monkeypatch, tmp_path): + result, captured = self._run(["-t", " , , "], monkeypatch, tmp_path) + assert result.exit_code == 1 + assert "No AirPlay target specified" in result.output + assert "config" not in captured + + def test_cli_target_overrides_toml(self, monkeypatch, tmp_path): + toml = tmp_path / "cusp.toml" + toml.write_text('[airplay]\ntarget = ["Living Room", "Kitchen"]\n') + result, captured = self._run( + ["-c", str(toml), "-t", "Office"], monkeypatch, tmp_path + ) + assert result.exit_code == 0 + assert captured["config"].airplay_target == ["Office"] + + def test_cli_no_target_uses_toml_list(self, monkeypatch, tmp_path): + toml = tmp_path / "cusp.toml" + toml.write_text('[airplay]\ntarget = ["Living Room", "Kitchen"]\n') + result, captured = self._run(["-c", str(toml)], monkeypatch, tmp_path) + assert result.exit_code == 0 + assert captured["config"].airplay_target == ["Living Room", "Kitchen"] diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 74d2837..35b74db 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -1,11 +1,12 @@ import asyncio import struct +from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch import pytest from cusp.config import CuspConfig -from cusp.pipeline import StreamingSession, _wav_header +from cusp.pipeline import GroupStreamingSession, StreamingSession, _wav_header class TestWavHeader: @@ -119,6 +120,235 @@ async def test_stop_calls_close(self, mock_atv): mock_atv.close.assert_called_once() +def _mock_atv(stream_file=None): + atv = MagicMock() + atv.stream.stream_file = stream_file or AsyncMock() + atv.close.return_value = set() + return atv + + +def _mock_target(name: str): + return SimpleNamespace(name=name, address="10.0.0.1") + + +class TestGroupStreamingSession: + @pytest.fixture + def connect_results(self, monkeypatch): + """Patch `connect_target` to return/raise from a caller-supplied list.""" + + def _install(results): + queue = list(results) + + async def fake_connect(target): + nxt = queue.pop(0) + if isinstance(nxt, BaseException): + raise nxt + return nxt + + monkeypatch.setattr("cusp.pipeline.connect_target", fake_connect) + + return _install + + async def test_start_connects_all_devices(self, connect_results): + atv_a = _mock_atv() + atv_b = _mock_atv() + connect_results([atv_a, atv_b]) + config = CuspConfig(sample_rate=48000, channels=2) + + session = await GroupStreamingSession.start( + [_mock_target("A"), _mock_target("B")], config + ) + await asyncio.sleep(0) + + assert len(session._sessions) == 2 + atv_a.stream.stream_file.assert_called_once() + atv_b.stream.stream_file.assert_called_once() + await session.stop() + + async def test_start_partial_connect_failure_drops_follower( + self, connect_results, caplog + ): + atv_a = _mock_atv() + connect_results([atv_a, ConnectionError("B offline")]) + config = CuspConfig(sample_rate=48000, channels=2) + + with caplog.at_level("WARNING", logger="cusp.pipeline"): + session = await GroupStreamingSession.start( + [_mock_target("A"), _mock_target("B")], config + ) + + assert len(session._sessions) == 1 + assert session._sessions[0].name == "A" + assert any( + "B" in r.message and "Failed to connect" in r.message + for r in caplog.records + ) + await session.stop() + + async def test_start_all_connects_fail_raises(self, connect_results): + connect_results([ConnectionError("A"), ConnectionError("B")]) + config = CuspConfig(sample_rate=48000, channels=2) + + with pytest.raises(ConnectionError): + await GroupStreamingSession.start( + [_mock_target("A"), _mock_target("B")], config + ) + + async def test_feed_fans_out_to_all_live_readers(self, connect_results): + never_a = asyncio.get_event_loop().create_future() + never_b = asyncio.get_event_loop().create_future() + atv_a = _mock_atv(AsyncMock(side_effect=lambda _: never_a)) + atv_b = _mock_atv(AsyncMock(side_effect=lambda _: never_b)) + connect_results([atv_a, atv_b]) + config = CuspConfig(sample_rate=48000, channels=2) + + session = await GroupStreamingSession.start( + [_mock_target("A"), _mock_target("B")], config + ) + await asyncio.sleep(0) + + reader_a = atv_a.stream.stream_file.call_args[0][0] + reader_b = atv_b.stream.stream_file.call_args[0][0] + + chunk = b"\x11\x22" * 100 + session.feed(chunk) + + assert bytes(reader_a._buffer).endswith(chunk) + assert bytes(reader_b._buffer).endswith(chunk) + + never_a.cancel() + never_b.cancel() + await session.stop() + + async def test_feed_skips_failed_sub_session(self, connect_results, caplog): + never_a = asyncio.get_event_loop().create_future() + atv_a = _mock_atv(AsyncMock(side_effect=lambda _: never_a)) + # B raises as soon as its consumer task runs. + atv_b = _mock_atv(AsyncMock(side_effect=ConnectionError("B dropped"))) + connect_results([atv_a, atv_b]) + config = CuspConfig(sample_rate=48000, channels=2) + + session = await GroupStreamingSession.start( + [_mock_target("A"), _mock_target("B")], config + ) + # Let the consumer tasks run so B's failure is visible. + await asyncio.sleep(0.01) + + assert session._sessions[1].failed() + assert not session.failed() # A is still alive + + with caplog.at_level("WARNING", logger="cusp.pipeline"): + session.feed(b"\x00\x01" * 50) + # Feeding again shouldn't log a second time. + session.feed(b"\x00\x01" * 50) + + drop_msgs = [r for r in caplog.records if "dropped" in r.message] + assert len(drop_msgs) == 1 + assert "B" in drop_msgs[0].message + + # A's reader kept receiving chunks. + reader_a = atv_a.stream.stream_file.call_args[0][0] + assert bytes(reader_a._buffer).endswith(b"\x00\x01" * 50) + + never_a.cancel() + await session.stop() + + async def test_feed_continues_when_leader_drops(self, connect_results, caplog): + # Leader (index 0) fails immediately; follower stays alive. + atv_leader = _mock_atv(AsyncMock(side_effect=ConnectionError("leader gone"))) + never_follower = asyncio.get_event_loop().create_future() + atv_follower = _mock_atv(AsyncMock(side_effect=lambda _: never_follower)) + connect_results([atv_leader, atv_follower]) + config = CuspConfig(sample_rate=48000, channels=2) + + session = await GroupStreamingSession.start( + [_mock_target("leader"), _mock_target("follower")], config + ) + await asyncio.sleep(0.01) + + assert session._sessions[0].failed() + assert not session._sessions[1].failed() + # Group as a whole is still alive because the follower is streaming. + assert not session.failed() + assert session.exception() is None + + with caplog.at_level("WARNING", logger="cusp.pipeline"): + session.feed(b"\xde\xad" * 64) + + drop_msgs = [r for r in caplog.records if "dropped" in r.message] + assert len(drop_msgs) == 1 + assert "leader" in drop_msgs[0].message + + reader_follower = atv_follower.stream.stream_file.call_args[0][0] + assert bytes(reader_follower._buffer).endswith(b"\xde\xad" * 64) + + never_follower.cancel() + await session.stop() + + async def test_failed_only_when_all_sub_sessions_dead(self, connect_results): + atv_a = _mock_atv(AsyncMock(side_effect=ConnectionError("A gone"))) + atv_b = _mock_atv(AsyncMock(side_effect=ConnectionError("B gone"))) + connect_results([atv_a, atv_b]) + config = CuspConfig(sample_rate=48000, channels=2) + + session = await GroupStreamingSession.start( + [_mock_target("A"), _mock_target("B")], config + ) + await asyncio.sleep(0.01) + + assert session.failed() + exc = session.exception() + assert isinstance(exc, ConnectionError) + await session.stop() + + async def test_exception_none_while_any_alive(self, connect_results): + never_a = asyncio.get_event_loop().create_future() + atv_a = _mock_atv(AsyncMock(side_effect=lambda _: never_a)) + atv_b = _mock_atv(AsyncMock(side_effect=ConnectionError("B gone"))) + connect_results([atv_a, atv_b]) + config = CuspConfig(sample_rate=48000, channels=2) + + session = await GroupStreamingSession.start( + [_mock_target("A"), _mock_target("B")], config + ) + await asyncio.sleep(0.01) + + assert not session.failed() + assert session.exception() is None + + never_a.cancel() + await session.stop() + + async def test_stop_closes_every_device(self, connect_results): + atv_a = _mock_atv() + atv_b = _mock_atv() + connect_results([atv_a, atv_b]) + config = CuspConfig(sample_rate=48000, channels=2) + + session = await GroupStreamingSession.start( + [_mock_target("A"), _mock_target("B")], config + ) + await session.stop() + + atv_a.close.assert_called_once() + atv_b.close.assert_called_once() + + async def test_stop_tolerates_already_failed_sub_sessions(self, connect_results): + atv_a = _mock_atv() + atv_b = _mock_atv(AsyncMock(side_effect=ConnectionError("B gone"))) + connect_results([atv_a, atv_b]) + config = CuspConfig(sample_rate=48000, channels=2) + + session = await GroupStreamingSession.start( + [_mock_target("A"), _mock_target("B")], config + ) + await asyncio.sleep(0.01) + # Must not raise even though B's consumer already errored. + await session.stop() + atv_a.close.assert_called_once() + atv_b.close.assert_called_once() + + class TestRunWithReconnect: async def test_clean_exit(self): config = CuspConfig()