feat(tests): MPEG DASH support

Author: JuniorJPDJ

.github/workflows/tests.yml

@@ -43,6 +43,6 @@ jobs:
 
  - run: poetry run pytest
  env:
- TIDAL_APK_URL: ${{ secrets.TIDAL_NONDASH_APK_URL }}
- TIDAL_CLIENT_ID: ${{ secrets.TIDAL_NONDASH_CLIENT_ID }}
- TIDAL_REFRESH_TOKEN: ${{ secrets.TIDAL_NONDASH_REFRESH_TOKEN }}
+ TIDAL_APK_URL: ${{ secrets.TIDAL_APK_URL }}
+ TIDAL_CLIENT_ID: ${{ secrets.TIDAL_CLIENT_ID }}
+ TIDAL_REFRESH_TOKEN: ${{ secrets.TIDAL_REFRESH_TOKEN }}

pyproject.toml

@@ -10,6 +10,7 @@ aiohttp = "^3.6"
 music-service-async-interface = { git = "https://github.com/FUMR/music-service-async-interface.git", rev = "2bc9cd8a1fda0486f325e127e4d7e80a5b0fedcf" }
 androguard = { version = "^3.3.5", optional = true }
 http-seekable-file = { git = "https://github.com/JuniorJPDJ/http-seekable-file.git", tag = "v0.3.0", extras = ["async"], optional = true }
+mpegdash = "^0.2"
 
 [tool.poetry.dev-dependencies]
 pre-commit = "^2.12"

tests/test_tidal_async.py

@@ -2,9 +2,19 @@
 import os
 from typing import Optional, Sized
 
+import mpegdash
 import pytest
 
-from tidal_async import Album, Artist, AudioQuality, Playlist, TidalSession, Track, extract_client_id
+from tidal_async import (
+ Album,
+ Artist,
+ AudioQuality,
+ Playlist,
+ TidalSession,
+ Track,
+ dash_mpd_from_data_url,
+ extract_client_id,
+)
 
 # TODO [#63]: Unit tests!
 # - [ ] login process (not sure how to do this - it's interactive oauth2)
@@ -283,24 +293,6 @@ async def test_cover_download(sess: TidalSession, object_url, cover_size, sha256
 @pytest.mark.parametrize(
  "id_, required_quality, preferred_quality, file_size, mimetype, etag",
  (
- (
- 152676390,
- AudioQuality.Normal,
- AudioQuality.Normal,
- 3114802,
- "audio/mp4",
- '"780130927f84364021b1300423d60f47"',
- ),
- # DASH Support needed (#53)
- # (
- # 152676390,
- # AudioQuality.High,
- # AudioQuality.High,
- # 10347474,
- # "audio/mp4",
- # '"970df936b04363528662c9c74b714d13-2"',
- # ),
- (152676390, AudioQuality.HiFi, AudioQuality.HiFi, 30980344, "audio/flac", '"3bb27f3e6d8f7fd987bcc0d3cdc7c452"'),
  (
  152676390,
  AudioQuality.Master,
@@ -311,13 +303,63 @@ async def test_cover_download(sess: TidalSession, object_url, cover_size, sha256
  ),
  ),
 )
-async def test_track_download(sess: TidalSession, id_, required_quality, preferred_quality, file_size, mimetype, etag):
+async def test_track_download_direct(
+ sess: TidalSession, id_, required_quality, preferred_quality, file_size, mimetype, etag
+):
  track = await sess.track(id_)
  file = await track.get_async_file(required_quality, preferred_quality)
 
  assert file.mimetype == mimetype and file.resp_headers["ETag"] == etag and len(file) == file_size
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+ "id_, required_quality, preferred_quality, codec, bandwidth, length, segments",
+ (
+ (
+ 152676390,
+ AudioQuality.Normal,
+ AudioQuality.Normal,
+ "mp4a.40.5",
+ 96984,
+ "PT4M17.614S",
+ 64,
+ ),
+ (
+ 152676390,
+ AudioQuality.High,
+ AudioQuality.High,
+ "mp4a.40.2",
+ 321691,
+ "PT4M17.545S",
+ 64,
+ ),
+ (
+ 152676390,
+ AudioQuality.HiFi,
+ AudioQuality.HiFi,
+ "flac",
+ 957766,
+ "PT4M17.499S",
+ 64,
+ ),
+ ),
+)
+async def test_track_download_dash(
+ sess: TidalSession, id_, required_quality, preferred_quality, codec, bandwidth, length, segments
+):
+ track = await sess.track(id_)
+ url = await track.get_file_url(required_quality, preferred_quality)
+ mpd = dash_mpd_from_data_url(url)
+
+ rep = mpd.periods[0].adaptation_sets[0].representations[0]
+
+ assert rep.codecs == codec
+ assert rep.bandwidth == bandwidth
+ assert mpd.media_presentation_duration == length
+ assert sum(s.r if s.r else 1 for s in rep.segment_templates[0].segment_timelines[0].Ss)
+
+
 @pytest.mark.asyncio
 async def test_client_id_extraction():
  from io import BytesIO

tidal_async/__init__.py

@@ -2,7 +2,7 @@
 
 from .api import Album, Artist, AudioMode, AudioQuality, Cover, Playlist, TidalObject, Track
 from .session import TidalMultiSession, TidalSession
-from .utils import cli_auth_url_getter, extract_client_id
+from .utils import cli_auth_url_getter, dash_mpd_from_data_url, extract_client_id
 
 __all__ = [
  "AudioMode",
@@ -17,4 +17,5 @@
  "TidalMultiSession",
  "cli_auth_url_getter",
  "extract_client_id",
+ "dash_mpd_from_data_url",
 ]

tidal_async/utils.py

@@ -1,8 +1,11 @@
 import asyncio
+import base64
 import contextlib
 import functools
 from urllib.parse import urlparse
 
+import mpegdash.nodes
+import mpegdash.parser
 from music_service_async_interface import InvalidURL
 
 
@@ -107,6 +110,18 @@ def gen_artist(obj) -> str:
  return main if not feat else f"{main} feat. {feat}"
 
 
+def dash_mpd_from_data_url(url: str) -> "mpegdash.nodes.MPEGDASH":
+ """Parses MPEG-DASH MPD manifest
+
+ :param url: URL with `data` scheme containing encoded MPD manifest returned from `Track.get_file_url()`
+ :return: parsed MPEG DASH MPD manifest
+ """
+ assert url.startswith("data:application/dash+xml;base64,")
+ mpd_str = base64.b64decode(url.rsplit(",", 1)[1]).decode("utf-8")
+ mpd = mpegdash.parser.MPEGDASHParser.parse(mpd_str)
+ return mpd
+
+
 try:
  from zipfile import ZipFile