casting_test.py

from types import SimpleNamespace
from unittest.mock import MagicMock, patch

from fastapi import HTTPException
from pychromecast.error import NotConnected

import pytest

import casting


@pytest.fixture
def cast_env():
    manager = casting.CastManager()
    cast = MagicMock()
    cast.cast_info.friendly_name = "Living Room"
    cast.status.volume_level = 0.5
    cast.socket_client.is_connected = True
    status = SimpleNamespace(
        content_id=None,
        player_state="IDLE",
        current_time=30,
        adjusted_current_time=30,
        idle_reason=None,
    )
    cast.media_controller.status = status

    def load(url, *_args, **_kwargs):
        status.content_id = url
        status.player_state = "PLAYING"

    cast.media_controller.play_media.side_effect = load
    session = {
        "username": "alice",
        "url": "http://provider/live",
        "is_vod": False,
        "duration": 3600,
        "seek_offset": 120,
    }
    with (
        patch("casting.get_chromecast_from_host", return_value=cast) as connect,
        patch("casting.ffmpeg_session.get_session", return_value=session),
        patch("casting.ffmpeg_session.touch_session") as touch,
        patch("casting.ffmpeg_session.stop_session") as stop,
        patch("casting.save_watch_position") as save,
        patch.object(manager, "_ensure_worker"),
    ):
        yield SimpleNamespace(
            manager=manager,
            cast=cast,
            status=status,
            session=session,
            connect=connect,
            touch=touch,
            stop=stop,
            save=save,
        )
    manager.close()


def start(env, username="alice", host="192.168.1.50"):
    return env.manager.start(username, host, "session123", "http://192.168.1.10:8000", "News", 42)


@pytest.mark.parametrize(
    "host", ["localhost", "127.0.0.1", "0.0.0.0", "8.8.8.8", "169.254.1.1", "::1", "224.0.0.1"]
)
def test_device_must_be_private_lan_ip(host):
    with pytest.raises(HTTPException) as error:
        casting.lan_host(host)
    assert error.value.status_code == 400


@pytest.mark.parametrize("host", ["192.168.1.50", "10.0.0.2", "172.16.0.2", "fd00::1"])
def test_private_device_addresses(host):
    assert casting.lan_host(host) == host


@pytest.mark.parametrize(
    "origin",
    [
        "http://localhost:8000",
        "http://127.0.0.1:8000",
        "http://[::1]:8000",
        "http://0.0.0.0:8000",
        "file:///etc/passwd",
        "http://user:secret@server",
        "http://server/path",
        "http://server:bad",
        "http://server?token=1",
        "http://host.docker.internal:8000",
    ],
)
def test_unreachable_or_invalid_media_origin(origin):
    with pytest.raises(HTTPException):
        casting.media_origin(origin)


def test_http_cast_reuses_owned_session_and_live_edge(cast_env):
    result = start(cast_env)
    assert result["active"] is True
    assert result["session_id"] == "session123"
    assert result["name"] == "Living Room"
    args, kwargs = cast_env.cast.media_controller.play_media.call_args
    assert args == (
        "http://192.168.1.10:8000/transcode/session123/stream.m3u8",
        "application/x-mpegurl",
    )
    assert kwargs["stream_type"] == "LIVE"
    assert kwargs["current_time"] is None
    assert cast_env.cast.media_controller.app_must_match is True
    assert cast_env.manager.owns_session("session123")
    assert cast_env.manager.owns_url("http://provider/live")


def test_vod_and_series_use_buffered_media_and_relative_position(cast_env):
    cast_env.session["is_vod"] = True
    start(cast_env)
    kwargs = cast_env.cast.media_controller.play_media.call_args.kwargs
    assert kwargs["stream_type"] == "BUFFERED"
    assert kwargs["current_time"] == 42
    cast_env.manager.maintain_sessions()
    cast_env.save.assert_called_with("alice", "http://provider/live", 150, 3600)


def test_adaptive_cast_uses_master_playlist(cast_env):
    cast_env.session["fast_start"] = True
    start(cast_env)
    assert cast_env.cast.media_controller.play_media.call_args.args[0].endswith("/master.m3u8")


def test_other_user_cannot_cast_or_control_session(cast_env):
    with pytest.raises(HTTPException) as error:
        start(cast_env, username="bob")
    assert error.value.status_code == 404
    cast_env.connect.assert_not_called()
    start(cast_env)
    assert cast_env.manager.status("bob") == {"active": False}
    with pytest.raises(HTTPException) as error:
        cast_env.manager.command("bob", "stop")
    assert error.value.status_code == 404
    cast_env.cast.media_controller.stop.assert_not_called()


def test_existing_cast_cannot_be_overwritten(cast_env):
    start(cast_env)
    with pytest.raises(HTTPException) as error:
        start(cast_env, host="192.168.1.51")
    assert error.value.status_code == 409
    cast_env.session["username"] = "bob"
    with pytest.raises(HTTPException) as error:
        start(cast_env, username="bob")
    assert error.value.status_code == 409


def test_connection_failure_releases_reservation_but_keeps_local_stream(cast_env):
    cast_env.cast.wait.side_effect = NotConnected()
    with pytest.raises(HTTPException) as error:
        start(cast_env)
    assert error.value.status_code == 502
    assert not cast_env.manager.owns_session("session123")
    cast_env.cast.disconnect.assert_called_once()
    cast_env.stop.assert_not_called()


def test_load_timeout_does_not_report_success(cast_env):
    cast_env.cast.media_controller.play_media.side_effect = None
    with (
        patch("casting.time.monotonic", side_effect=[0, 16]),
        pytest.raises(HTTPException) as error,
    ):
        start(cast_env)
    assert error.value.status_code == 502
    assert cast_env.manager.status("alice") == {"active": False}
    cast_env.cast.disconnect.assert_called_once()


def test_server_keeps_cast_alive_without_browser(cast_env):
    start(cast_env)
    cast_env.touch.reset_mock()
    cast_env.status.player_state = "PAUSED"
    cast_env.manager.maintain_sessions()
    cast_env.touch.assert_called_once_with("session123")
    assert cast_env.manager.status("alice")["active"]


def test_disconnected_receiver_releases_ownership(cast_env):
    start(cast_env)
    cast_env.cast.socket_client.is_connected = False
    cast_env.manager.maintain_sessions()
    assert cast_env.manager.status("alice") == {"active": False}


def test_other_receiver_app_is_not_stopped(cast_env):
    start(cast_env)
    cast_env.status.content_id = "another-app"
    with pytest.raises(HTTPException) as error:
        cast_env.manager.command("alice", "stop")
    assert error.value.status_code == 409
    cast_env.cast.media_controller.stop.assert_not_called()


def test_remote_controls_and_stop_release_transcode(cast_env):
    start(cast_env)
    cast_env.manager.command("alice", "pause")
    cast_env.manager.command("alice", "play")
    cast_env.manager.command("alice", "volume", 0.25)
    assert cast_env.manager.command("alice", "stop") == {"active": False}
    cast_env.cast.media_controller.pause.assert_called_once()
    cast_env.cast.media_controller.play.assert_called_once()
    cast_env.cast.set_volume.assert_called_once_with(0.25)
    cast_env.stop.assert_called_once_with("session123", force=True)


def test_discovery_closes_browser_and_returns_lan_devices():
    manager = casting.CastManager()
    with (
        patch("casting.Zeroconf"),
        patch("casting.CastBrowser") as browser_type,
        patch.object(manager._stop, "wait"),
    ):
        browser = browser_type.return_value
        browser.devices = {
            "a": SimpleNamespace(host="192.168.1.50", friendly_name="TV"),
            "b": SimpleNamespace(host="127.0.0.1", friendly_name="Not a TV"),
        }
        assert manager.discover() == [{"host": "192.168.1.50", "name": "TV"}]
        browser.stop_discovery.assert_called_once()