changedetection.io/changedetectionio/tests/apprise/test_apprise_custom_api_call.py

import json
from unittest.mock import patch

import pytest
import requests
from apprise.utils.parse import parse_url as apprise_parse_url

from ...apprise_plugin.custom_handlers import (
    _get_auth,
    _get_headers,
    _get_params,
    apprise_http_custom_handler,
    SUPPORTED_HTTP_METHODS,
)


@pytest.mark.parametrize(
    "url,expected_auth",
    [
        ("get://user:pass@localhost:9999", ("user", "pass")),
        ("get://user@localhost:9999", "user"),
        ("get://localhost:9999", ""),
        ("get://user%20name:pass%20word@localhost:9999", ("user name", "pass word")),
    ],
)
def test_get_auth(url, expected_auth):
    """Test authentication extraction with various URL formats."""
    parsed_url = apprise_parse_url(url)
    assert _get_auth(parsed_url) == expected_auth


@pytest.mark.parametrize(
    "url,body,expected_content_type",
    [
        (
            "get://localhost:9999?+content-type=application/xml",
            "test",
            "application/xml",
        ),
        ("get://localhost:9999", '{"key": "value"}', "application/json; charset=utf-8"),
        ("get://localhost:9999", "plain text", None),
        ("get://localhost:9999?+content-type=text/plain", "test", "text/plain"),
    ],
)
def test_get_headers(url, body, expected_content_type):
    """Test header extraction and content type detection."""
    parsed_url = apprise_parse_url(url)
    headers = _get_headers(parsed_url, body)

    if expected_content_type:
        assert headers.get("Content-Type") == expected_content_type


@pytest.mark.parametrize(
    "url,expected_params",
    [
        ("get://localhost:9999?param1=value1", {"param1": "value1"}),
        ("get://localhost:9999?param1=value1&-param2=ignored", {"param1": "value1"}),
        ("get://localhost:9999?param1=value1&+header=test", {"param1": "value1"}),
        (
            "get://localhost:9999?encoded%20param=encoded%20value",
            {"encoded param": "encoded value"},
        ),
    ],
)
def test_get_params(url, expected_params):
    """Test parameter extraction with URL encoding and exclusion logic."""
    parsed_url = apprise_parse_url(url)
    params = _get_params(parsed_url)
    assert dict(params) == expected_params


@pytest.mark.parametrize(
    "url,schema,method",
    [
        ("get://localhost:9999", "get", "GET"),
        ("post://localhost:9999", "post", "POST"),
        ("delete://localhost:9999", "delete", "DELETE"),
    ],
)
@patch("requests.request")
def test_apprise_custom_api_call_success(mock_request, url, schema, method):
    """Test successful API calls with different HTTP methods and schemas."""
    mock_request.return_value.raise_for_status.return_value = None

    meta = {"url": url, "schema": schema}
    result = apprise_http_custom_handler(
        body="test body", title="Test Title", notify_type="info", meta=meta
    )

    assert result is True
    mock_request.assert_called_once()

    call_args = mock_request.call_args
    assert call_args[1]["method"] == method.upper()
    assert call_args[1]["url"].startswith("http")


@patch("requests.request")
def test_apprise_custom_api_call_with_auth(mock_request):
    """Test API call with authentication."""
    mock_request.return_value.raise_for_status.return_value = None

    url = "get://user:pass@localhost:9999/secure"
    meta = {"url": url, "schema": "get"}

    result = apprise_http_custom_handler(
        body=json.dumps({"key": "value"}),
        title="Secure Test",
        notify_type="info",
        meta=meta,
    )

    assert result is True
    mock_request.assert_called_once()
    call_args = mock_request.call_args
    assert call_args[1]["auth"] == ("user", "pass")


@pytest.mark.parametrize(
    "exception_type,expected_result",
    [
        (requests.RequestException, False),
        (requests.HTTPError, False),
        (Exception, False),
    ],
)
@patch("requests.request")
def test_apprise_custom_api_call_failure(mock_request, exception_type, expected_result):
    """Test various failure scenarios."""
    url = "get://localhost:9999/error"
    meta = {"url": url, "schema": "get"}

    # Simulate different types of exceptions
    mock_request.side_effect = exception_type("Error occurred")

    result = apprise_http_custom_handler(
        body="error body", title="Error Test", notify_type="error", meta=meta
    )

    assert result == expected_result


def test_invalid_url_parsing():
    """Test handling of invalid URL parsing."""
    meta = {"url": "invalid://url", "schema": "invalid"}
    result = apprise_http_custom_handler(
        body="test", title="Invalid URL", notify_type="info", meta=meta
    )

    assert result is False


@pytest.mark.parametrize(
    "schema,expected_method",
    [
        (http_method, http_method.upper())
        for http_method in SUPPORTED_HTTP_METHODS
    ],
)
@patch("requests.request")
def test_http_methods(mock_request, schema, expected_method):
    """Test all supported HTTP methods."""
    mock_request.return_value.raise_for_status.return_value = None

    url = f"{schema}://localhost:9999"

    result = apprise_http_custom_handler(
        body="test body",
        title="Test Title",
        notify_type="info",
        meta={"url": url, "schema": schema},
    )

    assert result is True
    mock_request.assert_called_once()

    call_args = mock_request.call_args
    assert call_args[1]["method"] == expected_method


@pytest.mark.parametrize(
    "input_schema,expected_method",
    [
        (f"{http_method}s", http_method.upper())
        for http_method in SUPPORTED_HTTP_METHODS
    ],
)
@patch("requests.request")
def test_https_method_conversion(
    mock_request, input_schema, expected_method
):
    """Validate that methods ending with 's' use HTTPS and correct HTTP method."""
    mock_request.return_value.raise_for_status.return_value = None

    url = f"{input_schema}://localhost:9999"

    result = apprise_http_custom_handler(
        body="test body",
        title="Test Title",
        notify_type="info",
        meta={"url": url, "schema": input_schema},
    )

    assert result is True
    mock_request.assert_called_once()

    call_args = mock_request.call_args

    assert call_args[1]["method"] == expected_method
    assert call_args[1]["url"].startswith("https")