import os
import platform
import shutil
import ssl
import stat
from collections.abc import Iterable, Iterator
from pathlib import Path
from unittest import mock

import httpcore
import httpx
import pytest

from briefcase.exceptions import (
    BadNetworkResourceError,
    MissingNetworkResourceError,
    NetworkFailure,
)
from briefcase.integrations.base import ToolCache

TEMPORARY_DOWNLOAD_FILE_SUFFIX = ".download"


@pytest.fixture
def mock_tools(mock_tools) -> ToolCache:
    mock_tools.httpx = mock.MagicMock(spec_set=httpx)
    # Restore move so the temporary file can be moved after downloaded
    mock_tools.shutil.move = mock.MagicMock(wraps=shutil.move)
    return mock_tools


class _IteratorByteSteam(httpx.SyncByteStream):
    """Shim that satisfies ``httpx.Response`` ``stream`` parameter type.

    Cannot be replaced by any ``Iterable[bytes]`` because the base class requires
    an explicit finalization method ``close``.
    """

    def __init__(self, iterable: Iterable[bytes]) -> None:
        self.iterable = iterable

    def __iter__(self) -> Iterator[bytes]:
        return iter(self.iterable)


def _make_httpx_response(
    *,
    url: str,
    status_code: int,
    stream: list[bytes],
    method: str = "GET",
    headers: dict | None = None,
) -> httpx.Response:
    """Create a real ``httpx.Response`` with key methods wrapped by ``mock.Mock`` for
    spying.

    Wrapped methods:
        response.read
        response.iter_bytes
        response.headers.get
    """
    if headers is None:
        headers = {}

    response = httpx.Response(
        request=httpx.Request(
            method=method,
            url=httpx.URL(url),
        ),
        status_code=status_code,
        headers=httpx.Headers(headers),
        # Always use ``stream`` rather than content because it's more flexible
        # even if the request is made non-streaming or the response is read with
        # ``response.read()``, httpx will still consume the ``stream`` response
        # content internally. This allows testing both the non-streaming and
        # streaming download paths without needing to complicate the response params
        stream=_IteratorByteSteam(stream),
    )

    response.read = mock.Mock(wraps=response.read)
    response.iter_bytes = mock.Mock(wraps=response.iter_bytes)
    response.headers.get = mock.Mock(wraps=response.headers.get)

    return response


@pytest.fixture
def file_perms() -> int:
    """The expected permissions for the downloaded file.

    Since umask can vary on different systems, it is updated to a known value and reset
    after the test finishes.

    On Windows, the umask seems to always be zero and chmod doesn't really do anything
    anyway.
    """
    if platform.system() == "Windows":
        yield 0o666
    else:
        orig_umask = os.umask(0o077)
        yield 0o600
        os.umask(orig_umask)


@pytest.mark.parametrize(
    "url,content_disposition",
    [
        # A `None` value for `content_disposition` means we skip the header.
        # Other values are passed through as HTTP header values.
        ("https://example.com/path/to/something.zip", None),
        # Ensure empty header is ignored.
        ("https://example.com/path/to/something.zip", ""),
        # Paradigmatic case for Content-Disposition: attachment.
        (
            "https://example.com/path/to/irrelevant.zip",
            "attachment; filename=something.zip",
        ),
        # Ensure extra parameters are ignored.
        (
            "https://example.com/path/to/irrelevant.zip",
            "attachment; filename=something.zip; ignored=okay",
        ),
        # Ensure garbage headers are ignored.
        ("https://example.com/path/to/something.zip", "garbage"),
        # Ensure we respect unusual quoting & case permitted by RFC 6266.
        (
            "https://example.com/path/to/irrelevant.zip",
            'ATTACHment; filename=    "something.zip"',
        ),
        # Ensure we use filename=, even if filename*= is provided. This makes us a
        # "legacy user agent" in the terms of RFC 6266, for our own simplicity.
        (
            "https://example.com/path/to/irrelevant.zip",
            'attachment; filename="something.zip"; filename*=utf-8'
            "''%e2%82%ac%20rates",
        ),
    ],
)
def test_new_download_oneshot(mock_tools, file_perms, url, content_disposition):
    """If no content-length is provided, ``File`` downloadds the file all at once."""
    response = _make_httpx_response(
        method="GET",
        url=url,
        status_code=200,
        headers=(
            {
                "content-disposition": content_disposition,
            }
            if content_disposition is not None
            else {}
        ),
        stream=[b"all content"],
    )
    mock_tools.httpx.stream.return_value.__enter__.return_value = response

    # Download the file
    filename = mock_tools.file.download(
        url="https://example.com/support?useful=Yes",
        download_path=mock_tools.base_path / "downloads",
    )

    # httpx.stream has been invoked, but content isn't iterated
    mock_tools.httpx.stream.assert_called_with(
        "GET",
        "https://example.com/support?useful=Yes",
        follow_redirects=True,
        verify=mock_tools.file.ssl_context,
    )
    response.headers.get.assert_called_with("content-length")
    response.read.assert_called_once()

    # The filename is derived from the URL or header
    assert filename == mock_tools.base_path / "downloads/something.zip"

    # Temporary file was created in download dir and was renamed
    temp_filename = Path(mock_tools.shutil.move.call_args_list[0].args[0])
    assert temp_filename.parent == mock_tools.base_path / "downloads"
    assert temp_filename.name.startswith("something.zip.")
    assert temp_filename.name.endswith(TEMPORARY_DOWNLOAD_FILE_SUFFIX)
    mock_tools.shutil.move.assert_called_with(str(temp_filename), filename)

    # File permissions were set
    mock_tools.os.chmod.assert_called_with(filename, file_perms)
    assert stat.S_IMODE(os.stat(filename).st_mode) == file_perms

    # Attempt to delete Temporary file was made
    mock_tools.os.remove.assert_called_with(str(temp_filename))

    # File content is as expected
    with (mock_tools.base_path / "downloads/something.zip").open(encoding="utf-8") as f:
        assert f.read() == "all content"


def test_new_download_chunked(mock_tools, file_perms):
    """If a content-length is provided, ``File`` streams the response rather than
    downloading it all at once."""
    response = _make_httpx_response(
        method="GET",
        url="https://example.com/path/to/something.zip",
        status_code=200,
        headers={"content-length": "24"},
        stream=[
            b"chunk-1;",
            b"chunk-2;",
            b"chunk-3;",
        ],
    )
    mock_tools.httpx.stream.return_value.__enter__.return_value = response

    # Download the file
    filename = mock_tools.file.download(
        url="https://example.com/support?useful=Yes",
        download_path=mock_tools.base_path,
    )

    # httpx.stream has been invoked, and content is chunked.
    mock_tools.httpx.stream.assert_called_with(
        "GET",
        "https://example.com/support?useful=Yes",
        follow_redirects=True,
        verify=mock_tools.file.ssl_context,
    )
    response.headers.get.assert_called_with("content-length")
    response.iter_bytes.assert_called_once_with(chunk_size=1048576)

    # The filename is derived from the URL
    assert filename == mock_tools.base_path / "something.zip"

    # Temporary file was created in download dir and was renamed
    temp_filename = Path(mock_tools.shutil.move.call_args_list[0].args[0])
    assert temp_filename.parent == mock_tools.base_path
    assert temp_filename.name.startswith("something.zip.")
    assert temp_filename.name.endswith(TEMPORARY_DOWNLOAD_FILE_SUFFIX)
    mock_tools.shutil.move.assert_called_with(str(temp_filename), filename)

    # File permissions were set
    mock_tools.os.chmod.assert_called_with(filename, file_perms)
    assert stat.S_IMODE(os.stat(filename).st_mode) == file_perms

    # Attempt to delete Temporary file was made
    mock_tools.os.remove.assert_called_with(str(temp_filename))

    # The downloaded file exists, and content is as expected
    assert filename.exists()
    with (mock_tools.base_path / "something.zip").open(encoding="utf-8") as f:
        assert f.read() == "chunk-1;chunk-2;chunk-3;"


def test_already_downloaded(mock_tools):
    """If the file already exists on disk, it isn't re-downloaded.

    The request is still made to derive the filename, but the content is never streamed.
    """

    # Create an existing file
    existing_file = mock_tools.base_path / "something.zip"
    with existing_file.open("w", encoding="utf-8") as f:
        f.write("existing content")

    url = "https://example.com/path/to/something.zip"

    response = _make_httpx_response(
        status_code=200,
        url=url,
        # Use content and a content-encoding that would cause a DecodeError
        # if ``File`` tried to read the response content.
        # Because the file already exists, ``File`` shouldn't try to read
        # the response, and the ``DecodingError`` won't occur.
        headers={"content-length": "100", "content-encoding": "gzip"},
        stream=[b"definitely not gzip content"],
    )
    mock_tools.httpx.stream.return_value.__enter__.return_value = response

    # Download the file
    filename = mock_tools.file.download(
        url=url,
        download_path=mock_tools.base_path,
    )

    # The GET request will have been made
    mock_tools.httpx.stream.assert_called_with(
        "GET",
        url,
        follow_redirects=True,
        verify=mock_tools.file.ssl_context,
    )

    # The request's Content-Disposition header is consumed to
    # examine the filename; the request is abandoned before
    # any other headers are read.
    response.headers.get.assert_called_once_with("Content-Disposition")

    # but the file existed, so the method returns
    assert filename == existing_file
    assert filename.exists()

    # Temporary file was not created, moved, or deleted
    mock_tools.shutil.move.assert_not_called()
    mock_tools.os.chmod.assert_not_called()
    mock_tools.os.remove.assert_not_called()


def test_missing_resource(mock_tools):
    """MissingNetworkResourceError raises for 404 status code."""
    url = "https://example.com/something.zip?useful=Yes"
    response = _make_httpx_response(
        url=url,
        status_code=404,
        stream=[],
    )

    mock_tools.httpx.stream.return_value.__enter__.return_value = response

    # Download the file
    with pytest.raises(MissingNetworkResourceError):
        mock_tools.file.download(
            url=url,
            download_path=mock_tools.base_path,
        )

    # httpx.stream has been invoked, but nothing else.
    mock_tools.httpx.stream.assert_called_with(
        "GET",
        url,
        follow_redirects=True,
        verify=mock_tools.file.ssl_context,
    )
    response.headers.get.assert_not_called()

    # The file doesn't exist as a result of the download failure
    assert not (mock_tools.base_path / "something.zip").exists()

    # Temporary file was not created, moved, or deleted
    mock_tools.shutil.move.assert_not_called()
    mock_tools.os.chmod.assert_not_called()
    mock_tools.os.remove.assert_not_called()


def test_bad_resource(mock_tools):
    """BadNetworkResourceError raises for non-200 status code."""
    url = "https://example.com/something.zip?useful=Yes"
    response = _make_httpx_response(
        status_code=500,
        url=url,
        stream=[],
    )

    mock_tools.httpx.stream.return_value.__enter__.return_value = response

    # Download the file
    with pytest.raises(BadNetworkResourceError):
        mock_tools.file.download(
            url=url,
            download_path=mock_tools.base_path,
        )

    # httpx.stream has been invoked, but nothing else.
    mock_tools.httpx.stream.assert_called_with(
        "GET",
        url,
        follow_redirects=True,
        verify=mock_tools.file.ssl_context,
    )
    response.headers.get.assert_not_called()

    # The file doesn't exist as a result of the download failure
    assert not (mock_tools.base_path / "something.zip").exists()

    # Temporary file was not created, moved, or deleted
    mock_tools.shutil.move.assert_not_called()
    mock_tools.os.chmod.assert_not_called()
    mock_tools.os.remove.assert_not_called()


def test_iter_bytes_connection_error(mock_tools):
    """NetworkFailure raised if response.iter_bytes() errors and cleans up temporary
    files."""
    url = "https://example.com/something.zip?useful=Yes"
    response = _make_httpx_response(
        status_code=200,
        url=url,
        # Force a real DecodingError by setting the response encoding to
        # gzip with response content that is _not_ gzip
        headers={"content-length": "100", "content-encoding": "gzip"},
        stream=[b"definitely not gzip content"],
    )
    mock_tools.httpx.stream.return_value.__enter__.return_value = response

    # Download the file
    with pytest.raises(NetworkFailure, match="Unable to download something.zip"):
        mock_tools.file.download(
            url="https://example.com/something.zip?useful=Yes",
            download_path=mock_tools.base_path,
        )

    # httpx.stream has been invoked, but nothing else.
    mock_tools.httpx.stream.assert_called_with(
        "GET",
        "https://example.com/something.zip?useful=Yes",
        follow_redirects=True,
        verify=mock_tools.file.ssl_context,
    )
    response.headers.get.assert_called_with("content-length")

    # The file doesn't exist as a result of the download failure
    assert not (mock_tools.base_path / "something.zip").exists()

    # Temporary file was not moved
    mock_tools.shutil.move.assert_not_called()
    mock_tools.os.chmod.assert_not_called()

    # Failure happens during response streaming, so the temporary file is create
    # but then also correctly cleaned up after the failure
    temp_filename = Path(mock_tools.os.remove.call_args_list[0].args[0])
    assert temp_filename.parent == mock_tools.base_path
    assert temp_filename.name.startswith("something.zip.")
    assert temp_filename.name.endswith(TEMPORARY_DOWNLOAD_FILE_SUFFIX)
    mock_tools.os.remove.assert_called_with(str(temp_filename))


def test_connection_error(mock_tools):
    """NetworkFailure raised if the connection fails."""
    # Use ftp scheme to force raising a real ProtocolError without needing to mock
    url = "ftp://example.com/something.zip"

    # Use the real httpx for this test instead of the MagicMock'd one from mock_tools
    # Keep using the fixture though, so that it still gets cleaned up after the test
    mock_tools.httpx = mock.Mock(wraps=httpx)

    # Failure leads to filename never being read, so the error message will use the full URL
    # rather than the filename
    with pytest.raises(NetworkFailure, match=f"Unable to download {url}"):
        mock_tools.file.download(
            url=url,
            download_path=mock_tools.base_path,
        )

    # httpx.stream has been invoked, but nothing else.
    mock_tools.httpx.stream.assert_called_with(
        "GET",
        url,
        follow_redirects=True,
        verify=mock_tools.file.ssl_context,
    )

    # The file doesn't exist as a result of the download failure
    assert not (mock_tools.base_path / "something.zip").exists()

    # Temporary file was never created because the failure happens before a response
    # is every received
    assert not (mock_tools.base_path / "something.zip.download").exists()

    # Temporary file was not created, moved, or deleted
    mock_tools.shutil.move.assert_not_called()
    mock_tools.os.chmod.assert_not_called()
    mock_tools.os.remove.assert_not_called()


def test_redirect_connection_error(mock_tools):
    """NetworkFailure raises if the request leads to too many redirects."""
    mock_tools.httpx.stream.side_effect = [
        httpx.TooManyRedirects("Exceeded max redirects")
    ]

    # Download the file
    with pytest.raises(
        NetworkFailure,
        match=r"Unable to download https\:\/\/example.com\/something\.zip\?useful=Yes",
    ):
        mock_tools.file.download(
            url="https://example.com/something.zip?useful=Yes",
            download_path=mock_tools.base_path,
        )

    # httpx.stream has been invoked, but nothing else.
    mock_tools.httpx.stream.assert_called_with(
        "GET",
        "https://example.com/something.zip?useful=Yes",
        follow_redirects=True,
        verify=mock_tools.file.ssl_context,
    )

    # The file doesn't exist as a result of the download failure
    assert not (mock_tools.base_path / "something.zip").exists()

    # Temporary file was not created, moved, or deleted
    mock_tools.shutil.move.assert_not_called()
    mock_tools.os.chmod.assert_not_called()
    mock_tools.os.remove.assert_not_called()


def test_ssl_verification_error(mock_tools):
    """NetworkFailure is raised if the request fails due to SSL."""
    # Mock an SSL verification error
    error = httpx.ConnectError("connection error")
    error.__context__ = httpcore.ConnectError()
    error.__context__.__context__ = ssl.SSLCertVerificationError()
    mock_tools.httpx.stream.side_effect = [error]

    # Download the file
    with pytest.raises(
        NetworkFailure,
        match=(
            r"a connection to the server could not be established "
            r"due to a certificate verification problem"
        ),
    ):
        mock_tools.file.download(
            url="https://example.com/something.zip",
            download_path=mock_tools.base_path,
        )

    # httpx.stream has been invoked, but nothing else.
    mock_tools.httpx.stream.assert_called_with(
        "GET",
        "https://example.com/something.zip",
        follow_redirects=True,
        verify=mock_tools.file.ssl_context,
    )

    # The file doesn't exist as a result of the download failure
    assert not (mock_tools.base_path / "something.zip").exists()

    # Temporary file was not created, moved, or deleted
    mock_tools.shutil.move.assert_not_called()
    mock_tools.os.chmod.assert_not_called()
    mock_tools.os.remove.assert_not_called()


def test_unknown_httpcore_connectionerror(mock_tools):
    """NetworkFailure is raised if an unknown core connection error occurs."""
    # Mock a connection error at the level of httpcore
    error = httpx.ConnectError("connection error")
    error.__context__ = httpcore.ConnectError()
    mock_tools.httpx.stream.side_effect = [error]

    # Download the file
    with pytest.raises(
        NetworkFailure,
        match=(
            r"a connection to the server could not be established.\n\n"
            r"The reported cause of the problem "
        ),
    ):
        mock_tools.file.download(
            url="https://example.com/something.zip",
            download_path=mock_tools.base_path,
        )

    # httpx.stream has been invoked, but nothing else.
    mock_tools.httpx.stream.assert_called_with(
        "GET",
        "https://example.com/something.zip",
        follow_redirects=True,
        verify=mock_tools.file.ssl_context,
    )

    # The file doesn't exist as a result of the download failure
    assert not (mock_tools.base_path / "something.zip").exists()

    # Temporary file was not created, moved, or deleted
    mock_tools.shutil.move.assert_not_called()
    mock_tools.os.chmod.assert_not_called()
    mock_tools.os.remove.assert_not_called()


def test_unknown_httpx_connectionerror(mock_tools):
    """NetworkFailure is raised if an unknown httpx connection error occurs."""
    # Mock a connection error at the level of httpx
    error = httpx.ConnectError("connection error")
    mock_tools.httpx.stream.side_effect = [error]

    # Download the file
    with pytest.raises(
        NetworkFailure,
        match=(
            r"a connection to the server could not be established.\n\n"
            r"The reported cause of the problem "
        ),
    ):
        mock_tools.file.download(
            url="https://example.com/something.zip",
            download_path=mock_tools.base_path,
        )

    # httpx.stream has been invoked, but nothing else.
    mock_tools.httpx.stream.assert_called_with(
        "GET",
        "https://example.com/something.zip",
        follow_redirects=True,
        verify=mock_tools.file.ssl_context,
    )

    # The file doesn't exist as a result of the download failure
    assert not (mock_tools.base_path / "something.zip").exists()

    # Temporary file was not created, moved, or deleted
    mock_tools.shutil.move.assert_not_called()
    mock_tools.os.chmod.assert_not_called()
    mock_tools.os.remove.assert_not_called()