import mock
import os
import re

from distutils import log
from distutils.errors import DistutilsError
from distutils.version import StrictVersion

import pytest

from setuptools.command.upload import upload
from setuptools.dist import Distribution
from setuptools.extern import six


def _parse_upload_body(body):
    boundary = u'\r\n----------------GHSKFJDLGDS7543FJKLFHRE75642756743254'
    entries = []
    name_re = re.compile(u'^Content-Disposition: form-data; name="([^\"]+)"')

    for entry in body.split(boundary):
        pair = entry.split(u'\r\n\r\n')
        if not len(pair) == 2:
            continue

        key, value = map(six.text_type.strip, pair)
        m = name_re.match(key)
        if m is not None:
            key = m.group(1)

        entries.append((key, value))

    return entries


@pytest.fixture
def patched_upload(tmpdir):
    class Fix:
        def __init__(self, cmd, urlopen):
            self.cmd = cmd
            self.urlopen = urlopen

        def __iter__(self):
            return iter((self.cmd, self.urlopen))

        def get_uploaded_metadata(self):
            request = self.urlopen.call_args_list[0][0][0]
            body = request.data.decode('utf-8')
            entries = dict(_parse_upload_body(body))

            return entries

    class ResponseMock(mock.Mock):
        def getheader(self, name, default=None):
            """Mocked getheader method for response object"""
            return {
                'content-type': 'text/plain; charset=utf-8',
            }.get(name.lower(), default)

    with mock.patch('setuptools.command.upload.urlopen') as urlopen:
        urlopen.return_value = ResponseMock()
        urlopen.return_value.getcode.return_value = 200
        urlopen.return_value.read.return_value = b''

        content = os.path.join(str(tmpdir), "content_data")

        with open(content, 'w') as f:
            f.write("Some content")

        dist = Distribution()
        dist.dist_files = [('sdist', '3.7.0', content)]

        cmd = upload(dist)
        cmd.announce = mock.Mock()
        cmd.username = 'user'
        cmd.password = 'hunter2'

        yield Fix(cmd, urlopen)


class TestUploadTest:
    def test_upload_metadata(self, patched_upload):
        cmd, patch = patched_upload

        # Set the metadata version to 2.1
        cmd.distribution.metadata.metadata_version = '2.1'

        # Run the command
        cmd.ensure_finalized()
        cmd.run()

        # Make sure we did the upload
        patch.assert_called_once()

        # Make sure the metadata version is correct in the headers
        entries = patched_upload.get_uploaded_metadata()
        assert entries['metadata_version'] == '2.1'

    def test_warns_deprecation(self):
        dist = Distribution()
        dist.dist_files = [(mock.Mock(), mock.Mock(), mock.Mock())]

        cmd = upload(dist)
        cmd.upload_file = mock.Mock()
        cmd.announce = mock.Mock()

        cmd.run()

        cmd.announce.assert_called_once_with(
            "WARNING: Uploading via this command is deprecated, use twine to "
            "upload instead (https://pypi.org/p/twine/)",
            log.WARN
        )

    def test_warns_deprecation_when_raising(self):
        dist = Distribution()
        dist.dist_files = [(mock.Mock(), mock.Mock(), mock.Mock())]

        cmd = upload(dist)
        cmd.upload_file = mock.Mock()
        cmd.upload_file.side_effect = Exception
        cmd.announce = mock.Mock()

        with pytest.raises(Exception):
            cmd.run()

        cmd.announce.assert_called_once_with(
            "WARNING: Uploading via this command is deprecated, use twine to "
            "upload instead (https://pypi.org/p/twine/)",
            log.WARN
        )

    @pytest.mark.parametrize('url', [
        'https://example.com/a;parameter',    # Has parameters
        'https://example.com/a?query',        # Has query
        'https://example.com/a#fragment',     # Has fragment
        'ftp://example.com',                  # Invalid scheme

    ])
    def test_upload_file_invalid_url(self, url, patched_upload):
        patched_upload.urlopen.side_effect = Exception("Should not be reached")

        cmd = patched_upload.cmd
        cmd.repository = url

        cmd.ensure_finalized()
        with pytest.raises(AssertionError):
            cmd.run()

    def test_upload_file_http_error(self, patched_upload):
        patched_upload.urlopen.side_effect = six.moves.urllib.error.HTTPError(
            'https://example.com',
            404,
            'File not found',
            None,
            None
        )

        cmd = patched_upload.cmd
        cmd.ensure_finalized()

        with pytest.raises(DistutilsError):
            cmd.run()

        cmd.announce.assert_any_call(
            'Upload failed (404): File not found',
            log.ERROR)

    def test_upload_file_os_error(self, patched_upload):
        patched_upload.urlopen.side_effect = OSError("Invalid")

        cmd = patched_upload.cmd
        cmd.ensure_finalized()

        with pytest.raises(OSError):
            cmd.run()

        cmd.announce.assert_any_call('Invalid', log.ERROR)

    @mock.patch('setuptools.command.upload.spawn')
    def test_upload_file_gpg(self, spawn, patched_upload):
        cmd, urlopen = patched_upload

        cmd.sign = True
        cmd.identity = "Alice"
        cmd.dry_run = True
        content_fname = cmd.distribution.dist_files[0][2]
        signed_file = content_fname + '.asc'

        with open(signed_file, 'wb') as f:
            f.write("signed-data".encode('utf-8'))

        cmd.ensure_finalized()
        cmd.run()

        # Make sure that GPG was called
        spawn.assert_called_once_with([
            "gpg", "--detach-sign", "--local-user", "Alice", "-a",
            content_fname
        ], dry_run=True)

        # Read the 'signed' data that was transmitted
        entries = patched_upload.get_uploaded_metadata()
        assert entries['gpg_signature'] == 'signed-data'

    def test_show_response_no_error(self, patched_upload):
        # This test is just that show_response doesn't throw an error
        # It is not really important what the printed response looks like
        # in a deprecated command, but we don't want to introduce new
        # errors when importing this function from distutils

        patched_upload.cmd.show_response = True
        patched_upload.cmd.ensure_finalized()
        patched_upload.cmd.run()