"""Tests for transfer.py."""
import string

import mock
import six
from six.moves import http_client
import unittest2

from apitools.base.py import base_api
from apitools.base.py import http_wrapper
from apitools.base.py import transfer


class TransferTest(unittest2.TestCase):

    def assertRangeAndContentRangeCompatible(self, request, response):
        request_prefix = 'bytes='
        self.assertIn('range', request.headers)
        self.assertTrue(request.headers['range'].startswith(request_prefix))
        request_range = request.headers['range'][len(request_prefix):]

        response_prefix = 'bytes '
        self.assertIn('content-range', response.info)
        response_header = response.info['content-range']
        self.assertTrue(response_header.startswith(response_prefix))
        response_range = (
            response_header[len(response_prefix):].partition('/')[0])

        msg = ('Request range ({0}) not a prefix of '
               'response_range ({1})').format(
                   request_range, response_range)
        self.assertTrue(response_range.startswith(request_range), msg=msg)

    def testComputeEndByte(self):
        total_size = 100
        chunksize = 10
        download = transfer.Download.FromStream(
            six.StringIO(), chunksize=chunksize, total_size=total_size)
        self.assertEqual(chunksize - 1,
                         download._Download__ComputeEndByte(0, end=50))

    def testComputeEndByteReturnNone(self):
        download = transfer.Download.FromStream(six.StringIO())
        self.assertIsNone(
            download._Download__ComputeEndByte(0, use_chunks=False))

    def testComputeEndByteNoChunks(self):
        total_size = 100
        download = transfer.Download.FromStream(
            six.StringIO(), chunksize=10, total_size=total_size)
        for end in (None, 1000):
            self.assertEqual(
                total_size - 1,
                download._Download__ComputeEndByte(0, end=end,
                                                   use_chunks=False),
                msg='Failed on end={0}'.format(end))

    def testComputeEndByteNoTotal(self):
        download = transfer.Download.FromStream(six.StringIO())
        default_chunksize = download.chunksize
        for chunksize in (100, default_chunksize):
            download.chunksize = chunksize
            for start in (0, 10):
                self.assertEqual(
                    download.chunksize + start - 1,
                    download._Download__ComputeEndByte(start),
                    msg='Failed on start={0}, chunksize={1}'.format(
                        start, chunksize))

    def testComputeEndByteSmallTotal(self):
        total_size = 100
        download = transfer.Download.FromStream(six.StringIO(),
                                                total_size=total_size)
        for start in (0, 10):
            self.assertEqual(total_size - 1,
                             download._Download__ComputeEndByte(start),
                             msg='Failed on start={0}'.format(start))

    def testGetRange(self):
        for (start_byte, end_byte) in [(0, 25), (5, 15), (0, 0), (25, 25)]:
            bytes_http = object()
            http = object()
            download_stream = six.StringIO()
            download = transfer.Download.FromStream(download_stream,
                                                    total_size=26,
                                                    auto_transfer=False)
            download.bytes_http = bytes_http
            base_url = 'https://part.one/'
            with mock.patch.object(http_wrapper, 'MakeRequest',
                                   autospec=True) as make_request:
                make_request.return_value = http_wrapper.Response(
                    info={
                        'content-range': 'bytes %d-%d/26' %
                                         (start_byte, end_byte),
                        'status': http_client.OK,
                    },
                    content=string.ascii_lowercase[start_byte:end_byte+1],
                    request_url=base_url,
                )
                request = http_wrapper.Request(url='https://part.one/')
                download.InitializeDownload(request, http=http)
                download.GetRange(start_byte, end_byte)
                self.assertEqual(1, make_request.call_count)
                received_request = make_request.call_args[0][1]
                self.assertEqual(base_url, received_request.url)
                self.assertRangeAndContentRangeCompatible(
                    received_request, make_request.return_value)

    def testNonChunkedDownload(self):
        bytes_http = object()
        http = object()
        download_stream = six.StringIO()
        download = transfer.Download.FromStream(download_stream, total_size=52)
        download.bytes_http = bytes_http
        base_url = 'https://part.one/'

        with mock.patch.object(http_wrapper, 'MakeRequest',
                               autospec=True) as make_request:
            make_request.return_value = http_wrapper.Response(
                info={
                    'content-range': 'bytes 0-51/52',
                    'status': http_client.OK,
                },
                content=string.ascii_lowercase * 2,
                request_url=base_url,
            )
            request = http_wrapper.Request(url='https://part.one/')
            download.InitializeDownload(request, http=http)
            self.assertEqual(1, make_request.call_count)
            received_request = make_request.call_args[0][1]
            self.assertEqual(base_url, received_request.url)
            self.assertRangeAndContentRangeCompatible(
                received_request, make_request.return_value)
            download_stream.seek(0)
            self.assertEqual(string.ascii_lowercase * 2,
                             download_stream.getvalue())

    def testChunkedDownload(self):
        bytes_http = object()
        http = object()
        download_stream = six.StringIO()
        download = transfer.Download.FromStream(
            download_stream, chunksize=26, total_size=52)
        download.bytes_http = bytes_http

        # Setting autospec on a mock with an iterable side_effect is
        # currently broken (http://bugs.python.org/issue17826), so
        # instead we write a little function.
        def _ReturnBytes(unused_http, http_request,
                         *unused_args, **unused_kwds):
            url = http_request.url
            if url == 'https://part.one/':
                return http_wrapper.Response(
                    info={
                        'content-location': 'https://part.two/',
                        'content-range': 'bytes 0-25/52',
                        'status': http_client.PARTIAL_CONTENT,
                    },
                    content=string.ascii_lowercase,
                    request_url='https://part.one/',
                )
            elif url == 'https://part.two/':
                return http_wrapper.Response(
                    info={
                        'content-range': 'bytes 26-51/52',
                        'status': http_client.OK,
                    },
                    content=string.ascii_uppercase,
                    request_url='https://part.two/',
                )
            else:
                self.fail('Unknown URL requested: %s' % url)

        with mock.patch.object(http_wrapper, 'MakeRequest',
                               autospec=True) as make_request:
            make_request.side_effect = _ReturnBytes
            request = http_wrapper.Request(url='https://part.one/')
            download.InitializeDownload(request, http=http)
            self.assertEqual(2, make_request.call_count)
            for call in make_request.call_args_list:
                self.assertRangeAndContentRangeCompatible(
                    call[0][1], _ReturnBytes(*call[0]))
            download_stream.seek(0)
            self.assertEqual(string.ascii_lowercase + string.ascii_uppercase,
                             download_stream.getvalue())

    def testFromEncoding(self):
        # Test a specific corner case in multipart encoding.

        # Python's mime module by default encodes lines that start with
        # "From " as ">From ", which we need to make sure we don't run afoul
        # of when sending content that isn't intended to be so encoded. This
        # test calls out that we get this right. We test for both the
        # multipart and non-multipart case.
        multipart_body = '{"body_field_one": 7}'
        upload_contents = 'line one\nFrom \nline two'
        upload_config = base_api.ApiUploadInfo(
            accept=['*/*'],
            max_size=None,
            resumable_multipart=True,
            resumable_path=u'/resumable/upload',
            simple_multipart=True,
            simple_path=u'/upload',
        )
        url_builder = base_api._UrlBuilder('http://www.uploads.com')

        # Test multipart: having a body argument in http_request forces
        # multipart here.
        upload = transfer.Upload.FromStream(
            six.StringIO(upload_contents),
            'text/plain',
            total_size=len(upload_contents))
        http_request = http_wrapper.Request(
            'http://www.uploads.com',
            headers={'content-type': 'text/plain'},
            body=multipart_body)
        upload.ConfigureRequest(upload_config, http_request, url_builder)
        self.assertEqual(url_builder.query_params['uploadType'], 'multipart')
        rewritten_upload_contents = '\n'.join(
            http_request.body.split('--')[2].splitlines()[1:])
        self.assertTrue(rewritten_upload_contents.endswith(upload_contents))

        # Test non-multipart (aka media): no body argument means this is
        # sent as media.
        upload = transfer.Upload.FromStream(
            six.StringIO(upload_contents),
            'text/plain',
            total_size=len(upload_contents))
        http_request = http_wrapper.Request(
            'http://www.uploads.com',
            headers={'content-type': 'text/plain'})
        upload.ConfigureRequest(upload_config, http_request, url_builder)
        self.assertEqual(url_builder.query_params['uploadType'], 'media')
        rewritten_upload_contents = http_request.body
        self.assertTrue(rewritten_upload_contents.endswith(upload_contents))