mopidy-webhooks/tests/test_frontend.py

from unittest import mock

import pytest

from mopidy import models
from mopidy_webhooks import frontend as frontend_lib
from mopidy_webhooks import metadata as metadata_lib


@pytest.fixture
def frontend():
    config = {
        "webhooks": {
            "urls": "http://127.0.0.1/receiver/,http://127.0.0.1/receiver/two/",
            "tokens": "secrettoken,anotherone",
            "media_dirs": "",
            "podcast_dirs": "",
        }
    }
    core = mock.sentinel.core
    return frontend_lib.WebhooksFrontend(config, core)


@pytest.fixture
def track():
    return models.Track(
        uri="local:track:Sublime%20-%20Sublime/Disc%201%20-%2004%20-%20Same%20in%20the%20End.mp3",
        name="Same in the End",
        artists=[models.Artist(name="Sublime", uri="local:artist:md5:abc")],
        album=models.Album(name="Sublime"),
        track_no=4,
        length=156604,
    )


@pytest.fixture
def podcast_track():
    return models.Track(
        uri="local:podcast:TED%20Talks/2026-09-22_Do%20the%20hard%20thing.mp3",
        name="2026-09-22_Do the hard thing",
        artists=[models.Artist(name="TED Talks Daily")],
        album=models.Album(name="TED Talks Daily"),
        length=1800000,
    )


def test_on_start_parses_urls_tokens_and_dirs(frontend):
    frontend.config["webhooks"]["media_dirs"] = "/media,/srv/media"
    frontend.config["webhooks"]["podcast_dirs"] = "/media/podcasts"

    frontend.on_start()

    assert frontend.webhook_urls == [
        "http://127.0.0.1/receiver/",
        "http://127.0.0.1/receiver/two/",
    ]
    assert frontend.webhook_tokens == ["secrettoken", "anotherone"]
    assert frontend.media_dirs == ["/media", "/srv/media"]
    assert frontend.podcast_dirs == ["/media/podcasts"]


def test_build_post_data_for_track(frontend, track):
    post_data = frontend._build_post_data(track, time_position=0)

    assert post_data["name"] == "Same in the End"
    assert post_data["artist"] == "Sublime"
    assert post_data["album"] == "Sublime"
    assert post_data["track_number"] == 4
    assert post_data["run_time"] == "156"
    assert post_data["media_type"] == "track"
    assert "podcast_name" not in post_data


def test_build_post_data_for_podcast_prefers_tags(
    frontend, podcast_track, monkeypatch
):
    monkeypatch.setattr(
        metadata_lib,
        "resolve_media_path",
        lambda uri, dirs: "/media/podcasts/ep.mp3",
    )
    monkeypatch.setattr(
        metadata_lib,
        "read_metadata",
        lambda path: {
            "title": "#564: EVE Online Departs for Python 3",
            "artist": "Michael Kennedy",
            "album": "Talk Python To Me",
            "track_number": "564",
            "date": "2026-09-11T15:46:15",
            "comment": "A great episode",
            "feed_url": "https://talkpython.fm/episodes/rss",
            "guid": "talk-python-564",
            "episode_url": "https://talkpython.fm/564",
            "duration": 4071,
        },
    )
    monkeypatch.setattr(
        metadata_lib, "is_podcast", lambda uri, **kwargs: "podcast" in uri
    )

    post_data = frontend._build_post_data(podcast_track, time_position=0)

    assert post_data["media_type"] == "podcast"
    assert post_data["name"] == "#564: EVE Online Departs for Python 3"
    assert post_data["artist"] == "Michael Kennedy"
    assert post_data["album"] == "Talk Python To Me"
    assert post_data["track_number"] == "564"
    assert post_data["podcast_name"] == "Talk Python To Me"
    assert post_data["podcast_producer"] == "Michael Kennedy"
    assert post_data["podcast_feed_url"] == "https://talkpython.fm/episodes/rss"
    assert post_data["podcast_description"] == "A great episode"
    assert post_data["episode_num"] == 564
    assert post_data["pub_date"] == "2026-09-11"
    assert post_data["episode_url"] == "https://talkpython.fm/564"
    assert post_data["episode_guid"] == "talk-python-564"
    assert post_data["duration_seconds"] == 4071


def test_build_post_data_for_podcast_falls_back_to_track(
    frontend, podcast_track
):
    post_data = frontend._build_post_data(podcast_track, time_position=0)

    assert post_data["media_type"] == "podcast"
    assert post_data["name"] == podcast_track.name
    assert post_data["podcast_name"] == "TED Talks Daily"
    assert post_data["podcast_producer"] == "TED Talks Daily"
    assert post_data["podcast_feed_url"] == ""


def test_build_post_data_for_remote_podcast_uri(frontend):
    track = models.Track(
        uri="podcast+https://feeds.npr.org/510318/podcast.xml#85b9c4c4-guid",
        name="UN General Assembly Week",
        artists=[models.Artist(name="NPR")],
        album=models.Album(name="Up First"),
        length=900000,
    )

    post_data = frontend._build_post_data(track, time_position=0)

    assert post_data["media_type"] == "podcast"
    assert post_data["podcast_name"] == "Up First"
    assert (
        post_data["podcast_feed_url"]
        == "https://feeds.npr.org/510318/podcast.xml"
    )
    assert post_data["episode_guid"] == "85b9c4c4-guid"


@mock.patch("mopidy_webhooks.frontend.requests.post")
def test_track_playback_started_posts_started(mock_post, frontend, track):
    frontend.on_start()

    frontend.track_playback_started(models.TlTrack(track=track, tlid=1))

    assert mock_post.call_count == 2
    first_call = mock_post.call_args_list[0]
    assert first_call.args[0] == "http://127.0.0.1/receiver/"
    assert first_call.kwargs["headers"] == {
        "Authorization": "Token secrettoken"
    }
    assert '"status": "started"' in first_call.kwargs["json"]


@mock.patch("mopidy_webhooks.frontend.requests.post")
def test_track_playback_ended_scrobbles_played_track(
    mock_post, frontend, track
):
    frontend.on_start()

    frontend.track_playback_ended(models.TlTrack(track=track, tlid=1), 150000)

    assert mock_post.call_count == 2
    assert '"status": "stopped"' in mock_post.call_args.kwargs["json"]


@mock.patch("mopidy_webhooks.frontend.requests.post")
def test_does_not_scrobble_if_played_less_than_half(mock_post, frontend, track):
    frontend.on_start()

    frontend.track_playback_ended(models.TlTrack(track=track, tlid=1), 60432)

    assert mock_post.call_count == 0


@mock.patch("mopidy_webhooks.frontend.requests.post")
def test_track_playback_paused_and_resumed_post(mock_post, frontend, track):
    frontend.on_start()

    frontend.track_playback_paused(models.TlTrack(track=track, tlid=1), 15000)
    frontend.track_playback_resumed(models.TlTrack(track=track, tlid=1), 15000)

    statuses = [call.kwargs["json"] for call in mock_post.call_args_list]
    assert any('"status": "paused"' in payload for payload in statuses)
    assert any('"status": "resumed"' in payload for payload in statuses)


@mock.patch("mopidy_webhooks.frontend.requests.post")
def test_no_urls_configured_posts_nothing(mock_post, frontend, track):
    frontend.on_start()
    frontend.webhook_urls = []

    frontend.track_playback_started(models.TlTrack(track=track, tlid=1))

    assert mock_post.call_count == 0


def test_metadata_cache_avoids_repeated_reads(frontend, monkeypatch, tmp_path):
    media_file = tmp_path / "song.mp3"
    media_file.write_bytes(b"data")
    frontend.media_dirs = [str(tmp_path)]
    track = models.Track(
        uri=f"local:track:{media_file.name}",
        name="Song",
        artists=[models.Artist(name="Someone", uri="local:artist:someone")],
        length=180000,
    )
    reads = []
    monkeypatch.setattr(
        metadata_lib,
        "read_metadata",
        lambda path: reads.append(path) or {"title": "Tagged"},
    )

    frontend._build_post_data(track, time_position=0)
    frontend._build_post_data(track, time_position=1000)

    assert len(reads) == 1