def test_http_connection(self):
     url = 'http://www.test.com'
     _junk, conn = c.http_connection(url)
     self.assertTrue(isinstance(conn, c.HTTPConnection))
     url = 'https://www.test.com'
     _junk, conn = c.http_connection(url)
     self.assertTrue(isinstance(conn, c.HTTPConnection))
     url = 'ftp://www.test.com'
     self.assertRaises(c.ClientException, c.http_connection, url)

    def __init__(self, conn, container, name, content_type=None):

        parsed, self.conn = http_connection(conn.url)

        logging.debug("ChunkObject: new connection open (%r, %r)" % (parsed, self.conn))

        path = '%s/%s/%s' % (parsed.path.rstrip('/'),
                             quote(smart_str(container)),
                             quote(smart_str(name)),
                             )
        headers = { 'X-Auth-Token': conn.token,
                    'Content-Type': content_type or 'application/octet-stream',
                    'Transfer-Encoding': 'chunked',
                    'Connection': 'close',
                    # User-Agent ?
                    }
        if conn.real_ip:
            headers['X-Forwarded-For'] = conn.real_ip

        # we can't use the generator interface offered by requests to do a
        # chunked transfer encoded PUT, so we do this is to get control over the
        # "real" http connection and do the HTTP request ourselves
        self.raw_conn = self.conn.request_session.get_adapter(conn.url).get_connection(conn.url)._get_conn()

        self.raw_conn.putrequest('PUT', path, skip_accept_encoding=True)
        for key, value in headers.iteritems():
            self.raw_conn.putheader(key, value)
        self.raw_conn.endheaders()
        logging.debug("ChunkedObject: path=%r, headers=%r" % (path, headers))

        self.already_sent = 0

def create_container(request):
    """ Creates a container (empty object of type application/directory) """

    storage_url = get_endpoint(request, 'adminURL')
    auth_token = get_token_id(request)
    http_conn = client.http_connection(storage_url,
                                       insecure=settings.SWIFT_INSECURE)

    form = CreateContainerForm(request.POST or None)
    if form.is_valid():
        container = form.cleaned_data['containername']
        try:
            client.put_container(storage_url,
                                 auth_token,
                                 container,
                                 http_conn=http_conn)
            messages.add_message(request, messages.SUCCESS,
                                 _("Container created."))

            actionlog.log(request.user.username, "create", container)
        except client.ClientException as err:
            log.exception('{}{}'.format(_('Exception:').encode('UTF-8'), err))
            messages.add_message(request, messages.ERROR, _('Access denied.'))

        return redirect(containerview)

    context = utils.update_default_context(request, {
        'form': form,
    })

    return render_to_response('create_container.html', context,
                              context_instance=RequestContext(request))

    def connect(self, force=False):
        if self.http_conn is not None and not force:
            return

        swift_url, self.token = swift.get_auth(
            auth_url=self.auth_url,
            user=self.username,
            key=self.password,
            auth_version=self.auth_ver,
            tenant_name=self.tenant,
        )
        if self.debug:
            print(self.auth_url)
            print(self.token)
            print(self.swift_url)
            print(swift_url)
            print
        if not swift_url == self.swift_url:
            print("Different swift_url returned from swift")

        self.http_conn = swift.http_connection(self.swift_url)

        if self.debug:
            print(self.http_conn)
            print

def containerview(request):
    """ Returns a list of all containers in current account. """

    if not request.session.get('project_id'):
        messages.add_message(request, messages.ERROR, _('Select a project'))
        return HttpResponseRedirect(reverse('dashboard'))

    storage_url = get_endpoint(request, 'adminURL')
    auth_token = get_token_id(request)
    http_conn = client.http_connection(storage_url,
                                       insecure=settings.SWIFT_INSECURE)

    page = request.GET.get('page', 1)

    try:
        account_stat, containers = client.get_account(storage_url, auth_token,
                                                      http_conn=http_conn)
    except client.ClientException as err:
        log.exception('{}{}'.format(_('Exception:').encode('UTF-8'), err))
        messages.add_message(request, messages.ERROR,
                             _('Unable to list containers'))

        account_stat = {}
        containers = []

    containers = _hide_containers_with_prefixes(containers)
    account_stat = replace_hyphens(account_stat)

    context = utils.update_default_context(request, {
        'account_stat': account_stat,
        'containers': utils.generic_pagination(containers, page),
    })

    return render_to_response('containerview.html', context,
                              context_instance=RequestContext(request))

    def test_raw_upload(self):
        # Raw upload happens when content_length is passed to put_object
        conn = c.http_connection(u'http://www.test.com/')
        resp = MockHttpResponse(status=200)
        conn[1].getresponse = resp.fake_response
        conn[1]._request = resp._fake_request
        astring = 'asdf'
        astring_len = len(astring)
        mock_file = six.StringIO(astring)

        c.put_object(url='http://www.test.com', http_conn=conn,
                     contents=mock_file, content_length=astring_len)
        self.assertTrue(isinstance(resp.requests_params['data'],
                                   swiftclient.utils.LengthWrapper))
        self.assertEqual(astring_len,
                         len(resp.requests_params['data'].read()))

        mock_file = six.StringIO(astring)
        c.put_object(url='http://www.test.com', http_conn=conn,
                     headers={'Content-Length': str(astring_len)},
                     contents=mock_file)
        self.assertTrue(isinstance(resp.requests_params['data'],
                                   swiftclient.utils.LengthWrapper))
        self.assertEqual(astring_len,
                         len(resp.requests_params['data'].read()))

def objectview(request, container, prefix=None):
    """ Returns list of all objects in current container. """

    storage_url = get_endpoint(request, 'adminURL')
    public_url = get_endpoint(request, 'publicURL') + '/' + container
    auth_token = get_token_id(request)
    http_conn = client.http_connection(storage_url,
                                       insecure=settings.SWIFT_INSECURE)

    page = request.GET.get('page', 1)

    try:
        _, objects = client.get_container(storage_url, auth_token,
                                             container, delimiter='/',
                                             prefix=prefix,
                                             http_conn=http_conn)
    except client.ClientException as err:
        log.exception('{}{}'.format(ugettext('Exception:').encode('UTF-8'), err))
        messages.add_message(request, messages.ERROR, ugettext('Access denied.'))
        return redirect(containerview)

    prefixes = prefix_list(prefix)
    object_list = pseudofolder_object_list(objects, prefix, public_url)
    context = utils.update_default_context(request, {
        'container': container,
        'objects': utils.generic_pagination(object_list, page),
        'prefix': prefix,
        'prefixes': prefixes,
    })

    return render_to_response("objectview.html", context,
                              context_instance=RequestContext(request))

def delete_container(request, container, force=True):
    """
    Deletes a container. If force is True, it will deletes all objects first.
    """

    storage_url = get_endpoint(request, 'adminURL')
    auth_token = get_token_id(request)
    http_conn = client.http_connection(storage_url,
                                       insecure=settings.SWIFT_INSECURE)

    if force:
        try:
            _, objects = client.get_container(storage_url,
                                           auth_token,
                                           container,
                                           http_conn=http_conn)

        except client.ClientException as err:
            log.exception('{}{}'.format(ugettext('Exception:').encode('UTF-8'), err))
            return False

        for obj in objects:
            delete_object(request=request,
                          container=container,
                          objectname=obj['name'])

    try:
        client.delete_container(storage_url, auth_token,
                                container, http_conn=http_conn)
        actionlog.log(request.user.username, "delete", container)
    except client.ClientException as err:
        log.exception('{}{}'.format(ugettext('Exception:').encode('UTF-8'), err))
        return False

    return True

    def __init__(self, conn, container, name, content_type=None, reuse_token = True):
        self.raw_conn = None

        if reuse_token:
            self.url = conn.url
            token = conn.token
        else:
            self.url, token = conn.get_auth()
        self.parsed, self.conn = http_connection(self.url)

        self.path = '%s/%s/%s' % (self.parsed.path.rstrip('/'),
                                  quote(smart_str(container)),
                                  quote(smart_str(name)),
                                  )
        self.headers = { 'X-Auth-Token': token,
                         'Content-Type': content_type or 'application/octet-stream',
                         'Transfer-Encoding': 'chunked',
                         'Connection': 'close',
                         # User-Agent ?
                         }
        if conn.real_ip:
            self.headers['X-Forwarded-For'] = conn.real_ip

        logging.debug("ChunkedObject: path=%r, headers=%r" % (self.path, self.headers))

        self.already_sent = 0

    def test_unicode_ok(self):
        conn = c.http_connection(u'http://www.test.com/')
        args = (u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91',
                u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91',
                u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91',
                u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91')
        text = u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91'
        headers = {'X-Header1': text,
                   b'X-Header2': 'value',
                   'X-2': '1', 'X-3': {'a': 'b'}, 'a-b': '.x:yz mn:kl:qr',
                   'X-Object-Meta-Header-not-encoded': text,
                   b'X-Object-Meta-Header-encoded': 'value'}

        resp = MockHttpResponse()
        conn[1].getresponse = resp.fake_response
        conn[1]._request = resp._fake_request
        c.post_object(*args, headers=headers, http_conn=conn)
        # Test for RFC-2616 encoded symbols
        self.assertIn(('a-b', b".x:yz mn:kl:qr"), resp.buffer)
        # Test unicode header
        self.assertIn(('x-header1', text.encode('utf8')),
                      resp.buffer)
        self.assertIn((b'x-object-meta-header-not-encoded',
                      text.encode('utf8')), resp.buffer)
        self.assertIn((b'x-object-meta-header-encoded', b'value'),
                      resp.buffer)
        self.assertIn((b'x-header2', b'value'), resp.buffer)

def object_versioning(request, container, prefix=None):
    storage_url = get_endpoint(request, 'adminURL')
    auth_token = get_token_id(request)
    public_url = get_endpoint(request, 'publicURL') + '/' + container
    http_conn = client.http_connection(storage_url,
                                       insecure=settings.SWIFT_INSECURE)

    objects = []

    page = request.GET.get('page', 1)

    if request.method == 'GET':
        headers = client.head_container(storage_url,
                                auth_token,
                                container,
                                http_conn=http_conn)

        version_location = headers.get('x-versions-location', None)

        if version_location:
            try:
                _, objects = client.get_container(storage_url,
                                                  auth_token,
                                                  version_location,
                                                  prefix=prefix,
                                                  delimiter='/',
                                                  http_conn=http_conn)
            except client.ClientException:
                pass

        prefixes = prefix_list(prefix)
        object_list = pseudofolder_object_list(objects, prefix, public_url)

        context = utils.update_default_context(request, {
            'container': container,
            'objects': utils.generic_pagination(object_list, page),
            'version_location': version_location,
            'prefix': prefix,
            'prefixes': prefixes,
        })

        return render_to_response('container_versioning.html',
                                  dictionary=context,
                                  context_instance=RequestContext(request))

    if request.method == 'POST':

        action = request.POST.get('action', None)

        if action == 'enable':
            enable_versioning(request, container)
            actionlog.log(request.user.username, "enable", 'Versioning. Container: %s' % container)
        elif action == 'disable':
            disable_versioning(request, container)
            actionlog.log(request.user.username, "disable", 'Versioning. Container: %s' % container)
        else:
            messages.add_message(request, messages.ERROR, ugettext('Action is required.'))

        return redirect(object_versioning, container=container)

def get_info(url):
    parts = urlparse(url)
    url = parts.scheme + '://' + parts.netloc + '/info'
    http_conn = client.http_connection(url)
    try:
        return client.get_capabilities(http_conn)
    except client.ClientException:
        raise unittest.SkipTest('Unable to retrieve cluster info')

    def test_no_content_type(self):
        conn = c.http_connection(u'http://www.test.com/')
        resp = MockHttpResponse(status=200)
        conn[1].getresponse = resp.fake_response
        conn[1]._request = resp._fake_request

        c.put_object(url='http://www.test.com', http_conn=conn)
        request_header = resp.requests_params['headers']
        self.assertEqual(request_header['content-type'], b'')

    def __init__(self, environ, path):
        dav_provider.DAVCollection.__init__(self, '/', environ)
        self.path = path

        self.auth_token = self.environ.get('swift_auth_token')
        self.storage_url = self.environ.get('swift_storage_url')
        self.http_connection = client.http_connection(
            self.storage_url,
            insecure=self.environ.get('insecure'))

    def close(self):

        if not self.closed and hasattr(self, 'request_body'):

            try:

                self.request_body.write(self.CRLF)
                self.request_body.write('--' + self.BOUNDARY + '--')
                self.request_body.write(self.CRLF)

                length = self.request_body.tell()

                try:

                    parsed_url, connection = http_connection(self.url)

                    if parsed_url.query:
                        url_path = parsed_url.path + '?' + parsed_url.query
                    else:
                        url_path = parsed_url.path

                    connection.putrequest('POST', url_path)

                    connection.putheader('Content-Type', 'multipart/form-data; boundary=%s' % self.BOUNDARY)
                    connection.putheader('Content-Length', str(length))

                    if self.password:
                        credentials = base64.b64encode(b'%s:%s' % (self.username, self.password))
                        connection.putheader('Authorization', 'Basic %s' % credentials)

                    connection.endheaders()

                    self.request_body.seek(0)
                    connection.send(self.request_body)

                    response = connection.getresponse()

                except Exception as error:
                    raise UnexpectedHTTPResponse('%s: %s' % (error.__class__, error))

                if response.status // 100 != 2:

                    message = '%d: %s' % (response.status, response.reason)

                    if response.getheader('Content-Type') == 'text/plain':
                        try:
                            response_content = response.read(100).splitlines()[0]
                            message = '%s. %s' % (message, response_content)
                        except Exception:
                            pass

                    raise UnexpectedHTTPResponse(message)

            finally:
                self.closed = True
                self.request_body.close()

    def test_params(self):
        conn = c.http_connection(u"http://www.test.com/")
        resp = MockHttpResponse(status=200)
        conn[1].getresponse = resp.fake_response
        conn[1]._request = resp._fake_request

        c.put_object(url="http://www.test.com", http_conn=conn, etag="1234-5678", content_type="text/plain")
        request_header = resp.requests_params["headers"]
        self.assertTrue(request_header["etag"], "1234-5678")
        self.assertTrue(request_header["content-type"], "text/plain")

    def test_params(self):
        conn = c.http_connection(u'http://www.test.com/')
        resp = MockHttpResponse(status=200)
        conn[1].getresponse = resp.fake_response
        conn[1]._request = resp._fake_request

        c.put_object(url='http://www.test.com', http_conn=conn,
                     etag='1234-5678', content_type='text/plain')
        request_header = resp.requests_params['headers']
        self.assertEqual(request_header['etag'], b'1234-5678')
        self.assertEqual(request_header['content-type'], b'text/plain')

    def test_set_user_agent_default(self):
        _junk, conn = c.http_connection('http://www.example.com')
        req_headers = {}

        def my_request_handler(*a, **kw):
            req_headers.update(kw.get('headers', {}))
        conn._request = my_request_handler

        # test the default
        conn.request('GET', '/')
        ua = req_headers.get('user-agent', 'XXX-MISSING-XXX')
        self.assertTrue(ua.startswith('python-swiftclient-'))

def upload(request, container, prefix=None):
    """ Display upload form using swift formpost """

    storage_url = get_endpoint(request, 'adminURL')
    auth_token = get_token_id(request)
    http_conn = client.http_connection(storage_url,
                                       insecure=settings.SWIFT_INSECURE)

    redirect_url = 'http://{}'.format(request.get_host())
    redirect_url += reverse('objectview', kwargs={'container': container, })

    swift_url = storage_url + '/' + container + '/'
    if prefix:
        swift_url += prefix
        redirect_url += prefix

    redirect_url += '?p={}'.format(request.session.get('project_id'))

    url_parts = urlparse(swift_url)
    path = url_parts.path

    max_file_size = 5 * 1024 * 1024 * 1024
    max_file_count = 1
    expires = int(time.time() + 15 * 60)
    key = get_temp_key(storage_url, auth_token, http_conn)

    if not key:
        messages.add_message(request, messages.ERROR, _('Access denied.'))
        if prefix:
            return redirect(objectview, container=container, prefix=prefix)
        else:
            return redirect(objectview, container=container)

    hmac_body = '%s\n%s\n%s\n%s\n%s' % (path, redirect_url, max_file_size,
                                        max_file_count, expires)
    signature = hmac.new(key, hmac_body, sha1).hexdigest()

    prefixes = prefix_list(prefix)

    context = utils.update_default_context(request, {
        'swift_url': swift_url,
        'redirect_url': redirect_url,
        'max_file_size': max_file_size,
        'max_file_count': max_file_count,
        'expires': expires,
        'signature': signature,
        'container': container,
        'prefix': prefix if prefix is not None else '',
        'prefixes': prefixes,
    })

    return render_to_response('upload_form.html', context,
                              context_instance=RequestContext(request))

    def test_set_user_agent_per_request_override(self):
        _junk, conn = c.http_connection('http://www.example.com')
        req_headers = {}

        def my_request_handler(*a, **kw):
            req_headers.update(kw.get('headers', {}))
        conn._request = my_request_handler

        # test if it's actually set
        conn.request('GET', '/', headers={'User-Agent': 'Me'})
        ua = req_headers.get('user-agent', 'XXX-MISSING-XXX')
        self.assertEqual(ua, b'Me', req_headers)