test_webservice.py

# -*- coding: utf-8 -*-
#
# Picard, the next-generation MusicBrainz tagger
#
# Copyright (C) 2017 Sambhav Kothari
# Copyright (C) 2017-2018 Wieland Hoffmann
# Copyright (C) 2018, 2020-2024 Laurent Monin
# Copyright (C) 2019-2024 Philipp Wolfer
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.


import sys
from unittest.mock import (
    MagicMock,
    patch,
)

from PyQt6.QtCore import QUrl
from PyQt6.QtNetwork import (
    QNetworkProxy,
    QNetworkRequest,
)

from test.picardtestcase import PicardTestCase

from picard import config
from picard.webservice import (
    TEMP_ERRORS_RETRIES,
    RequestPriorityQueue,
    RequestTask,
    UnknownResponseParserError,
    WebService,
    WSRequest,
    ratecontrol,
)
from picard.webservice.utils import (
    host_port_to_url,
    hostkey_from_url,
    port_from_qurl,
)


PROXY_SETTINGS = {
    "use_proxy": True,
    "proxy_type": 'http',
    "proxy_server_host": '127.0.0.1',
    "proxy_server_port": 3128,
    "proxy_username": 'user',
    "proxy_password": 'password',
    "network_transfer_timeout_seconds": 30,
    "network_cache_size_bytes": 100*1000*1000,
}


def dummy_handler(*args, **kwargs):
    """Dummy handler method for tests"""


class WebServiceTest(PicardTestCase):

    def setUp(self):
        super().setUp()
        self.set_config_values({
            'use_proxy': False,
            'server_host': '',
            'network_transfer_timeout_seconds': 30,
            'network_cache_size_bytes': 100*1000*1000,
        })
        self.ws = WebService()

    @patch.object(WebService, 'add_task')
    def test_webservice_url_method_calls(self, mock_add_task):
        url = "http://abc.xyz"
        handler = dummy_handler
        data = None

        def get_wsreq(mock_add_task):
            return mock_add_task.call_args[0][1]

        self.ws.get_url(url=url, handler=handler)
        self.assertEqual(1, mock_add_task.call_count)
        self.assertEqual('abc.xyz', get_wsreq(mock_add_task).host)
        self.assertEqual(80, get_wsreq(mock_add_task).port)
        self.assertIn("GET", get_wsreq(mock_add_task).method)
        self.ws.post_url(url=url, data=data, handler=handler)
        self.assertIn("POST", get_wsreq(mock_add_task).method)
        self.ws.put_url(url=url, data=data, handler=handler)
        self.assertIn("PUT", get_wsreq(mock_add_task).method)
        self.ws.delete_url(url=url, handler=handler)
        self.assertIn("DELETE", get_wsreq(mock_add_task).method)
        self.ws.download_url(url=url, handler=handler)
        self.assertIn("GET", get_wsreq(mock_add_task).method)
        self.assertEqual(5, mock_add_task.call_count)


class WebServiceTaskTest(PicardTestCase):

    def setUp(self):
        super().setUp()
        self.set_config_values({
            'use_proxy': False,
            'network_transfer_timeout_seconds': 30,
            'network_cache_size_bytes': 100*1000*1000,
        })
        self.ws = WebService()
        self.queue = self.ws._queue = MagicMock()

        # Patching the QTimers since they can only be started in a QThread
        self.ws._timer_run_next_task = MagicMock()
        self.ws._timer_count_pending_requests = MagicMock()

    def test_add_task(self):
        request = WSRequest(
            method='GET',
            url='http://abc.xyz',
            handler=dummy_handler,
        )
        func = 1
        task = self.ws.add_task(func, request)
        self.assertEqual((request.get_host_key(), func, 0), task)
        self.ws._queue.add_task.assert_called_with(task, False)
        request.important = True
        task = self.ws.add_task(func, request)
        self.ws._queue.add_task.assert_called_with(task, True)

    def test_add_task_calls_timers(self):
        mock_timer1 = self.ws._timer_run_next_task
        mock_timer2 = self.ws._timer_count_pending_requests
        request = WSRequest(
            method='GET',
            url='http://abc.xyz',
            handler=dummy_handler,
        )
        self.ws.add_task(0, request)
        mock_timer1.start.assert_not_called()
        mock_timer2.start.assert_not_called()

        # Test if timer start was called in case it was inactive
        mock_timer1.isActive.return_value = False
        mock_timer2.isActive.return_value = False
        self.ws.add_task(0, request)
        mock_timer1.start.assert_called_with(0)
        mock_timer2.start.assert_called_with(0)

    def test_remove_task(self):
        task = RequestTask(('example.com', 80), dummy_handler, priority=0)
        self.ws.remove_task(task)
        self.ws._queue.remove_task.assert_called_with(task)

    def test_remove_task_calls_timers(self):
        mock_timer = self.ws._timer_count_pending_requests
        task = RequestTask(('example.com', 80), dummy_handler, priority=0)
        self.ws.remove_task(task)
        mock_timer.start.assert_not_called()
        mock_timer.isActive.return_value = False
        self.ws.remove_task(task)
        mock_timer.start.assert_called_with(0)

    def test_run_next_task(self):
        mock_timer = self.ws._timer_run_next_task
        self.ws._queue.run_ready_tasks.return_value = sys.maxsize
        self.ws._run_next_task()
        self.ws._queue.run_ready_tasks.assert_called()
        mock_timer.start.assert_not_called()

    def test_run_next_task_starts_next(self):
        mock_timer = self.ws._timer_run_next_task
        delay = 42
        self.ws._queue.run_ready_tasks.return_value = delay
        self.ws._run_next_task()
        self.ws._queue.run_ready_tasks.assert_called()
        mock_timer.start.assert_called_with(42)


class RequestTaskTest(PicardTestCase):

    def test_from_request(self):
        request = WSRequest(
            method='GET',
            url='https://example.com',
            handler=dummy_handler,
            priority=True,
        )
        func = 1
        task = RequestTask.from_request(request, func)
        self.assertEqual(request.get_host_key(), task.hostkey)
        self.assertEqual(func, task.func)
        self.assertEqual(1, task.priority)
        self.assertEqual((request.get_host_key(), func, 1), task)


class RequestPriorityQueueTest(PicardTestCase):

    def test_add_task(self):
        queue = RequestPriorityQueue(ratecontrol)
        key = ("abc.xyz", 80)

        task1 = RequestTask(key, dummy_handler, priority=0)
        queue.add_task(task1)
        task2 = RequestTask(key, dummy_handler, priority=1)
        queue.add_task(task2)
        task3 = RequestTask(key, dummy_handler, priority=0)
        queue.add_task(task3, important=True)
        task4 = RequestTask(key, dummy_handler, priority=1)
        queue.add_task(task4, important=True)

        # Test if 2 requests were added in each queue
        self.assertEqual(len(queue._queues[0][key]), 2)
        self.assertEqual(len(queue._queues[1][key]), 2)

        # Test if important request was added ahead in the queue
        self.assertEqual(queue._queues[0][key][0], task3.func)
        self.assertEqual(queue._queues[0][key][1], task1.func)
        self.assertEqual(queue._queues[1][key][0], task4.func)
        self.assertEqual(queue._queues[1][key][1], task2.func)

    def test_remove_task(self):
        queue = RequestPriorityQueue(ratecontrol)
        key = ("abc.xyz", 80)

        # Add a task and check for its existence
        task = RequestTask(key, dummy_handler, priority=0)
        task = queue.add_task(task)
        self.assertIn(key, queue._queues[0])
        self.assertEqual(len(queue._queues[0][key]), 1)

        # Remove the task and check
        queue.remove_task(task)
        self.assertIn(key, queue._queues[0])
        self.assertEqual(len(queue._queues[0][key]), 0)

        # Try to remove a non existing task and check for errors
        non_existing_task = (1, "a", "b")
        queue.remove_task(non_existing_task)

    def test_run_task(self):
        mock_ratecontrol = MagicMock()
        delay_func = mock_ratecontrol.get_delay_to_next_request = MagicMock()

        queue = RequestPriorityQueue(mock_ratecontrol)
        key = ("abc.xyz", 80)

        # Patching the get delay function to delay the 2nd task on queue to the next call
        delay_func.side_effect = [(False, 0), (True, 0), (False, 0), (False, 0), (False, 0), (False, 0)]
        func1 = MagicMock()
        task1 = RequestTask(key, func1, priority=0)
        queue.add_task(task1)
        func2 = MagicMock()
        task2 = RequestTask(key, func2, priority=1)
        queue.add_task(task2)
        task3 = RequestTask(key, func1, priority=0)
        queue.add_task(task3)
        task4 = RequestTask(key, func1, priority=0)
        queue.add_task(task4)

        # Ensure no tasks are run before run_next_task is called
        self.assertEqual(func1.call_count, 0)
        queue.run_ready_tasks()

        # Ensure priority task is run first
        self.assertEqual(func2.call_count, 1)
        self.assertEqual(func1.call_count, 0)
        self.assertIn(key, queue._queues[1])

        # Ensure that the calls are run as expected
        queue.run_ready_tasks()
        self.assertEqual(func1.call_count, 1)

        # Checking if the cleanup occurred on the prio queue
        self.assertNotIn(key, queue._queues[1])

        # Check the call counts on proper execution of tasks
        queue.run_ready_tasks()
        self.assertEqual(func1.call_count, 2)
        queue.run_ready_tasks()
        self.assertEqual(func1.call_count, 3)

        # Ensure that the clean up happened on the normal queue
        queue.run_ready_tasks()
        self.assertEqual(func1.call_count, 3)
        self.assertNotIn(key, queue._queues[0])

    def test_count(self):
        queue = RequestPriorityQueue(ratecontrol)
        key = ("abc.xyz", 80)

        self.assertEqual(0, queue.count())
        task1 = RequestTask(key, dummy_handler, priority=0)
        queue.add_task(task1)
        self.assertEqual(1, queue.count())
        task2 = RequestTask(key, dummy_handler, priority=1)
        queue.add_task(task2)
        self.assertEqual(2, queue.count())
        task3 = RequestTask(key, dummy_handler, priority=0)
        queue.add_task(task3, important=True)
        self.assertEqual(3, queue.count())
        task4 = RequestTask(key, dummy_handler, priority=1)
        queue.add_task(task4, important=True)
        self.assertEqual(4, queue.count())
        queue.remove_task(task1)
        self.assertEqual(3, queue.count())
        queue.remove_task(task2)
        self.assertEqual(2, queue.count())
        queue.remove_task(task3)
        self.assertEqual(1, queue.count())
        queue.remove_task(task4)
        self.assertEqual(0, queue.count())


class WebServiceProxyTest(PicardTestCase):

    def setUp(self):
        super().setUp()
        self.set_config_values(PROXY_SETTINGS)

    def test_proxy_setup(self):
        proxy_types = [
            ('http', QNetworkProxy.ProxyType.HttpProxy),
            ('socks', QNetworkProxy.ProxyType.Socks5Proxy),
        ]
        for proxy_type, expected_qt_type in proxy_types:
            config.setting['proxy_type'] = proxy_type
            ws = WebService()
            proxy = ws.manager.proxy()
            self.assertEqual(proxy.type(), expected_qt_type)
            self.assertEqual(proxy.user(), PROXY_SETTINGS['proxy_username'])
            self.assertEqual(proxy.password(), PROXY_SETTINGS['proxy_password'])
            self.assertEqual(proxy.hostName(), PROXY_SETTINGS['proxy_server_host'])
            self.assertEqual(proxy.port(), PROXY_SETTINGS['proxy_server_port'])


class ParserHookTest(PicardTestCase):

    def test_parser_hook(self):
        WebService.add_parser('A', 'mime', 'parser')

        self.assertIn('A', WebService.PARSERS)
        self.assertEqual(WebService.PARSERS['A'].mimetype, 'mime')
        self.assertEqual(WebService.PARSERS['A'].mimetype, WebService.get_response_mimetype('A'))
        self.assertEqual(WebService.PARSERS['A'].parser, 'parser')
        self.assertEqual(WebService.PARSERS['A'].parser, WebService.get_response_parser('A'))

        with self.assertRaises(UnknownResponseParserError):
            WebService.get_response_parser('B')
        with self.assertRaises(UnknownResponseParserError):
            WebService.get_response_mimetype('B')


class WSRequestTest(PicardTestCase):

    def test_init_minimal(self):
        request = WSRequest(url='https://example.org/path', method='GET', handler=dummy_handler)
        self.assertEqual(request.host, 'example.org')
        self.assertEqual(request.port, 443)
        self.assertEqual(request.path, '/path')
        self.assertEqual(request.handler, dummy_handler)
        self.assertEqual(request.method, 'GET')
        self.assertEqual(request.get_host_key(), ('example.org', 443))
        self.assertIsNone(request.parse_response_type)
        self.assertIsNone(request.data)
        self.assertIsNone(request.cacheloadcontrol)
        self.assertIsNone(request.request_mimetype)
        self.assertFalse(request.mblogin)
        self.assertFalse(request.refresh)
        self.assertFalse(request.priority)
        self.assertFalse(request.important)
        self.assertFalse(request.has_auth)

    def test_init_minimal_extra(self):
        request = WSRequest(
            url='https://example.org/path',
            method='GET',
            handler=dummy_handler,
            priority=True,
            important=True,
            refresh=True,
        )
        self.assertTrue(request.priority)
        self.assertTrue(request.important)
        self.assertTrue(request.refresh)

    def test_init_minimal_qurl(self):
        url = 'https://example.org/path?q=1'
        request = WSRequest(url=QUrl(url), method='GET', handler=dummy_handler)
        self.assertEqual(request.url().toString(), url)

    def test_init_port_80(self):
        request = WSRequest(url='http://example.org/path', method='GET', handler=dummy_handler)
        self.assertEqual(request.port, 80)

    def test_init_port_other(self):
        request = WSRequest(url='http://example.org:666/path', method='GET', handler=dummy_handler)
        self.assertEqual(request.port, 666)

    def test_missing_url(self):
        with self.assertRaises(AssertionError):
            WSRequest(method='GET', handler=dummy_handler)

    def test_missing_method(self):
        with self.assertRaises(AssertionError):
            WSRequest(url='http://x', handler=dummy_handler)

    def test_missing_handler(self):
        with self.assertRaises(AssertionError):
            WSRequest(url='http://x', method='GET')

    def test_invalid_method(self):
        with self.assertRaises(AssertionError):
            WSRequest(url='http://x', method='XXX', handler=dummy_handler)

    def test_set_cacheloadcontrol(self):
        request = WSRequest(
            url='http://example.org/path',
            method='GET',
            handler=dummy_handler,
            cacheloadcontrol=QNetworkRequest.CacheLoadControl.AlwaysNetwork,
        )
        self.assertEqual(request.cacheloadcontrol, QNetworkRequest.CacheLoadControl.AlwaysNetwork)

    def test_set_parse_response_type(self):
        WebService.add_parser('A', 'mime', 'parser')
        request = WSRequest(
            url='http://example.org/path',
            method='GET',
            handler=dummy_handler,
            parse_response_type='A',
        )
        self.assertEqual(request.response_mimetype, 'mime')
        self.assertEqual(request.response_parser, 'parser')

    def test_set_invalid_parse_response_type(self):
        WebService.add_parser('A', 'mime', 'parser')
        request = WSRequest(
            url='http://example.org/path',
            method='GET',
            handler=dummy_handler,
            parse_response_type='invalid',
        )
        self.assertEqual(request.response_mimetype, None)
        self.assertEqual(request.response_parser, None)

    def test_set_mblogin_access_token(self):
        request = WSRequest(
            url='http://example.org/path',
            method='POST',
            handler=dummy_handler,
        )
        # setter
        request.mblogin = 'test'
        # getter
        self.assertEqual(request.mblogin, 'test')

        # auth needs a token too
        self.assertFalse(request.has_auth)

        # setter
        request.access_token = 'token'  # nosec
        # getter
        self.assertEqual(request.access_token, 'token')

        # auth is now possible
        self.assertTrue(request.has_auth)

    def test_set_data(self):
        request = WSRequest(
            url='http://example.org/path',
            method='POST',
            handler=dummy_handler,
            data='data',
        )
        self.assertEqual(request.data, 'data')

    def test_set_retries_reached(self):
        request = WSRequest(
            url='http://example.org/path',
            method='GET',
            handler=dummy_handler,
        )
        for i in range(0, TEMP_ERRORS_RETRIES):
            self.assertEqual(request.mark_for_retry(), i+1)

        self.assertTrue(request.max_retries_reached())

    def test_set_retries_not_reached(self):
        request = WSRequest(
            url='http://example.org/path',
            method='GET',
            handler=dummy_handler,
        )
        self.assertTrue(TEMP_ERRORS_RETRIES > 1)
        self.assertEqual(request.mark_for_retry(), 1)
        self.assertFalse(request.max_retries_reached())

    def test_queryargs(self):
        request = WSRequest(
            url='http://example.org/path?a=1',
            method='GET',
            handler=dummy_handler,
            queryargs={'a': 2, 'b': 'x%20x', 'c': '1+2', 'd': '&', 'e': '?'},
        )
        expected = 'http://example.org/path?a=1&a=2&b=x x&c=1+2&d=%26&e=?'
        self.assertEqual(request.url().toString(), expected)

    def test_unencoded_queryargs(self):
        request = WSRequest(
            url='http://example.org/path?a=1',
            method='GET',
            handler=dummy_handler,
            unencoded_queryargs={'a': 2, 'b': 'x%20x', 'c': '1+2', 'd': '&', 'e': '?'},
        )
        expected = 'http://example.org/path?a=1&a=2&b=x%2520x&c=1%2B2&d=%26&e=%3F'
        self.assertEqual(request.url().toString(), expected)

    def test_mixed_queryargs(self):
        request = WSRequest(
            url='http://example.org/path?a=1',
            method='GET',
            handler=dummy_handler,
            queryargs={'a': '2&', 'b': '1&', 'c': '&'},
            unencoded_queryargs={'a': '1&', 'b': '2&', 'd': '&'},
        )
        expected = 'http://example.org/path?a=1&a=1%26&b=2%26&c=%26&d=%26'
        self.assertEqual(request.url().toString(), expected)


class WebServiceUtilsTest(PicardTestCase):

    def test_port_from_qurl_http(self):
        self.assertEqual(port_from_qurl(QUrl('http://example.org')), 80)

    def test_port_from_qurl_http_other(self):
        self.assertEqual(port_from_qurl(QUrl('http://example.org:666')), 666)

    def test_port_from_qurl_https(self):
        self.assertEqual(port_from_qurl(QUrl('https://example.org')), 443)

    def test_port_from_qurl_https_other(self):
        self.assertEqual(port_from_qurl(QUrl('https://example.org:666')), 666)

    def test_port_from_qurl_exception(self):
        with self.assertRaises(AttributeError):
            port_from_qurl('xxx')

    def test_hostkey_from_qurl_http(self):
        self.assertEqual(hostkey_from_url(QUrl('http://example.org')), ('example.org', 80))

    def test_hostkey_from_url_https_other(self):
        self.assertEqual(hostkey_from_url('https://example.org:666'), ('example.org', 666))

    def test_host_port_to_url_http_80(self):
        self.assertEqual(host_port_to_url('example.org', 80, as_string=True), 'http://example.org')

    def test_host_port_to_url_http_80_qurl(self):
        self.assertEqual(host_port_to_url('example.org', 80).toString(), 'http://example.org')

    def test_host_port_to_url_https_443(self):
        self.assertEqual(host_port_to_url('example.org', 443, as_string=True), 'https://example.org')

    def test_host_port_to_url_https_scheme_80(self):
        self.assertEqual(host_port_to_url('example.org', 80, scheme='https', as_string=True), 'https://example.org:80')

    def test_host_port_to_url_http_666_with_path(self):
        self.assertEqual(host_port_to_url('example.org', 666, path='/abc', as_string=True), 'http://example.org:666/abc')