MusicWorkshop/backend/tests/test_match_providers.py

import json
import time
import unittest
from unittest.mock import patch
from urllib import error

from backend.app.matcher import MatchHttpClient, MusicBrainzProvider, SpotifyProvider


class MatchProviderTests(unittest.TestCase):
  def test_match_http_client_retries_url_errors(self):
    client = MatchHttpClient()

    with patch(
      'backend.app.matcher.request.urlopen',
      side_effect=[
        error.URLError('temporary dns error'),
        FakeResponse({'ok': True})
      ]
    ) as mock_urlopen:
      payload = client.request_json('test', 'https://example.com', retries=1)

    self.assertEqual(payload['ok'], True)
    self.assertEqual(mock_urlopen.call_count, 2)

  def test_match_http_client_retries_timeout(self):
    client = MatchHttpClient()

    with patch(
      'backend.app.matcher.request.urlopen',
      side_effect=[
        TimeoutError('timeout'),
        FakeResponse({'ok': True})
      ]
    ) as mock_urlopen:
      payload = client.request_json('test', 'https://example.com', retries=1)

    self.assertEqual(payload['ok'], True)
    self.assertEqual(mock_urlopen.call_count, 2)

  def test_musicbrainz_requests_use_user_agent_and_throttle(self):
    client = MatchHttpClient()
    provider = MusicBrainzProvider(client)
    observed_headers = []

    def fake_urlopen(req, timeout):
      observed_headers.append(dict(req.header_items()))
      return FakeResponse({'recordings': []})

    with patch('backend.app.matcher.request.urlopen', side_effect=fake_urlopen) as mock_urlopen:
      with patch('backend.app.matcher.time.sleep') as mock_sleep:
        with patch(
          'backend.app.matcher.time.monotonic',
          side_effect=[0.0, 0.0, 0.1, 0.1]
        ):
          provider._request_json(
            'musicbrainz',
            'https://musicbrainz.org/ws/2/recording',
            params={'fmt': 'json'}
          )
          provider._request_json(
            'musicbrainz',
            'https://musicbrainz.org/ws/2/recording',
            params={'fmt': 'json'}
          )

    self.assertEqual(mock_urlopen.call_count, 2)
    self.assertTrue(any('User-agent' in headers or 'User-Agent' in headers for headers in observed_headers))
    self.assertTrue(mock_sleep.called)

  def test_spotify_provider_refreshes_expired_token(self):
    provider = SpotifyProvider(MatchHttpClient())
    config = {
      'metadata': {
        'spotifyUrl': 'https://api.spotify.com/v1',
        'spotifyClientId': 'spotify-id',
        'spotifySecret': 'spotify-secret'
      }
    }
    observed_authorization = []
    token_counter = {'value': 0}

    def fake_urlopen(req, timeout):
      url = req.full_url
      if 'api/token' in url:
        token_counter['value'] += 1
        return FakeResponse(
          {
            'access_token': f'token-{token_counter["value"]}',
            'expires_in': 3600
          }
        )

      observed_authorization.append(req.headers.get('Authorization'))
      return FakeResponse(
        {
          'tracks': {
            'items': [
              {
                'id': 'track-1',
                'name': 'Song Title',
                'artists': [{'name': 'Song Artist'}],
                'album': {
                  'id': 'album-1',
                  'name': 'Album Name',
                  'release_date': '2024-01-01',
                  'images': []
                },
                'track_number': 1,
                'disc_number': 1,
                'duration_ms': 201000
              }
            ]
          }
        }
      )

    with patch('backend.app.matcher.request.urlopen', side_effect=fake_urlopen):
      provider.search(
        {
          'title': 'Song Title',
          'artist': 'Song Artist',
          'album': 'Album Name'
        },
        config
      )
      cache_key = 'spotify-id:spotify-secret'
      provider._token_cache[cache_key]['expires_at'] = time.time() - 1
      provider.search(
        {
          'title': 'Song Title',
          'artist': 'Song Artist',
          'album': 'Album Name'
        },
        config
      )

    self.assertEqual(token_counter['value'], 2)
    self.assertEqual(observed_authorization, ['Bearer token-1', 'Bearer token-2'])

  def test_spotify_provider_skips_when_credentials_are_missing(self):
    provider = SpotifyProvider(MatchHttpClient())
    config = {
      'metadata': {
        'spotifyUrl': 'https://api.spotify.com/v1',
        'spotifyClientId': '',
        'spotifySecret': ''
      }
    }

    with patch('backend.app.matcher.request.urlopen') as mock_urlopen:
      candidates = provider.search(
        {
          'title': 'Song Title',
          'artist': 'Song Artist',
          'album': 'Album Name'
        },
        config
      )

    self.assertEqual(candidates, [])
    self.assertEqual(mock_urlopen.call_count, 0)


class FakeHeaders:
  def get_content_charset(self):
    return 'utf-8'


class FakeResponse:
  def __init__(self, payload):
    self._payload = json.dumps(payload).encode('utf-8')
    self.headers = FakeHeaders()

  def __enter__(self):
    return self

  def __exit__(self, exc_type, exc, tb):
    return False

  def read(self):
    return self._payload


if __name__ == '__main__':
  unittest.main()