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