import mock
import os
import re

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

import pytest
import six

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


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'

    @pytest.mark.parametrize('bdist', ['bdist_rpm', 'bdist_dumb'])
    @mock.patch('setuptools.command.upload.platform')
    def test_bdist_rpm_upload(self, platform, bdist, patched_upload):
        # Set the upload command to include bdist_rpm
        cmd = patched_upload.cmd
        dist_files = cmd.distribution.dist_files
        dist_files = [(bdist,) + dist_files[0][1:]]
        cmd.distribution.dist_files = dist_files

        # Mock out the platform commands to make this platform-independent
        platform.dist.return_value = ('redhat', '', '')

        cmd.ensure_finalized()
        cmd.run()

        entries = patched_upload.get_uploaded_metadata()

        assert entries['comment'].startswith(u'built for')
        assert len(entries['comment']) > len(u'built for')